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 let (typed_args, output_type) = match function {
3065 TableFunctionKind::Unnest => {
3066 if args.len() != 1 {
3067 return Err(PlannerError::invalid_expression(format!(
3068 "table function UNNEST takes exactly 1 argument, found {}",
3069 args.len()
3070 )));
3071 }
3072 let typed = self.infer_expr_with_scope(&args[0], lateral_scope, ctes)?;
3073 if !matches!(
3074 typed.resolved_type,
3075 ResolvedType::Vector { .. } | ResolvedType::Null
3076 ) {
3077 return Err(PlannerError::type_mismatch(
3078 "VECTOR",
3079 typed.resolved_type.to_string(),
3080 args[0].span,
3081 ));
3082 }
3083 (vec![typed], ResolvedType::Float)
3084 }
3085 TableFunctionKind::GenerateSeries => {
3086 if !(2..=3).contains(&args.len()) {
3087 return Err(PlannerError::invalid_expression(format!(
3088 "table function GENERATE_SERIES takes 2 or 3 arguments, found {}",
3089 args.len()
3090 )));
3091 }
3092 let mut typed = Vec::with_capacity(args.len());
3093 let mut output_type = ResolvedType::Integer;
3094 for arg in args {
3095 let value = self.infer_expr_with_scope(arg, lateral_scope, ctes)?;
3096 match value.resolved_type {
3097 ResolvedType::BigInt => output_type = ResolvedType::BigInt,
3098 ResolvedType::Integer | ResolvedType::Null => {}
3099 _ => {
3100 return Err(PlannerError::type_mismatch(
3101 "INTEGER or BIGINT",
3102 value.resolved_type.to_string(),
3103 arg.span,
3104 ));
3105 }
3106 }
3107 typed.push(value);
3108 }
3109 (typed, output_type)
3110 }
3111 };
3112
3113 let mut schema = vec![ColumnMetadata::new(
3114 function.default_relation_name(),
3115 output_type,
3116 )];
3117 let relation_name = alias
3118 .map(str::to_string)
3119 .unwrap_or_else(|| function.default_relation_name().to_string());
3120 apply_alias_columns(&relation_name, columns, &mut schema, span)?;
3121
3122 Ok(PlannedRelation {
3123 plan: LogicalPlan::TableFunction {
3124 function,
3125 args: typed_args,
3126 schema: schema.clone(),
3127 },
3128 schema: schema.clone(),
3129 scope: vec![ScopedTable::new(
3130 TableMetadata::new(relation_name, schema),
3131 start_index,
3132 )],
3133 })
3134 }
3135
3136 fn combine_lateral_join_relation(
3137 &self,
3138 left: PlannedRelation,
3139 right: PlannedRelation,
3140 join_type: JoinType,
3141 condition: Option<TypedExpr>,
3142 using: Option<&[String]>,
3143 _span: crate::ast::Span,
3144 ) -> Result<PlannedRelation, PlannerError> {
3145 let (schema, scope) = combine_join_shape(&left, &right, using);
3150 Ok(PlannedRelation {
3151 plan: LogicalPlan::LateralJoin {
3152 left: Box::new(left.plan),
3153 right: Box::new(right.plan),
3154 join_type,
3155 condition,
3156 right_schema: right.schema,
3157 },
3158 schema,
3159 scope,
3160 })
3161 }
3162
3163 fn combine_join_relation(
3164 &self,
3165 left: PlannedRelation,
3166 right: PlannedRelation,
3167 join_type: JoinType,
3168 condition: Option<TypedExpr>,
3169 using: Option<Vec<String>>,
3170 _span: crate::ast::Span,
3171 ) -> Result<PlannedRelation, PlannerError> {
3172 let (schema, scope) = combine_join_shape(&left, &right, using.as_deref());
3173 Ok(PlannedRelation {
3174 plan: LogicalPlan::Join {
3175 left: Box::new(left.plan),
3176 right: Box::new(right.plan),
3177 join_type,
3178 condition,
3179 using,
3180 },
3181 schema,
3182 scope,
3183 })
3184 }
3185
3186 fn build_using_condition(
3187 &self,
3188 using: Option<&[String]>,
3189 left: &PlannedRelation,
3190 right: &PlannedRelation,
3191 span: crate::ast::Span,
3192 ) -> Result<Option<TypedExpr>, PlannerError> {
3193 let Some(columns) = using else {
3194 return Ok(None);
3195 };
3196 let mut condition = None;
3197 for column in columns {
3198 let left_col = find_scoped_column(&left.scope, column, span)?;
3199 let right_col = find_scoped_column(&right.scope, column, span)?;
3200 let left_expr = merged_scoped_column_expr(&left_col, column, span);
3201 let right_expr = merged_scoped_column_expr(&right_col, column, span);
3202 self.type_checker
3203 .check_comparison_op(&left_col.ty, &right_col.ty, span)?;
3204 let eq = TypedExpr::binary_op(
3205 left_expr,
3206 crate::ast::expr::BinaryOp::Eq,
3207 right_expr,
3208 ResolvedType::Boolean,
3209 span,
3210 );
3211 condition = Some(match condition {
3212 Some(prev) => TypedExpr::binary_op(
3213 prev,
3214 crate::ast::expr::BinaryOp::And,
3215 eq,
3216 ResolvedType::Boolean,
3217 span,
3218 ),
3219 None => eq,
3220 });
3221 }
3222 Ok(condition)
3223 }
3224
3225 fn infer_expr_with_scope(
3226 &self,
3227 expr: &crate::ast::expr::Expr,
3228 scope: &[ScopedTable],
3229 ctes: &CtePlans,
3230 ) -> Result<TypedExpr, PlannerError> {
3231 self.type_checker
3232 .infer_type_with_scope(expr, scope, &|stmt, outer_scope| {
3233 let relation = match &stmt.kind {
3234 StatementKind::Select(select) => {
3235 self.plan_select_relation(select, outer_scope, ctes)?
3236 }
3237 StatementKind::Values(values) => {
3238 self.plan_values_relation(values, outer_scope, ctes)?
3239 }
3240 _ => {
3241 return Err(PlannerError::unsupported_feature(
3242 "non-query subquery",
3243 "a future version",
3244 stmt.span(),
3245 ));
3246 }
3247 };
3248 Ok((relation.plan, relation.schema))
3249 })
3250 }
3251
3252 #[allow(dead_code)]
3253 fn build_projection(
3254 &self,
3255 items: &[SelectItem],
3256 table: &TableMetadata,
3257 ) -> Result<Projection, PlannerError> {
3258 if items.len() == 1 && matches!(&items[0], SelectItem::Wildcard { .. }) {
3260 let columns = self.name_resolver.expand_wildcard(table);
3261 return Ok(Projection::All(columns));
3262 }
3263
3264 let mut projected_columns = Vec::new();
3266 for item in items {
3267 match item {
3268 SelectItem::Wildcard { span } => {
3269 for col in &table.columns {
3271 let column_index = table.get_column_index(&col.name).unwrap();
3272 let typed_expr = TypedExpr::column_ref(
3273 table.name.clone(),
3274 col.name.clone(),
3275 column_index,
3276 col.data_type.clone(),
3277 *span,
3278 );
3279 projected_columns.push(ProjectedColumn::new(typed_expr));
3280 }
3281 }
3282 SelectItem::QualifiedWildcard {
3283 table: qualifier,
3284 span,
3285 } => {
3286 if qualifier != &table.name {
3287 return Err(PlannerError::invalid_expression(format!(
3288 "table '{qualifier}' is not available for wildcard projection"
3289 )));
3290 }
3291 for col in &table.columns {
3292 let column_index = table.get_column_index(&col.name).unwrap();
3293 projected_columns.push(ProjectedColumn::new(TypedExpr::column_ref(
3294 table.name.clone(),
3295 col.name.clone(),
3296 column_index,
3297 col.data_type.clone(),
3298 *span,
3299 )));
3300 }
3301 }
3302 SelectItem::Expr { expr, alias, .. } => {
3303 let typed_expr = self.type_checker.infer_type(expr, table)?;
3304 let projected = if let Some(alias) = alias {
3305 ProjectedColumn::with_alias(typed_expr, alias.clone())
3306 } else {
3307 ProjectedColumn::new(typed_expr)
3308 };
3309 projected_columns.push(projected);
3310 }
3311 }
3312 }
3313
3314 Ok(Projection::Columns(projected_columns))
3315 }
3316
3317 fn build_projection_with_scope(
3318 &self,
3319 items: &[SelectItem],
3320 schema: &[ColumnMetadata],
3321 scope: &[ScopedTable],
3322 ctes: &CtePlans,
3323 ) -> Result<Projection, PlannerError> {
3324 if items.len() == 1 && matches!(&items[0], SelectItem::Wildcard { .. }) {
3325 return Ok(Projection::All(visible_wildcard_columns(schema, scope)));
3326 }
3327
3328 let mut projected_columns = Vec::new();
3329 for item in items {
3330 match item {
3331 SelectItem::Wildcard { span } => {
3332 for scoped in scope {
3333 for (local_idx, col) in scoped.table.columns.iter().enumerate() {
3334 projected_columns.push(ProjectedColumn::new(TypedExpr::column_ref(
3335 scoped.table.name.clone(),
3336 col.name.clone(),
3337 scoped.start_index + local_idx,
3338 col.data_type.clone(),
3339 *span,
3340 )));
3341 }
3342 }
3343 }
3344 SelectItem::QualifiedWildcard { table, span } => {
3345 let scoped = scope
3346 .iter()
3347 .filter(|scoped| scoped.table.name == *table)
3348 .collect::<Vec<_>>();
3349 match scoped.as_slice() {
3350 [] => {
3351 return Err(PlannerError::invalid_expression(format!(
3352 "table '{table}' is not available for wildcard projection"
3353 )));
3354 }
3355 [scoped] => {
3356 for (local_idx, col) in scoped.table.columns.iter().enumerate() {
3357 projected_columns.push(ProjectedColumn::new(
3358 TypedExpr::column_ref(
3359 scoped.table.name.clone(),
3360 col.name.clone(),
3361 scoped.start_index + local_idx,
3362 col.data_type.clone(),
3363 *span,
3364 ),
3365 ));
3366 }
3367 }
3368 _ => {
3369 return Err(PlannerError::ambiguous_column(
3370 table,
3371 scoped
3372 .iter()
3373 .map(|scoped| scoped.table.name.clone())
3374 .collect(),
3375 *span,
3376 ));
3377 }
3378 }
3379 }
3380 SelectItem::Expr { expr, alias, .. } => {
3381 let typed_expr = self.infer_expr_with_scope(expr, scope, ctes)?;
3382 let projected = if let Some(alias) = alias {
3383 ProjectedColumn::with_alias(typed_expr, alias.clone())
3384 } else {
3385 ProjectedColumn::new(typed_expr)
3386 };
3387 projected_columns.push(projected);
3388 }
3389 }
3390 }
3391
3392 Ok(Projection::Columns(projected_columns))
3393 }
3394
3395 #[allow(dead_code)]
3397 fn build_sort_exprs(
3398 &self,
3399 order_by: &[OrderByExpr],
3400 table: &TableMetadata,
3401 ) -> Result<Vec<SortExpr>, PlannerError> {
3402 let mut sort_exprs = Vec::new();
3403
3404 for order_expr in order_by {
3405 let typed_expr = self.type_checker.infer_type(&order_expr.expr, table)?;
3406
3407 let asc = order_expr.asc.unwrap_or(true);
3409
3410 let nulls_first = order_expr.nulls_first.unwrap_or(false);
3412
3413 sort_exprs.push(SortExpr::new(typed_expr, asc, nulls_first));
3414 }
3415
3416 Ok(sort_exprs)
3417 }
3418
3419 fn build_sort_exprs_with_scope(
3420 &self,
3421 order_by: &[OrderByExpr],
3422 scope: &[ScopedTable],
3423 projection_aliases: &HashMap<String, crate::ast::expr::Expr>,
3424 ctes: &CtePlans,
3425 ) -> Result<Vec<SortExpr>, PlannerError> {
3426 let mut sort_exprs = Vec::new();
3427 for order_expr in order_by {
3428 let sort_source = substitute_projection_aliases(&order_expr.expr, projection_aliases);
3429 let typed_expr = self.infer_expr_with_scope(&sort_source, scope, ctes)?;
3430 let asc = order_expr.asc.unwrap_or(true);
3431 let nulls_first = order_expr.nulls_first.unwrap_or(false);
3432 sort_exprs.push(SortExpr::new(typed_expr, asc, nulls_first));
3433 }
3434 Ok(sort_exprs)
3435 }
3436
3437 fn select_contains_aggregate(&self, stmt: &Select) -> bool {
3438 stmt.projection.iter().any(|item| match item {
3439 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => false,
3440 SelectItem::Expr { expr, .. } => expr_contains_aggregate(expr),
3441 }) || stmt
3442 .group_by
3443 .as_ref()
3444 .map(|items| {
3445 items
3446 .iter()
3447 .flat_map(GroupByItem::exprs)
3448 .any(expr_contains_aggregate)
3449 })
3450 .unwrap_or(false)
3451 || stmt
3452 .having
3453 .as_ref()
3454 .map(expr_contains_aggregate)
3455 .unwrap_or(false)
3456 || stmt
3457 .qualify
3458 .as_ref()
3459 .map(expr_contains_aggregate)
3460 .unwrap_or(false)
3461 || stmt
3462 .order_by
3463 .iter()
3464 .any(|order| expr_contains_aggregate(&order.expr))
3465 }
3466
3467 #[allow(dead_code)]
3468 fn build_group_keys(
3469 &self,
3470 stmt: &Select,
3471 table: &TableMetadata,
3472 ) -> Result<Vec<TypedExpr>, PlannerError> {
3473 let mut keys = Vec::new();
3474 if let Some(items) = &stmt.group_by {
3475 for expr in items.iter().flat_map(GroupByItem::exprs) {
3476 let typed = self.type_checker.infer_type(expr, table)?;
3477 if typed_expr_contains_aggregate(&typed) {
3478 return Err(PlannerError::invalid_expression(
3479 "GROUP BY cannot contain aggregate functions".to_string(),
3480 ));
3481 }
3482 if !matches!(typed.kind, TypedExprKind::ColumnRef { .. }) {
3483 return Err(PlannerError::invalid_expression(
3484 "GROUP BY expressions must be column references".to_string(),
3485 ));
3486 }
3487 keys.push(typed);
3488 }
3489 }
3490 Ok(keys)
3491 }
3492
3493 fn type_group_key_with_scope(
3495 &self,
3496 expr: &Expr,
3497 scope: &[ScopedTable],
3498 ctes: &CtePlans,
3499 ) -> Result<TypedExpr, PlannerError> {
3500 let typed = self.infer_expr_with_scope(expr, scope, ctes)?;
3501 if typed_expr_contains_aggregate(&typed) {
3502 return Err(PlannerError::invalid_expression(
3503 "GROUP BY cannot contain aggregate functions".to_string(),
3504 ));
3505 }
3506 if !matches!(typed.kind, TypedExprKind::ColumnRef { .. }) {
3507 return Err(PlannerError::invalid_expression(
3508 "GROUP BY expressions must be column references".to_string(),
3509 ));
3510 }
3511 Ok(typed)
3512 }
3513
3514 fn expand_group_by_items(
3522 &self,
3523 stmt: &Select,
3524 scope: &[ScopedTable],
3525 ctes: &CtePlans,
3526 ) -> Result<ExpandedGroupBy, PlannerError> {
3527 let Some(items) = &stmt.group_by else {
3528 return Ok(ExpandedGroupBy {
3529 group_keys: Vec::new(),
3530 grouping_sets: None,
3531 });
3532 };
3533
3534 if items
3535 .iter()
3536 .all(|item| matches!(item, GroupByItem::Expr { .. }))
3537 {
3538 let mut keys = Vec::new();
3540 for item in items {
3541 if let GroupByItem::Expr { expr } = item {
3542 keys.push(self.type_group_key_with_scope(expr, scope, ctes)?);
3543 }
3544 }
3545 return Ok(ExpandedGroupBy {
3546 group_keys: keys,
3547 grouping_sets: None,
3548 });
3549 }
3550
3551 let mut keys: Vec<TypedExpr> = Vec::new();
3552 let mut key_index: HashMap<String, usize> = HashMap::new();
3553 let mut add_key = |planner: &Self, expr: &Expr| -> Result<usize, PlannerError> {
3554 let typed = planner.type_group_key_with_scope(expr, scope, ctes)?;
3555 let signature = expr_key(&typed);
3556 if let Some(&index) = key_index.get(&signature) {
3557 return Ok(index);
3558 }
3559 let index = keys.len();
3560 keys.push(typed);
3561 key_index.insert(signature, index);
3562 Ok(index)
3563 };
3564
3565 let mut sets: Vec<Vec<usize>> = vec![Vec::new()];
3568 for item in items {
3569 let item_sets: Vec<Vec<usize>> = match item {
3570 GroupByItem::Expr { expr } => vec![vec![add_key(self, expr)?]],
3571 GroupByItem::Rollup { exprs } => {
3572 if exprs.is_empty() {
3573 return Err(PlannerError::invalid_expression(
3574 "ROLLUP requires at least one expression".to_string(),
3575 ));
3576 }
3577 let indexes = exprs
3578 .iter()
3579 .map(|expr| add_key(self, expr))
3580 .collect::<Result<Vec<_>, _>>()?;
3581 (0..=indexes.len())
3582 .rev()
3583 .map(|len| indexes[..len].to_vec())
3584 .collect()
3585 }
3586 GroupByItem::Cube { exprs } => {
3587 if exprs.is_empty() {
3588 return Err(PlannerError::invalid_expression(
3589 "CUBE requires at least one expression".to_string(),
3590 ));
3591 }
3592 if exprs.len() > MAX_CUBE_COLUMNS {
3593 return Err(PlannerError::invalid_expression(format!(
3594 "too many grouping sets (max {MAX_GROUPING_SETS})"
3595 )));
3596 }
3597 let indexes = exprs
3598 .iter()
3599 .map(|expr| add_key(self, expr))
3600 .collect::<Result<Vec<_>, _>>()?;
3601 let n = indexes.len();
3602 (0..(1usize << n))
3603 .rev()
3604 .map(|included| {
3605 indexes
3606 .iter()
3607 .enumerate()
3608 .filter(|(position, _)| (included >> (n - 1 - position)) & 1 == 1)
3609 .map(|(_, &index)| index)
3610 .collect()
3611 })
3612 .collect()
3613 }
3614 GroupByItem::GroupingSets { sets: listed } => {
3615 if listed.is_empty() {
3616 return Err(PlannerError::invalid_expression(
3617 "GROUPING SETS requires at least one grouping set".to_string(),
3618 ));
3619 }
3620 listed
3621 .iter()
3622 .map(|set| {
3623 set.iter()
3624 .map(|expr| add_key(self, expr))
3625 .collect::<Result<Vec<_>, _>>()
3626 })
3627 .collect::<Result<Vec<_>, _>>()?
3628 }
3629 };
3630
3631 let mut combined = Vec::with_capacity(sets.len().saturating_mul(item_sets.len()));
3632 for base in &sets {
3633 for item_set in &item_sets {
3634 if combined.len() >= MAX_GROUPING_SETS {
3635 return Err(PlannerError::invalid_expression(format!(
3636 "too many grouping sets (max {MAX_GROUPING_SETS})"
3637 )));
3638 }
3639 let mut set = base.clone();
3640 set.extend(item_set.iter().copied());
3641 combined.push(set);
3642 }
3643 }
3644 sets = combined;
3645 }
3646
3647 if keys.len() > MAX_GROUPING_KEYS {
3648 return Err(PlannerError::invalid_expression(format!(
3649 "too many grouping columns (max {MAX_GROUPING_KEYS})"
3650 )));
3651 }
3652
3653 let key_count = keys.len();
3654 let full_mask = grouping_full_mask(key_count);
3655 let masks = sets
3656 .iter()
3657 .map(|set| {
3658 let mut mask = full_mask;
3659 for &index in set {
3660 mask &= !(1u64 << (key_count - 1 - index));
3661 }
3662 mask
3663 })
3664 .collect();
3665
3666 Ok(ExpandedGroupBy {
3667 group_keys: keys,
3668 grouping_sets: Some(masks),
3669 })
3670 }
3671
3672 #[allow(dead_code)]
3673 fn build_projected_columns_for_aggregate(
3674 &self,
3675 items: &[SelectItem],
3676 table: &TableMetadata,
3677 ) -> Result<Vec<ProjectedColumn>, PlannerError> {
3678 let mut projected = Vec::new();
3679 for item in items {
3680 match item {
3681 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => {
3682 return Err(PlannerError::invalid_expression(
3683 "wildcard projection not supported with GROUP BY/aggregate".to_string(),
3684 ));
3685 }
3686 SelectItem::Expr { expr, alias, .. } => {
3687 let typed = self.type_checker.infer_type(expr, table)?;
3688 projected.push(ProjectedColumn {
3689 expr: typed,
3690 alias: alias.clone(),
3691 });
3692 }
3693 }
3694 }
3695 Ok(projected)
3696 }
3697
3698 fn build_projected_columns_for_aggregate_with_scope(
3699 &self,
3700 items: &[SelectItem],
3701 scope: &[ScopedTable],
3702 ctes: &CtePlans,
3703 ) -> Result<Vec<ProjectedColumn>, PlannerError> {
3704 let mut projected = Vec::new();
3705 for item in items {
3706 match item {
3707 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => {
3708 return Err(PlannerError::invalid_expression(
3709 "wildcard projection not supported with GROUP BY/aggregate".to_string(),
3710 ));
3711 }
3712 SelectItem::Expr { expr, alias, .. } => {
3713 let typed = self.infer_expr_with_scope(expr, scope, ctes)?;
3714 projected.push(ProjectedColumn {
3715 expr: typed,
3716 alias: alias.clone(),
3717 });
3718 }
3719 }
3720 }
3721 Ok(projected)
3722 }
3723
3724 #[allow(dead_code)]
3725 fn build_projected_columns_for_distinct(
3726 &self,
3727 items: &[SelectItem],
3728 table: &TableMetadata,
3729 ) -> Result<Vec<ProjectedColumn>, PlannerError> {
3730 let projection = self.build_projection(items, table)?;
3731 match projection {
3732 Projection::All(columns) => {
3733 let mut projected = Vec::with_capacity(columns.len());
3734 for column in columns {
3735 let column_index = table.get_column_index(&column).ok_or_else(|| {
3736 PlannerError::invalid_expression(format!(
3737 "column '{column}' not found for DISTINCT projection"
3738 ))
3739 })?;
3740 let column_meta = table.get_column(&column).ok_or_else(|| {
3741 PlannerError::invalid_expression(format!(
3742 "column '{column}' not found for DISTINCT projection"
3743 ))
3744 })?;
3745 let typed_expr = TypedExpr::column_ref(
3746 table.name.clone(),
3747 column.clone(),
3748 column_index,
3749 column_meta.data_type.clone(),
3750 crate::ast::Span::default(),
3751 );
3752 projected.push(ProjectedColumn::new(typed_expr));
3753 }
3754 Ok(projected)
3755 }
3756 Projection::Columns(columns) => Ok(columns),
3757 }
3758 }
3759
3760 fn build_projected_columns_for_distinct_with_scope(
3761 &self,
3762 items: &[SelectItem],
3763 schema: &[ColumnMetadata],
3764 scope: &[ScopedTable],
3765 ctes: &CtePlans,
3766 ) -> Result<Vec<ProjectedColumn>, PlannerError> {
3767 let projection = self.build_projection_with_scope(items, schema, scope, ctes)?;
3768 match projection {
3769 Projection::All(columns) => {
3770 let mut projected = Vec::with_capacity(columns.len());
3771 for (idx, column) in columns.into_iter().enumerate() {
3772 let column_meta = schema.get(idx).ok_or_else(|| {
3773 PlannerError::invalid_expression(format!(
3774 "column '{column}' not found for DISTINCT projection"
3775 ))
3776 })?;
3777 projected.push(ProjectedColumn::new(TypedExpr::column_ref(
3778 LITERAL_TABLE.to_string(),
3779 column,
3780 idx,
3781 column_meta.data_type.clone(),
3782 crate::ast::Span::default(),
3783 )));
3784 }
3785 Ok(projected)
3786 }
3787 Projection::Columns(columns) => Ok(columns),
3788 }
3789 }
3790
3791 fn collect_aggregates_from_typed_expr(
3792 &self,
3793 expr: &TypedExpr,
3794 aggregates: &mut Vec<AggregateExpr>,
3795 aggregate_map: &mut HashMap<AggregateSignature, usize>,
3796 ) -> Result<(), PlannerError> {
3797 match &expr.kind {
3798 TypedExprKind::FunctionCall {
3799 name,
3800 args,
3801 distinct,
3802 star,
3803 filter,
3804 order_by,
3805 over: None,
3806 } if is_aggregate_function(name) => {
3807 if args.iter().any(typed_expr_contains_window) {
3808 return Err(PlannerError::invalid_expression(
3809 "aggregate functions cannot contain window functions".to_string(),
3810 ));
3811 }
3812 for arg in args {
3813 if typed_expr_contains_aggregate(arg) {
3814 return Err(PlannerError::invalid_expression(
3815 "nested aggregate functions are not supported".to_string(),
3816 ));
3817 }
3818 }
3819 if let Some(filter) = filter {
3823 if typed_expr_contains_aggregate(filter) {
3824 return Err(PlannerError::invalid_expression(
3825 "aggregate functions are not allowed in FILTER".to_string(),
3826 ));
3827 }
3828 if typed_expr_contains_window(filter) {
3829 return Err(PlannerError::invalid_expression(
3830 "window functions are not allowed in FILTER".to_string(),
3831 ));
3832 }
3833 }
3834 for sort in order_by {
3835 if typed_expr_contains_aggregate(&sort.expr) {
3836 return Err(PlannerError::invalid_expression(
3837 "aggregate functions are not allowed in aggregate ORDER BY".to_string(),
3838 ));
3839 }
3840 if typed_expr_contains_window(&sort.expr) {
3841 return Err(PlannerError::invalid_expression(
3842 "window functions are not allowed in aggregate ORDER BY".to_string(),
3843 ));
3844 }
3845 }
3846 let (agg, signature) = self.build_aggregate_expr_from_typed(
3847 expr,
3848 name,
3849 args,
3850 *distinct,
3851 *star,
3852 filter.as_deref(),
3853 order_by,
3854 )?;
3855 aggregate_map.entry(signature).or_insert_with(|| {
3856 aggregates.push(agg);
3857 aggregates.len() - 1
3858 });
3859 Ok(())
3860 }
3861 TypedExprKind::BinaryOp { left, right, .. } => {
3862 self.collect_aggregates_from_typed_expr(left, aggregates, aggregate_map)?;
3863 self.collect_aggregates_from_typed_expr(right, aggregates, aggregate_map)?;
3864 Ok(())
3865 }
3866 TypedExprKind::UnaryOp { operand, .. } => {
3867 self.collect_aggregates_from_typed_expr(operand, aggregates, aggregate_map)
3868 }
3869 TypedExprKind::Cast { expr, .. } | TypedExprKind::TryCast { expr, .. } => {
3870 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)
3871 }
3872 TypedExprKind::Case {
3873 operand,
3874 branches,
3875 else_expr,
3876 } => {
3877 if let Some(operand) = operand {
3878 self.collect_aggregates_from_typed_expr(operand, aggregates, aggregate_map)?;
3879 }
3880 for branch in branches {
3881 self.collect_aggregates_from_typed_expr(
3882 &branch.when,
3883 aggregates,
3884 aggregate_map,
3885 )?;
3886 self.collect_aggregates_from_typed_expr(
3887 &branch.then,
3888 aggregates,
3889 aggregate_map,
3890 )?;
3891 }
3892 if let Some(else_expr) = else_expr {
3893 self.collect_aggregates_from_typed_expr(else_expr, aggregates, aggregate_map)?;
3894 }
3895 Ok(())
3896 }
3897 TypedExprKind::FunctionCall { args, over, .. } => {
3898 for arg in args {
3899 self.collect_aggregates_from_typed_expr(arg, aggregates, aggregate_map)?;
3900 }
3901 if let Some(window) = over {
3902 for expr in &window.partition_by {
3903 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)?;
3904 }
3905 for sort in &window.order_by {
3906 self.collect_aggregates_from_typed_expr(
3907 &sort.expr,
3908 aggregates,
3909 aggregate_map,
3910 )?;
3911 }
3912 }
3913 Ok(())
3914 }
3915 TypedExprKind::Between {
3916 expr, low, high, ..
3917 } => {
3918 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)?;
3919 self.collect_aggregates_from_typed_expr(low, aggregates, aggregate_map)?;
3920 self.collect_aggregates_from_typed_expr(high, aggregates, aggregate_map)?;
3921 Ok(())
3922 }
3923 TypedExprKind::Like {
3924 expr,
3925 pattern,
3926 escape,
3927 ..
3928 } => {
3929 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)?;
3930 self.collect_aggregates_from_typed_expr(pattern, aggregates, aggregate_map)?;
3931 if let Some(esc) = escape {
3932 self.collect_aggregates_from_typed_expr(esc, aggregates, aggregate_map)?;
3933 }
3934 Ok(())
3935 }
3936 TypedExprKind::InList { expr, list, .. } => {
3937 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)?;
3938 for item in list {
3939 self.collect_aggregates_from_typed_expr(item, aggregates, aggregate_map)?;
3940 }
3941 Ok(())
3942 }
3943 TypedExprKind::IsNull { expr, .. } => {
3944 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)
3945 }
3946 _ => Ok(()),
3947 }
3948 }
3949
3950 fn collect_windows_from_typed_expr(
3951 &self,
3952 expr: &TypedExpr,
3953 windows: &mut Vec<WindowExpr>,
3954 window_map: &mut HashMap<String, usize>,
3955 ) -> Result<(), PlannerError> {
3956 match &expr.kind {
3957 TypedExprKind::FunctionCall {
3958 name,
3959 args,
3960 distinct,
3961 star,
3962 filter,
3963 order_by,
3964 over: Some(over),
3965 } => {
3966 if filter.is_some() || !order_by.is_empty() {
3967 return Err(PlannerError::invalid_expression(
3971 "FILTER and aggregate ORDER BY cannot be combined with OVER".to_string(),
3972 ));
3973 }
3974 if args.iter().any(typed_expr_contains_window)
3975 || over.partition_by.iter().any(typed_expr_contains_window)
3976 || over
3977 .order_by
3978 .iter()
3979 .any(|sort| typed_expr_contains_window(&sort.expr))
3980 {
3981 return Err(PlannerError::invalid_expression(
3982 "nested window functions are not supported".to_string(),
3983 ));
3984 }
3985
3986 let key = expr_key(expr);
3987 if window_map.contains_key(&key) {
3988 return Ok(());
3989 }
3990 let function = match name.to_ascii_lowercase().as_str() {
3991 "row_number" => WindowFunction::RowNumber,
3992 "rank" => WindowFunction::Rank,
3993 "dense_rank" => WindowFunction::DenseRank,
3994 "percent_rank" => WindowFunction::PercentRank,
3995 "cume_dist" => WindowFunction::CumeDist,
3996 "ntile" => WindowFunction::Ntile(args[0].clone()),
3997 "first_value" => {
3998 WindowFunction::Value(ValueWindowFunction::FirstValue(args[0].clone()))
3999 }
4000 "last_value" => {
4001 WindowFunction::Value(ValueWindowFunction::LastValue(args[0].clone()))
4002 }
4003 "nth_value" => WindowFunction::Value(ValueWindowFunction::NthValue {
4004 value: args[0].clone(),
4005 nth: args[1].clone(),
4006 }),
4007 name if is_aggregate_function(name) => {
4008 let (aggregate, _) = self.build_aggregate_expr_from_typed(
4009 expr, name, args, *distinct, *star, None, order_by,
4010 )?;
4011 WindowFunction::Aggregate(aggregate)
4012 }
4013 "lag" | "lead" => {
4014 let positional = build_offset_window_function(name, args)?;
4015 if name.eq_ignore_ascii_case("lag") {
4016 WindowFunction::Lag(positional)
4017 } else {
4018 WindowFunction::Lead(positional)
4019 }
4020 }
4021 _ => {
4022 return Err(PlannerError::unsupported_feature(
4023 format!("function '{}' with OVER", name),
4024 "future",
4025 expr.span,
4026 ));
4027 }
4028 };
4029 let index = windows.len();
4030 windows.push(WindowExpr {
4031 function,
4032 partition_by: over.partition_by.clone(),
4033 order_by: over.order_by.clone(),
4034 frame: over.frame.clone(),
4035 result_type: expr.resolved_type.clone(),
4036 });
4037 window_map.insert(key, index);
4038 Ok(())
4039 }
4040 TypedExprKind::FunctionCall { args, .. } => {
4041 for arg in args {
4042 self.collect_windows_from_typed_expr(arg, windows, window_map)?;
4043 }
4044 Ok(())
4045 }
4046 TypedExprKind::BinaryOp { left, right, .. } => {
4047 self.collect_windows_from_typed_expr(left, windows, window_map)?;
4048 self.collect_windows_from_typed_expr(right, windows, window_map)
4049 }
4050 TypedExprKind::UnaryOp { operand, .. } => {
4051 self.collect_windows_from_typed_expr(operand, windows, window_map)
4052 }
4053 TypedExprKind::Cast { expr, .. }
4054 | TypedExprKind::TryCast { expr, .. }
4055 | TypedExprKind::IsNull { expr, .. } => {
4056 self.collect_windows_from_typed_expr(expr, windows, window_map)
4057 }
4058 TypedExprKind::Between {
4059 expr, low, high, ..
4060 } => {
4061 self.collect_windows_from_typed_expr(expr, windows, window_map)?;
4062 self.collect_windows_from_typed_expr(low, windows, window_map)?;
4063 self.collect_windows_from_typed_expr(high, windows, window_map)
4064 }
4065 TypedExprKind::Like {
4066 expr,
4067 pattern,
4068 escape,
4069 ..
4070 } => {
4071 self.collect_windows_from_typed_expr(expr, windows, window_map)?;
4072 self.collect_windows_from_typed_expr(pattern, windows, window_map)?;
4073 if let Some(escape) = escape {
4074 self.collect_windows_from_typed_expr(escape, windows, window_map)?;
4075 }
4076 Ok(())
4077 }
4078 TypedExprKind::InList { expr, list, .. } => {
4079 self.collect_windows_from_typed_expr(expr, windows, window_map)?;
4080 for item in list {
4081 self.collect_windows_from_typed_expr(item, windows, window_map)?;
4082 }
4083 Ok(())
4084 }
4085 _ => Ok(()),
4086 }
4087 }
4088
4089 #[allow(clippy::too_many_arguments)]
4090 fn build_aggregate_expr_from_typed(
4091 &self,
4092 expr: &TypedExpr,
4093 name: &str,
4094 args: &[TypedExpr],
4095 distinct: bool,
4096 star: bool,
4097 filter: Option<&TypedExpr>,
4098 order_by: &[SortExpr],
4099 ) -> Result<(AggregateExpr, AggregateSignature), PlannerError> {
4100 let lower = name.to_lowercase();
4101 let filter_owned = filter.cloned();
4102 let retained_order_by: Vec<SortExpr> = if is_order_sensitive_aggregate(&lower) {
4107 order_by.to_vec()
4108 } else {
4109 Vec::new()
4110 };
4111 match lower.as_str() {
4112 "count" => {
4113 if star {
4114 let mut agg = AggregateExpr::count_star();
4115 agg.filter = filter_owned;
4116 let signature = aggregate_signature(
4117 name,
4118 distinct,
4119 star,
4120 None,
4121 None,
4122 expr,
4123 filter,
4124 &retained_order_by,
4125 );
4126 return Ok((agg, signature));
4127 }
4128 if args.len() != 1 {
4129 return Err(PlannerError::type_mismatch(
4130 "1 argument",
4131 format!("{} arguments", args.len()),
4132 expr.span,
4133 ));
4134 }
4135 let agg = AggregateExpr {
4136 function: AggregateFunction::Count,
4137 arg: Some(args[0].clone()),
4138 extra_args: Vec::new(),
4139 distinct,
4140 result_type: ResolvedType::BigInt,
4141 filter: filter_owned,
4142 order_by: retained_order_by.clone(),
4143 };
4144 let signature = aggregate_signature(
4145 name,
4146 distinct,
4147 star,
4148 Some(&args[0]),
4149 None,
4150 expr,
4151 filter,
4152 &retained_order_by,
4153 );
4154 Ok((agg, signature))
4155 }
4156 "sum" => {
4157 let arg = self.require_single_aggregate_arg(args, expr.span)?;
4158 let agg = AggregateExpr {
4159 function: AggregateFunction::Sum,
4160 arg: Some(arg.clone()),
4161 extra_args: Vec::new(),
4162 distinct,
4163 result_type: crate::planner::aggregate_expr::sum_result_type(
4164 &arg.resolved_type,
4165 ),
4166 filter: filter_owned,
4167 order_by: retained_order_by.clone(),
4168 };
4169 let signature = aggregate_signature(
4170 name,
4171 distinct,
4172 star,
4173 Some(arg),
4174 None,
4175 expr,
4176 filter,
4177 &retained_order_by,
4178 );
4179 Ok((agg, signature))
4180 }
4181 "total" => {
4182 let arg = self.require_single_aggregate_arg(args, expr.span)?;
4183 let agg = AggregateExpr {
4184 function: AggregateFunction::Total,
4185 arg: Some(arg.clone()),
4186 extra_args: Vec::new(),
4187 distinct: false,
4188 result_type: ResolvedType::Double,
4189 filter: filter_owned,
4190 order_by: retained_order_by.clone(),
4191 };
4192 let signature = aggregate_signature(
4193 name,
4194 false,
4195 star,
4196 Some(arg),
4197 None,
4198 expr,
4199 filter,
4200 &retained_order_by,
4201 );
4202 Ok((agg, signature))
4203 }
4204 "avg" => {
4205 let arg = self.require_single_aggregate_arg(args, expr.span)?;
4206 let agg = AggregateExpr {
4207 function: AggregateFunction::Avg,
4208 arg: Some(arg.clone()),
4209 extra_args: Vec::new(),
4210 distinct,
4211 result_type: ResolvedType::Double,
4212 filter: filter_owned,
4213 order_by: retained_order_by.clone(),
4214 };
4215 let signature = aggregate_signature(
4216 name,
4217 distinct,
4218 star,
4219 Some(arg),
4220 None,
4221 expr,
4222 filter,
4223 &retained_order_by,
4224 );
4225 Ok((agg, signature))
4226 }
4227 "min" => {
4228 let arg = self.require_single_aggregate_arg(args, expr.span)?;
4229 let agg = AggregateExpr {
4230 function: AggregateFunction::Min,
4231 arg: Some(arg.clone()),
4232 extra_args: Vec::new(),
4233 distinct,
4234 result_type: arg.resolved_type.clone(),
4235 filter: filter_owned,
4236 order_by: retained_order_by.clone(),
4237 };
4238 let signature = aggregate_signature(
4239 name,
4240 distinct,
4241 star,
4242 Some(arg),
4243 None,
4244 expr,
4245 filter,
4246 &retained_order_by,
4247 );
4248 Ok((agg, signature))
4249 }
4250 "max" => {
4251 let arg = self.require_single_aggregate_arg(args, expr.span)?;
4252 let agg = AggregateExpr {
4253 function: AggregateFunction::Max,
4254 arg: Some(arg.clone()),
4255 extra_args: Vec::new(),
4256 distinct,
4257 result_type: arg.resolved_type.clone(),
4258 filter: filter_owned,
4259 order_by: retained_order_by.clone(),
4260 };
4261 let signature = aggregate_signature(
4262 name,
4263 distinct,
4264 star,
4265 Some(arg),
4266 None,
4267 expr,
4268 filter,
4269 &retained_order_by,
4270 );
4271 Ok((agg, signature))
4272 }
4273 "group_concat" => {
4274 if args.is_empty() || args.len() > 2 {
4275 return Err(PlannerError::type_mismatch(
4276 "1 or 2 arguments",
4277 format!("{} arguments", args.len()),
4278 expr.span,
4279 ));
4280 }
4281 let arg = &args[0];
4282 let mut separator = None;
4283 if args.len() == 2 {
4284 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
4285 separator = Some(value.clone());
4286 } else {
4287 return Err(PlannerError::invalid_expression(
4288 "GROUP_CONCAT separator must be a string literal".to_string(),
4289 ));
4290 }
4291 }
4292 let agg = AggregateExpr {
4293 function: AggregateFunction::GroupConcat { separator },
4294 arg: Some(arg.clone()),
4295 extra_args: Vec::new(),
4296 distinct,
4297 result_type: ResolvedType::Text,
4298 filter: filter_owned,
4299 order_by: retained_order_by.clone(),
4300 };
4301 let signature = aggregate_signature(
4302 name,
4303 distinct,
4304 star,
4305 Some(arg),
4306 match &agg.function {
4307 AggregateFunction::GroupConcat { separator } => separator.as_ref(),
4308 _ => None,
4309 },
4310 expr,
4311 filter,
4312 &retained_order_by,
4313 );
4314 Ok((agg, signature))
4315 }
4316 "string_agg" => {
4317 if args.len() != 2 {
4318 return Err(PlannerError::type_mismatch(
4319 "2 arguments",
4320 format!("{} arguments", args.len()),
4321 expr.span,
4322 ));
4323 }
4324 let arg = &args[0];
4325 let separator =
4326 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
4327 Some(value.clone())
4328 } else {
4329 return Err(PlannerError::invalid_expression(
4330 "STRING_AGG separator must be a string literal".to_string(),
4331 ));
4332 };
4333 let agg = AggregateExpr {
4334 function: AggregateFunction::StringAgg { separator },
4335 arg: Some(arg.clone()),
4336 extra_args: Vec::new(),
4337 distinct,
4338 result_type: ResolvedType::Text,
4339 filter: filter_owned,
4340 order_by: retained_order_by.clone(),
4341 };
4342 let signature = aggregate_signature(
4343 name,
4344 distinct,
4345 star,
4346 Some(arg),
4347 match &agg.function {
4348 AggregateFunction::StringAgg { separator } => separator.as_ref(),
4349 _ => None,
4350 },
4351 expr,
4352 filter,
4353 &retained_order_by,
4354 );
4355 Ok((agg, signature))
4356 }
4357 "percentile_disc" => {
4358 if args.len() != 1 {
4359 return Err(PlannerError::type_mismatch(
4360 "1 argument",
4361 format!("{} arguments", args.len()),
4362 expr.span,
4363 ));
4364 }
4365 let fraction = type_checker::percentile_fraction(&args[0])?;
4366 if retained_order_by.len() != 1 {
4367 return Err(PlannerError::invalid_expression(
4368 "PERCENTILE_DISC requires WITHIN GROUP (ORDER BY ...) with exactly \
4369 one sort expression"
4370 .to_string(),
4371 ));
4372 }
4373 let sort = &retained_order_by[0];
4374 let agg = AggregateExpr {
4375 function: AggregateFunction::PercentileDisc { fraction },
4376 arg: Some(sort.expr.clone()),
4377 extra_args: Vec::new(),
4378 distinct: false,
4379 result_type: sort.expr.resolved_type.clone(),
4380 filter: filter_owned,
4381 order_by: retained_order_by.clone(),
4382 };
4383 let fraction_key = format!("{fraction:?}");
4386 let signature = aggregate_signature(
4387 name,
4388 false,
4389 star,
4390 None,
4391 Some(&fraction_key),
4392 expr,
4393 filter,
4394 &retained_order_by,
4395 );
4396 Ok((agg, signature))
4397 }
4398 "percentile_cont" => {
4399 let fraction = type_checker::percentile_fraction_named(name, &args[0])?;
4400 let sort = retained_order_by.first().ok_or_else(|| {
4401 PlannerError::invalid_expression(
4402 "PERCENTILE_CONT requires WITHIN GROUP (ORDER BY ...)".to_string(),
4403 )
4404 })?;
4405 let agg = AggregateExpr {
4406 function: AggregateFunction::PercentileCont { fraction },
4407 arg: Some(sort.expr.clone()),
4408 extra_args: Vec::new(),
4409 distinct: false,
4410 result_type: ResolvedType::Double,
4411 filter: filter_owned,
4412 order_by: retained_order_by.clone(),
4413 };
4414 let fraction_key = format!("{fraction:?}");
4415 let signature = aggregate_signature(
4416 name,
4417 false,
4418 star,
4419 None,
4420 Some(&fraction_key),
4421 expr,
4422 filter,
4423 &retained_order_by,
4424 );
4425 Ok((agg, signature))
4426 }
4427 "mode" if args.is_empty() => {
4428 let sort = retained_order_by.first().ok_or_else(|| {
4429 PlannerError::invalid_expression(
4430 "MODE requires WITHIN GROUP (ORDER BY ...)".to_string(),
4431 )
4432 })?;
4433 let agg = AggregateExpr {
4434 function: AggregateFunction::Mode,
4435 arg: Some(sort.expr.clone()),
4436 extra_args: Vec::new(),
4437 distinct: false,
4438 result_type: sort.expr.resolved_type.clone(),
4439 filter: filter_owned,
4440 order_by: retained_order_by.clone(),
4441 };
4442 let signature = aggregate_signature(
4443 name,
4444 false,
4445 star,
4446 None,
4447 None,
4448 expr,
4449 filter,
4450 &retained_order_by,
4451 );
4452 Ok((agg, signature))
4453 }
4454 name if type_checker::is_portable_aggregate_name(name) => {
4455 let function = match name {
4456 "variance" | "var_samp" => AggregateFunction::Variance { sample: true },
4457 "var_pop" => AggregateFunction::Variance { sample: false },
4458 "stddev" | "stddev_samp" => AggregateFunction::Stddev { sample: true },
4459 "stddev_pop" => AggregateFunction::Stddev { sample: false },
4460 "covar_samp" => AggregateFunction::Covariance { sample: true },
4461 "covar_pop" => AggregateFunction::Covariance { sample: false },
4462 "corr" => AggregateFunction::Corr,
4463 "median" => AggregateFunction::Median,
4464 "mode" => AggregateFunction::Mode,
4465 "quantile_cont" => AggregateFunction::QuantileCont {
4466 fraction: type_checker::percentile_fraction_named(name, &args[1])?,
4467 },
4468 "regr_count" => AggregateFunction::RegrCount,
4469 "regr_avgx" => AggregateFunction::RegrAvgX,
4470 "regr_avgy" => AggregateFunction::RegrAvgY,
4471 "regr_sxx" => AggregateFunction::RegrSxx,
4472 "regr_syy" => AggregateFunction::RegrSyy,
4473 "regr_sxy" => AggregateFunction::RegrSxy,
4474 "regr_slope" => AggregateFunction::RegrSlope,
4475 "regr_intercept" => AggregateFunction::RegrIntercept,
4476 "regr_r2" => AggregateFunction::RegrR2,
4477 "any_value" => AggregateFunction::AnyValue,
4478 "first" => AggregateFunction::First,
4479 "last" => AggregateFunction::Last,
4480 "arg_min" | "min_by" => AggregateFunction::ArgMin,
4481 "arg_max" | "max_by" => AggregateFunction::ArgMax,
4482 "bit_and" => AggregateFunction::BitAnd,
4483 "bit_or" => AggregateFunction::BitOr,
4484 "bit_xor" => AggregateFunction::BitXor,
4485 "bool_and" => AggregateFunction::BoolAnd,
4486 "bool_or" => AggregateFunction::BoolOr,
4487 _ => unreachable!(),
4488 };
4489 let primary = args[0].clone();
4490 let extra_args = if matches!(name, "quantile_cont") {
4491 Vec::new()
4492 } else {
4493 args[1..].to_vec()
4494 };
4495 let config = match &function {
4496 AggregateFunction::PercentileCont { fraction }
4497 | AggregateFunction::QuantileCont { fraction } => Some(format!("{fraction:?}")),
4498 _ => None,
4499 };
4500 let agg = AggregateExpr {
4501 function,
4502 arg: Some(primary.clone()),
4503 extra_args: extra_args.clone(),
4504 distinct: false,
4505 result_type: expr.resolved_type.clone(),
4506 filter: filter_owned,
4507 order_by: retained_order_by.clone(),
4508 };
4509 let mut signature = aggregate_signature(
4510 name,
4511 false,
4512 star,
4513 Some(&primary),
4514 config.as_ref(),
4515 expr,
4516 filter,
4517 &retained_order_by,
4518 );
4519 signature.extra_arg_keys = extra_args.iter().map(expr_key).collect();
4520 Ok((agg, signature))
4521 }
4522 _ => Err(PlannerError::unsupported_feature(
4523 format!("function '{}'", name),
4524 "future",
4525 expr.span,
4526 )),
4527 }
4528 }
4529 fn require_single_aggregate_arg<'b>(
4530 &self,
4531 args: &'b [TypedExpr],
4532 span: crate::ast::Span,
4533 ) -> Result<&'b TypedExpr, PlannerError> {
4534 if args.len() != 1 {
4535 return Err(PlannerError::type_mismatch(
4536 "1 argument",
4537 format!("{} arguments", args.len()),
4538 span,
4539 ));
4540 }
4541 Ok(&args[0])
4542 }
4543
4544 fn build_aggregate_projection(
4545 &self,
4546 projected: Vec<ProjectedColumn>,
4547 group_keys: &[TypedExpr],
4548 aggregates: &[AggregateExpr],
4549 output_names: &[String],
4550 grouping: Option<&GroupingRewrite>,
4551 ) -> Result<Projection, PlannerError> {
4552 let mut columns = Vec::new();
4553 for col in projected {
4554 let rewritten = self.rewrite_expr_for_aggregate(
4555 &col.expr,
4556 group_keys,
4557 aggregates,
4558 output_names,
4559 grouping,
4560 )?;
4561 columns.push(ProjectedColumn {
4562 expr: rewritten,
4563 alias: col.alias,
4564 });
4565 }
4566 Ok(Projection::Columns(columns))
4567 }
4568
4569 fn rewrite_expr_for_aggregate(
4577 &self,
4578 expr: &TypedExpr,
4579 group_keys: &[TypedExpr],
4580 aggregates: &[AggregateExpr],
4581 output_names: &[String],
4582 grouping: Option<&GroupingRewrite>,
4583 ) -> Result<TypedExpr, PlannerError> {
4584 let group_key_map = build_group_key_map(group_keys);
4585 let aggregate_map = build_aggregate_map(aggregates);
4586
4587 let expr = match grouping {
4588 Some(context) => rewrite_grouping_calls(expr, context)?,
4589 None => expr.clone(),
4590 };
4591 rewrite_expr_with_maps(&expr, &group_key_map, &aggregate_map, output_names)
4592 }
4593
4594 fn resolve_pagination_count(
4605 &self,
4606 expr: &Option<crate::ast::expr::Expr>,
4607 clause: &str,
4608 ) -> Result<Option<u64>, PlannerError> {
4609 let Some(expr) = expr else {
4610 return Ok(None);
4611 };
4612 if expr_contains_subquery(expr) {
4613 return Err(PlannerError::unsupported_feature(
4614 format!("subquery in {clause}"),
4615 "a future version",
4616 expr.span,
4617 ));
4618 }
4619 let typed = self
4621 .type_checker
4622 .infer_type_with_scope(expr, &[], &|stmt, _outer| {
4623 Err(PlannerError::unsupported_feature(
4624 format!("subquery in {clause}"),
4625 "a future version",
4626 stmt.span(),
4627 ))
4628 })?;
4629 if typed_expr_contains_aggregate(&typed) {
4630 return Err(PlannerError::invalid_expression(format!(
4631 "aggregate functions are not allowed in {clause}"
4632 )));
4633 }
4634 if typed_expr_contains_window(&typed) {
4635 return Err(PlannerError::invalid_expression(format!(
4636 "window functions are not allowed in {clause}"
4637 )));
4638 }
4639 match typed.resolved_type {
4640 ResolvedType::Integer | ResolvedType::BigInt | ResolvedType::Null => {}
4641 ref other => {
4642 return Err(PlannerError::type_mismatch(
4643 "BIGINT",
4644 other.type_name().to_string(),
4645 expr.span,
4646 ));
4647 }
4648 }
4649 let context = crate::executor::evaluator::EvalContext::new(&[]);
4652 let value = crate::executor::evaluator::evaluate(&typed, &context).map_err(|error| {
4653 PlannerError::invalid_expression(format!("{clause} expression is invalid: {error}"))
4654 })?;
4655 let count = match value {
4656 crate::storage::SqlValue::Null => return Ok(None),
4657 crate::storage::SqlValue::Integer(value) => i64::from(value),
4658 crate::storage::SqlValue::BigInt(value) => value,
4659 other => {
4660 return Err(PlannerError::type_mismatch(
4661 "BIGINT",
4662 format!("{other:?}"),
4663 expr.span,
4664 ));
4665 }
4666 };
4667 if count < 0 {
4668 return Err(PlannerError::invalid_expression(format!(
4669 "{clause} must not be negative"
4670 )));
4671 }
4672 Ok(Some(count as u64))
4673 }
4674
4675 fn apply_pagination(
4680 &self,
4681 plan: LogicalPlan,
4682 limit: &Option<Expr>,
4683 offset: &Option<Expr>,
4684 with_ties: bool,
4685 ) -> Result<LogicalPlan, PlannerError> {
4686 self.apply_pagination_with_tie_keys(plan, limit, offset, with_ties, None)
4687 }
4688
4689 fn apply_pagination_with_tie_keys(
4697 &self,
4698 plan: LogicalPlan,
4699 limit: &Option<Expr>,
4700 offset: &Option<Expr>,
4701 with_ties: bool,
4702 tie_keys: Option<&[SortExpr]>,
4703 ) -> Result<LogicalPlan, PlannerError> {
4704 if limit.is_none() && offset.is_none() && !with_ties {
4705 return Ok(plan);
4706 }
4707 let ties = if with_ties {
4708 let keys = match tie_keys {
4709 Some(keys) => keys.to_vec(),
4710 None => match &plan {
4711 LogicalPlan::Sort { order_by, .. } => order_by.clone(),
4712 _ => Vec::new(),
4713 },
4714 };
4715 if keys.is_empty() {
4716 return Err(PlannerError::invalid_expression(
4717 "FETCH ... WITH TIES requires ORDER BY".to_string(),
4718 ));
4719 }
4720 Some(keys)
4721 } else {
4722 None
4723 };
4724 Ok(LogicalPlan::Limit {
4725 input: Box::new(plan),
4726 limit: self.resolve_pagination_count(limit, "LIMIT")?,
4727 offset: self.resolve_pagination_count(offset, "OFFSET")?,
4728 ties,
4729 })
4730 }
4731
4732 fn plan_insert(&self, stmt: &Insert) -> Result<LogicalPlan, PlannerError> {
4737 let table = self.name_resolver.resolve_table(&stmt.table, stmt.span)?;
4739
4740 let columns: Vec<String> = if let Some(ref cols) = stmt.columns {
4742 for col in cols {
4744 self.name_resolver.resolve_column(table, col, stmt.span)?;
4745 }
4746 cols.clone()
4747 } else {
4748 table.column_names().into_iter().map(String::from).collect()
4750 };
4751
4752 match &stmt.source {
4753 InsertSource::Values { values } => {
4754 let mut typed_values: Vec<Vec<TypedExpr>> = Vec::new();
4755
4756 for row in values {
4757 if row.len() != columns.len() {
4758 return Err(PlannerError::column_value_count_mismatch(
4759 columns.len(),
4760 row.len(),
4761 stmt.span,
4762 ));
4763 }
4764
4765 typed_values.push(self.type_check_insert_values(row, &columns, table)?);
4766 }
4767
4768 Ok(LogicalPlan::Insert {
4769 table: table.name.clone(),
4770 columns,
4771 values: typed_values,
4772 })
4773 }
4774 InsertSource::Select { select } => {
4775 let source = self.plan_select_relation(select, &[], &CtePlans::new())?;
4776 self.finish_insert_query(stmt, table, columns, source)
4777 }
4778 InsertSource::Query { query } => {
4779 let source = self.plan_query_body_relation(query, &[], &CtePlans::new())?;
4780 self.finish_insert_query(stmt, table, columns, source)
4781 }
4782 }
4783 }
4784
4785 fn finish_insert_query(
4786 &self,
4787 stmt: &Insert,
4788 table: &TableMetadata,
4789 columns: Vec<String>,
4790 source: PlannedRelation,
4791 ) -> Result<LogicalPlan, PlannerError> {
4792 if source.schema.len() != columns.len() {
4793 return Err(PlannerError::column_value_count_mismatch(
4794 columns.len(),
4795 source.schema.len(),
4796 stmt.span,
4797 ));
4798 }
4799
4800 for (source_column, target_column) in source.schema.iter().zip(&columns) {
4801 let target = table
4802 .get_column(target_column)
4803 .expect("validated target column");
4804 if target.not_null && source_column.data_type == ResolvedType::Null {
4805 return Err(PlannerError::null_constraint_violation(
4806 target_column,
4807 stmt.span,
4808 ));
4809 }
4810 self.validate_resolved_type_assignment(
4811 &source_column.data_type,
4812 &target.data_type,
4813 stmt.span,
4814 )?;
4815 }
4816
4817 Ok(LogicalPlan::InsertSelect {
4818 table: table.name.clone(),
4819 columns,
4820 source: Box::new(source.plan),
4821 })
4822 }
4823
4824 fn type_check_insert_values(
4826 &self,
4827 values: &[crate::ast::expr::Expr],
4828 columns: &[String],
4829 table: &TableMetadata,
4830 ) -> Result<Vec<TypedExpr>, PlannerError> {
4831 let mut typed_values = Vec::new();
4832
4833 for (i, value) in values.iter().enumerate() {
4834 let column_name = &columns[i];
4835 let column_meta = table.get_column(column_name).ok_or_else(|| {
4836 PlannerError::column_not_found(column_name, &table.name, value.span)
4837 })?;
4838
4839 let typed_value = self.type_checker.infer_type(value, table)?;
4841
4842 if column_meta.not_null
4844 && matches!(&typed_value.kind, TypedExprKind::Literal(Literal::Null))
4845 {
4846 return Err(PlannerError::null_constraint_violation(
4847 column_name,
4848 value.span,
4849 ));
4850 }
4851
4852 self.validate_type_assignment(&typed_value, &column_meta.data_type, value.span)?;
4854
4855 let typed_value =
4856 self.coerce_assignment_value(typed_value, &column_meta.data_type, value.span);
4857
4858 typed_values.push(typed_value);
4859 }
4860
4861 Ok(typed_values)
4862 }
4863
4864 fn validate_type_assignment(
4866 &self,
4867 value: &TypedExpr,
4868 target_type: &ResolvedType,
4869 span: crate::ast::Span,
4870 ) -> Result<(), PlannerError> {
4871 self.validate_resolved_type_assignment(&value.resolved_type, target_type, span)
4872 }
4873
4874 fn validate_resolved_type_assignment(
4875 &self,
4876 source_type: &ResolvedType,
4877 target_type: &ResolvedType,
4878 span: crate::ast::Span,
4879 ) -> Result<(), PlannerError> {
4880 if *source_type == ResolvedType::Null {
4882 return Ok(());
4883 }
4884
4885 if self.types_compatible(source_type, target_type) {
4887 return Ok(());
4888 }
4889
4890 Err(PlannerError::type_mismatch(
4891 target_type.to_string(),
4892 source_type.to_string(),
4893 span,
4894 ))
4895 }
4896
4897 fn types_compatible(&self, source: &ResolvedType, target: &ResolvedType) -> bool {
4899 use ResolvedType::*;
4900
4901 if source == target {
4903 return true;
4904 }
4905
4906 match (source, target) {
4908 (Integer, BigInt) | (Integer, Float) | (Integer, Double) => true,
4910 (BigInt, Float) | (BigInt, Double) => true,
4912 (Float, Double) => true,
4914 (Double, Float) => true,
4917 (Text | Integer | BigInt | Float | Double, Timestamp) => true,
4920 (Vector { dimension: d1, .. }, Vector { dimension: d2, .. }) => d1 == d2,
4922 _ => false,
4923 }
4924 }
4925
4926 fn coerce_assignment_value(
4927 &self,
4928 value: TypedExpr,
4929 target_type: &ResolvedType,
4930 span: crate::ast::Span,
4931 ) -> TypedExpr {
4932 if value.resolved_type != *target_type
4933 && value.resolved_type != ResolvedType::Null
4934 && matches!(
4935 target_type,
4936 ResolvedType::Integer
4937 | ResolvedType::BigInt
4938 | ResolvedType::Float
4939 | ResolvedType::Double
4940 | ResolvedType::Timestamp
4941 )
4942 {
4943 TypedExpr::cast(value, target_type.clone(), span)
4944 } else {
4945 value
4946 }
4947 }
4948
4949 fn plan_update(&self, stmt: &Update) -> Result<LogicalPlan, PlannerError> {
4953 let table = self.name_resolver.resolve_table(&stmt.table, stmt.span)?;
4955
4956 let mut typed_assignments = Vec::new();
4958
4959 for assignment in &stmt.assignments {
4960 let column_meta =
4962 self.name_resolver
4963 .resolve_column(table, &assignment.column, assignment.span)?;
4964 let column_index = table.get_column_index(&assignment.column).unwrap();
4965
4966 let typed_value = self.type_checker.infer_type(&assignment.value, table)?;
4968
4969 if column_meta.not_null
4971 && matches!(&typed_value.kind, TypedExprKind::Literal(Literal::Null))
4972 {
4973 return Err(PlannerError::null_constraint_violation(
4974 &assignment.column,
4975 assignment.value.span,
4976 ));
4977 }
4978
4979 self.validate_type_assignment(
4981 &typed_value,
4982 &column_meta.data_type,
4983 assignment.value.span,
4984 )?;
4985
4986 let typed_value = self.coerce_assignment_value(
4987 typed_value,
4988 &column_meta.data_type,
4989 assignment.value.span,
4990 );
4991
4992 typed_assignments.push(TypedAssignment::new(
4993 assignment.column.clone(),
4994 column_index,
4995 typed_value,
4996 ));
4997 }
4998
4999 let filter = if let Some(ref selection) = stmt.selection {
5001 let predicate = self.type_checker.infer_type(selection, table)?;
5002
5003 if predicate.resolved_type != ResolvedType::Boolean {
5005 return Err(PlannerError::type_mismatch(
5006 "Boolean",
5007 predicate.resolved_type.to_string(),
5008 selection.span,
5009 ));
5010 }
5011
5012 Some(predicate)
5013 } else {
5014 None
5015 };
5016
5017 Ok(LogicalPlan::Update {
5018 table: table.name.clone(),
5019 assignments: typed_assignments,
5020 filter,
5021 })
5022 }
5023
5024 fn plan_delete(&self, stmt: &Delete) -> Result<LogicalPlan, PlannerError> {
5028 let table = self.name_resolver.resolve_table(&stmt.table, stmt.span)?;
5030
5031 let filter = if let Some(ref selection) = stmt.selection {
5033 let predicate = self.type_checker.infer_type(selection, table)?;
5034
5035 if predicate.resolved_type != ResolvedType::Boolean {
5037 return Err(PlannerError::type_mismatch(
5038 "Boolean",
5039 predicate.resolved_type.to_string(),
5040 selection.span,
5041 ));
5042 }
5043
5044 Some(predicate)
5045 } else {
5046 None
5047 };
5048
5049 Ok(LogicalPlan::Delete {
5050 table: table.name.clone(),
5051 filter,
5052 })
5053 }
5054}
5055
5056#[derive(Debug, Clone, PartialEq, Eq, Hash)]
5057struct AggregateSignature {
5058 name: String,
5059 distinct: bool,
5060 star: bool,
5061 arg_key: Option<String>,
5062 extra_arg_keys: Vec<String>,
5063 separator: Option<String>,
5064 filter_key: Option<String>,
5067 order_key: Option<String>,
5071}
5072
5073fn collect_projection_aliases(items: &[SelectItem]) -> HashMap<String, crate::ast::expr::Expr> {
5083 let mut aliases = HashMap::new();
5084 for item in items {
5085 if let SelectItem::Expr {
5086 expr,
5087 alias: Some(alias),
5088 ..
5089 } = item
5090 {
5091 aliases.entry(alias.clone()).or_insert_with(|| expr.clone());
5092 }
5093 }
5094 aliases
5095}
5096
5097fn substitute_projection_aliases(
5115 expr: &crate::ast::expr::Expr,
5116 aliases: &HashMap<String, crate::ast::expr::Expr>,
5117) -> crate::ast::expr::Expr {
5118 use crate::ast::expr::ExprKind;
5119
5120 if aliases.is_empty() {
5121 return expr.clone();
5122 }
5123
5124 let recurse = |e: &crate::ast::expr::Expr| substitute_projection_aliases(e, aliases);
5125
5126 let kind = match &expr.kind {
5127 ExprKind::ColumnRef {
5128 table: None,
5129 column,
5130 } => match aliases.get(column) {
5131 Some(source) => {
5132 let mut replacement = source.clone();
5133 replacement.span = expr.span;
5134 return replacement;
5135 }
5136 None => return expr.clone(),
5137 },
5138 ExprKind::BinaryOp { left, op, right } => ExprKind::BinaryOp {
5139 left: Box::new(recurse(left)),
5140 op: *op,
5141 right: Box::new(recurse(right)),
5142 },
5143 ExprKind::UnaryOp { op, operand } => ExprKind::UnaryOp {
5144 op: *op,
5145 operand: Box::new(recurse(operand)),
5146 },
5147 ExprKind::FunctionCall {
5148 name,
5149 args,
5150 distinct,
5151 star,
5152 order_by,
5153 within_group,
5154 filter,
5155 over,
5156 } => ExprKind::FunctionCall {
5157 name: name.clone(),
5158 args: args.iter().map(recurse).collect(),
5159 distinct: *distinct,
5160 star: *star,
5161 order_by: order_by
5162 .iter()
5163 .map(|order| OrderByExpr {
5164 expr: recurse(&order.expr),
5165 asc: order.asc,
5166 nulls_first: order.nulls_first,
5167 span: order.span,
5168 })
5169 .collect(),
5170 within_group: within_group
5171 .iter()
5172 .map(|order| OrderByExpr {
5173 expr: recurse(&order.expr),
5174 asc: order.asc,
5175 nulls_first: order.nulls_first,
5176 span: order.span,
5177 })
5178 .collect(),
5179 filter: filter
5180 .as_deref()
5181 .map(|predicate| Box::new(recurse(predicate))),
5182 over: over.as_ref().map(|window| crate::ast::expr::WindowSpec {
5183 base: window.base.clone(),
5184 partition_by: window.partition_by.iter().map(recurse).collect(),
5185 order_by: window
5186 .order_by
5187 .iter()
5188 .map(|order| OrderByExpr {
5189 expr: recurse(&order.expr),
5190 asc: order.asc,
5191 nulls_first: order.nulls_first,
5192 span: order.span,
5193 })
5194 .collect(),
5195 frame: window.frame.clone(),
5196 }),
5197 },
5198 ExprKind::Case {
5199 operand,
5200 branches,
5201 else_expr,
5202 } => ExprKind::Case {
5203 operand: operand.as_deref().map(|e| Box::new(recurse(e))),
5204 branches: branches
5205 .iter()
5206 .map(|branch| crate::ast::expr::CaseWhen {
5207 when: recurse(&branch.when),
5208 then: recurse(&branch.then),
5209 })
5210 .collect(),
5211 else_expr: else_expr.as_deref().map(|e| Box::new(recurse(e))),
5212 },
5213 ExprKind::Cast { expr, target_type } => ExprKind::Cast {
5214 expr: Box::new(recurse(expr)),
5215 target_type: target_type.clone(),
5216 },
5217 ExprKind::TryCast { expr, target_type } => ExprKind::TryCast {
5218 expr: Box::new(recurse(expr)),
5219 target_type: target_type.clone(),
5220 },
5221 ExprKind::Between {
5222 expr,
5223 low,
5224 high,
5225 negated,
5226 } => ExprKind::Between {
5227 expr: Box::new(recurse(expr)),
5228 low: Box::new(recurse(low)),
5229 high: Box::new(recurse(high)),
5230 negated: *negated,
5231 },
5232 ExprKind::Like {
5233 expr,
5234 pattern,
5235 escape,
5236 negated,
5237 kind,
5238 } => ExprKind::Like {
5239 expr: Box::new(recurse(expr)),
5240 pattern: Box::new(recurse(pattern)),
5241 escape: escape.as_deref().map(|e| Box::new(recurse(e))),
5242 negated: *negated,
5243 kind: *kind,
5244 },
5245 ExprKind::InList {
5246 expr,
5247 list,
5248 negated,
5249 } => ExprKind::InList {
5250 expr: Box::new(recurse(expr)),
5251 list: list.iter().map(recurse).collect(),
5252 negated: *negated,
5253 },
5254 ExprKind::IsNull { expr, negated } => ExprKind::IsNull {
5255 expr: Box::new(recurse(expr)),
5256 negated: *negated,
5257 },
5258 ExprKind::TruthPredicate {
5259 expr,
5260 value,
5261 negated,
5262 } => ExprKind::TruthPredicate {
5263 expr: Box::new(recurse(expr)),
5264 value: *value,
5265 negated: *negated,
5266 },
5267 ExprKind::IsDistinctFrom {
5268 left,
5269 right,
5270 negated,
5271 } => ExprKind::IsDistinctFrom {
5272 left: Box::new(recurse(left)),
5273 right: Box::new(recurse(right)),
5274 negated: *negated,
5275 },
5276 ExprKind::Row { items } => ExprKind::Row {
5277 items: items.iter().map(recurse).collect(),
5278 },
5279 ExprKind::ColumnRef { .. }
5282 | ExprKind::Literal { .. }
5283 | ExprKind::VectorLiteral { .. }
5284 | ExprKind::ScalarSubquery { .. }
5285 | ExprKind::InSubquery { .. }
5286 | ExprKind::Exists { .. }
5287 | ExprKind::Quantified { .. } => return expr.clone(),
5288 };
5289
5290 crate::ast::expr::Expr {
5291 kind,
5292 span: expr.span,
5293 }
5294}
5295
5296fn build_offset_window_function(
5297 name: &str,
5298 args: &[TypedExpr],
5299) -> Result<OffsetWindowFunction, PlannerError> {
5300 let value = args.first().cloned().ok_or_else(|| {
5301 PlannerError::invalid_expression(format!(
5302 "{}() window function expects 1 to 3 arguments",
5303 name.to_ascii_uppercase()
5304 ))
5305 })?;
5306 if args.len() > 3 {
5307 return Err(PlannerError::invalid_expression(format!(
5308 "{}() window function expects 1 to 3 arguments",
5309 name.to_ascii_uppercase()
5310 )));
5311 }
5312 Ok(OffsetWindowFunction {
5313 value,
5314 offset: args.get(1).cloned(),
5315 default: args.get(2).cloned(),
5316 })
5317}
5318
5319fn expr_contains_aggregate(expr: &crate::ast::expr::Expr) -> bool {
5320 use crate::ast::expr::ExprKind;
5321
5322 match &expr.kind {
5323 ExprKind::FunctionCall {
5324 name,
5325 args,
5326 order_by,
5327 within_group,
5328 filter,
5329 over,
5330 ..
5331 } => {
5332 if over.is_none() && is_aggregate_function(name) {
5333 return true;
5334 }
5335 args.iter().any(expr_contains_aggregate)
5336 || order_by
5337 .iter()
5338 .any(|sort| expr_contains_aggregate(&sort.expr))
5339 || within_group
5340 .iter()
5341 .any(|sort| expr_contains_aggregate(&sort.expr))
5342 || filter.as_deref().is_some_and(expr_contains_aggregate)
5343 || over.as_ref().is_some_and(|window| {
5344 window.partition_by.iter().any(expr_contains_aggregate)
5345 || window
5346 .order_by
5347 .iter()
5348 .any(|sort| expr_contains_aggregate(&sort.expr))
5349 })
5350 }
5351 ExprKind::BinaryOp { left, right, .. } => {
5352 expr_contains_aggregate(left) || expr_contains_aggregate(right)
5353 }
5354 ExprKind::UnaryOp { operand, .. } => expr_contains_aggregate(operand),
5355 ExprKind::TruthPredicate { expr, .. } => expr_contains_aggregate(expr),
5356 ExprKind::IsDistinctFrom { left, right, .. } => {
5357 expr_contains_aggregate(left) || expr_contains_aggregate(right)
5358 }
5359 ExprKind::Row { items } => items.iter().any(expr_contains_aggregate),
5360 ExprKind::Case {
5361 operand,
5362 branches,
5363 else_expr,
5364 } => {
5365 operand.as_deref().is_some_and(expr_contains_aggregate)
5366 || branches.iter().any(|branch| {
5367 expr_contains_aggregate(&branch.when) || expr_contains_aggregate(&branch.then)
5368 })
5369 || else_expr.as_deref().is_some_and(expr_contains_aggregate)
5370 }
5371 ExprKind::Cast { expr, .. } | ExprKind::TryCast { expr, .. } => {
5372 expr_contains_aggregate(expr)
5373 }
5374 ExprKind::Between {
5375 expr, low, high, ..
5376 } => {
5377 expr_contains_aggregate(expr)
5378 || expr_contains_aggregate(low)
5379 || expr_contains_aggregate(high)
5380 }
5381 ExprKind::Like {
5382 expr,
5383 pattern,
5384 escape,
5385 ..
5386 } => {
5387 expr_contains_aggregate(expr)
5388 || expr_contains_aggregate(pattern)
5389 || escape.as_deref().is_some_and(expr_contains_aggregate)
5390 }
5391 ExprKind::InList { expr, list, .. } => {
5392 expr_contains_aggregate(expr) || list.iter().any(expr_contains_aggregate)
5393 }
5394 ExprKind::IsNull { expr, .. } => expr_contains_aggregate(expr),
5395 ExprKind::ScalarSubquery { .. }
5396 | ExprKind::InSubquery { .. }
5397 | ExprKind::Exists { .. }
5398 | ExprKind::Quantified { .. }
5399 | ExprKind::Literal { .. }
5400 | ExprKind::VectorLiteral { .. }
5401 | ExprKind::ColumnRef { .. } => false,
5402 }
5403}
5404
5405fn typed_expr_contains_aggregate(expr: &TypedExpr) -> bool {
5406 match &expr.kind {
5407 TypedExprKind::FunctionCall {
5408 name,
5409 args,
5410 filter,
5411 order_by,
5412 over,
5413 ..
5414 } => {
5415 if over.is_none() && is_aggregate_function(name) {
5416 return true;
5417 }
5418 args.iter().any(typed_expr_contains_aggregate)
5419 || filter.as_deref().is_some_and(typed_expr_contains_aggregate)
5420 || order_by
5421 .iter()
5422 .any(|sort| typed_expr_contains_aggregate(&sort.expr))
5423 || over.as_ref().is_some_and(|window| {
5424 window
5425 .partition_by
5426 .iter()
5427 .any(typed_expr_contains_aggregate)
5428 || window
5429 .order_by
5430 .iter()
5431 .any(|sort| typed_expr_contains_aggregate(&sort.expr))
5432 })
5433 }
5434 TypedExprKind::BinaryOp { left, right, .. } => {
5435 typed_expr_contains_aggregate(left) || typed_expr_contains_aggregate(right)
5436 }
5437 TypedExprKind::UnaryOp { operand, .. } => typed_expr_contains_aggregate(operand),
5438 TypedExprKind::Cast { expr, .. } | TypedExprKind::TryCast { expr, .. } => {
5439 typed_expr_contains_aggregate(expr)
5440 }
5441 TypedExprKind::Case {
5442 operand,
5443 branches,
5444 else_expr,
5445 } => {
5446 operand
5447 .as_deref()
5448 .is_some_and(typed_expr_contains_aggregate)
5449 || branches.iter().any(|branch| {
5450 typed_expr_contains_aggregate(&branch.when)
5451 || typed_expr_contains_aggregate(&branch.then)
5452 })
5453 || else_expr
5454 .as_deref()
5455 .is_some_and(typed_expr_contains_aggregate)
5456 }
5457 TypedExprKind::Between {
5458 expr, low, high, ..
5459 } => {
5460 typed_expr_contains_aggregate(expr)
5461 || typed_expr_contains_aggregate(low)
5462 || typed_expr_contains_aggregate(high)
5463 }
5464 TypedExprKind::Like {
5465 expr,
5466 pattern,
5467 escape,
5468 ..
5469 } => {
5470 typed_expr_contains_aggregate(expr)
5471 || typed_expr_contains_aggregate(pattern)
5472 || escape
5473 .as_ref()
5474 .is_some_and(|inner| typed_expr_contains_aggregate(inner))
5475 }
5476 TypedExprKind::InList { expr, list, .. } => {
5477 typed_expr_contains_aggregate(expr) || list.iter().any(typed_expr_contains_aggregate)
5478 }
5479 TypedExprKind::IsNull { expr, .. } => typed_expr_contains_aggregate(expr),
5480 TypedExprKind::InSubquery { expr, .. } => typed_expr_contains_aggregate(expr),
5481 TypedExprKind::Quantified { expr, .. } => typed_expr_contains_aggregate(expr),
5482 TypedExprKind::ScalarSubquery(_) | TypedExprKind::Exists { .. } => false,
5483 _ => false,
5484 }
5485}
5486
5487fn select_contains_window(stmt: &Select) -> bool {
5488 stmt.projection.iter().any(|item| match item {
5489 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => false,
5490 SelectItem::Expr { expr, .. } => expr_contains_window(expr),
5491 }) || stmt.qualify.as_ref().is_some_and(expr_contains_window)
5492 || stmt
5493 .order_by
5494 .iter()
5495 .any(|order| expr_contains_window(&order.expr))
5496}
5497
5498fn expr_contains_window(expr: &crate::ast::expr::Expr) -> bool {
5499 match &expr.kind {
5500 crate::ast::expr::ExprKind::FunctionCall {
5501 args,
5502 order_by,
5503 within_group,
5504 filter,
5505 over,
5506 ..
5507 } => {
5508 over.is_some()
5509 || args.iter().any(expr_contains_window)
5510 || order_by.iter().any(|sort| expr_contains_window(&sort.expr))
5511 || within_group
5512 .iter()
5513 .any(|sort| expr_contains_window(&sort.expr))
5514 || filter.as_deref().is_some_and(expr_contains_window)
5515 }
5516 crate::ast::expr::ExprKind::BinaryOp { left, right, .. } => {
5517 expr_contains_window(left) || expr_contains_window(right)
5518 }
5519 crate::ast::expr::ExprKind::UnaryOp { operand, .. }
5520 | crate::ast::expr::ExprKind::Cast { expr: operand, .. }
5521 | crate::ast::expr::ExprKind::TryCast { expr: operand, .. }
5522 | crate::ast::expr::ExprKind::IsNull { expr: operand, .. } => expr_contains_window(operand),
5523 crate::ast::expr::ExprKind::Between {
5524 expr, low, high, ..
5525 } => expr_contains_window(expr) || expr_contains_window(low) || expr_contains_window(high),
5526 crate::ast::expr::ExprKind::Like {
5527 expr,
5528 pattern,
5529 escape,
5530 ..
5531 } => {
5532 expr_contains_window(expr)
5533 || expr_contains_window(pattern)
5534 || escape.as_deref().is_some_and(expr_contains_window)
5535 }
5536 crate::ast::expr::ExprKind::InList { expr, list, .. } => {
5537 expr_contains_window(expr) || list.iter().any(expr_contains_window)
5538 }
5539 _ => false,
5540 }
5541}
5542
5543fn typed_expr_contains_window(expr: &TypedExpr) -> bool {
5544 match &expr.kind {
5545 TypedExprKind::FunctionCall {
5546 args,
5547 filter,
5548 order_by,
5549 over,
5550 ..
5551 } => {
5552 over.is_some()
5553 || args.iter().any(typed_expr_contains_window)
5554 || filter.as_deref().is_some_and(typed_expr_contains_window)
5555 || order_by
5556 .iter()
5557 .any(|sort| typed_expr_contains_window(&sort.expr))
5558 }
5559 TypedExprKind::BinaryOp { left, right, .. } => {
5560 typed_expr_contains_window(left) || typed_expr_contains_window(right)
5561 }
5562 TypedExprKind::UnaryOp { operand, .. }
5563 | TypedExprKind::Cast { expr: operand, .. }
5564 | TypedExprKind::TryCast { expr: operand, .. }
5565 | TypedExprKind::IsNull { expr: operand, .. } => typed_expr_contains_window(operand),
5566 TypedExprKind::Between {
5567 expr, low, high, ..
5568 } => {
5569 typed_expr_contains_window(expr)
5570 || typed_expr_contains_window(low)
5571 || typed_expr_contains_window(high)
5572 }
5573 TypedExprKind::Like {
5574 expr,
5575 pattern,
5576 escape,
5577 ..
5578 } => {
5579 typed_expr_contains_window(expr)
5580 || typed_expr_contains_window(pattern)
5581 || escape.as_deref().is_some_and(typed_expr_contains_window)
5582 }
5583 TypedExprKind::InList { expr, list, .. } => {
5584 typed_expr_contains_window(expr) || list.iter().any(typed_expr_contains_window)
5585 }
5586 _ => false,
5587 }
5588}
5589
5590fn rewrite_projection_for_windows(
5591 projection: &Projection,
5592 window_map: &HashMap<String, usize>,
5593 base_width: usize,
5594 window_names: &[String],
5595) -> Result<Projection, PlannerError> {
5596 match projection {
5597 Projection::All(names) => Ok(Projection::All(names.clone())),
5598 Projection::Columns(columns) => Ok(Projection::Columns(
5599 columns
5600 .iter()
5601 .map(|column| {
5602 Ok(ProjectedColumn {
5603 expr: rewrite_expr_for_windows(
5604 &column.expr,
5605 window_map,
5606 base_width,
5607 window_names,
5608 )?,
5609 alias: column.alias.clone(),
5610 })
5611 })
5612 .collect::<Result<Vec<_>, PlannerError>>()?,
5613 )),
5614 }
5615}
5616
5617fn rewrite_expr_for_windows(
5618 expr: &TypedExpr,
5619 window_map: &HashMap<String, usize>,
5620 base_width: usize,
5621 window_names: &[String],
5622) -> Result<TypedExpr, PlannerError> {
5623 if let Some(index) = window_map.get(&expr_key(expr)) {
5624 return Ok(TypedExpr::column_ref(
5625 "__window__".to_string(),
5626 window_names
5627 .get(*index)
5628 .cloned()
5629 .unwrap_or_else(|| format!("__window_{index}")),
5630 base_width + index,
5631 expr.resolved_type.clone(),
5632 expr.span,
5633 ));
5634 }
5635
5636 let rewrite =
5637 |inner: &TypedExpr| rewrite_expr_for_windows(inner, window_map, base_width, window_names);
5638 let kind = match &expr.kind {
5639 TypedExprKind::FunctionCall {
5640 name,
5641 args,
5642 distinct,
5643 star,
5644 filter,
5645 order_by,
5646 over,
5647 } => {
5648 if over.is_some() {
5649 return Err(PlannerError::invalid_expression(
5650 "window expression is not part of the window plan".to_string(),
5651 ));
5652 }
5653 TypedExprKind::FunctionCall {
5654 name: name.clone(),
5655 args: args.iter().map(rewrite).collect::<Result<Vec<_>, _>>()?,
5656 distinct: *distinct,
5657 star: *star,
5658 filter: filter.as_deref().map(rewrite).transpose()?.map(Box::new),
5659 order_by: order_by
5660 .iter()
5661 .map(|sort| {
5662 Ok(SortExpr::new(
5663 rewrite(&sort.expr)?,
5664 sort.asc,
5665 sort.nulls_first,
5666 ))
5667 })
5668 .collect::<Result<Vec<_>, PlannerError>>()?,
5669 over: None,
5670 }
5671 }
5672 TypedExprKind::BinaryOp { left, op, right } => TypedExprKind::BinaryOp {
5673 left: Box::new(rewrite(left)?),
5674 op: *op,
5675 right: Box::new(rewrite(right)?),
5676 },
5677 TypedExprKind::UnaryOp { op, operand } => TypedExprKind::UnaryOp {
5678 op: *op,
5679 operand: Box::new(rewrite(operand)?),
5680 },
5681 TypedExprKind::Cast {
5682 expr: inner,
5683 target_type,
5684 } => TypedExprKind::Cast {
5685 expr: Box::new(rewrite(inner)?),
5686 target_type: target_type.clone(),
5687 },
5688 TypedExprKind::TryCast {
5689 expr: inner,
5690 target_type,
5691 } => TypedExprKind::TryCast {
5692 expr: Box::new(rewrite(inner)?),
5693 target_type: target_type.clone(),
5694 },
5695 TypedExprKind::Between {
5696 expr: inner,
5697 low,
5698 high,
5699 negated,
5700 } => TypedExprKind::Between {
5701 expr: Box::new(rewrite(inner)?),
5702 low: Box::new(rewrite(low)?),
5703 high: Box::new(rewrite(high)?),
5704 negated: *negated,
5705 },
5706 TypedExprKind::Like {
5707 expr: inner,
5708 pattern,
5709 escape,
5710 negated,
5711 kind,
5712 } => TypedExprKind::Like {
5713 expr: Box::new(rewrite(inner)?),
5714 pattern: Box::new(rewrite(pattern)?),
5715 escape: escape.as_deref().map(rewrite).transpose()?.map(Box::new),
5716 negated: *negated,
5717 kind: *kind,
5718 },
5719 TypedExprKind::InList {
5720 expr: inner,
5721 list,
5722 negated,
5723 } => TypedExprKind::InList {
5724 expr: Box::new(rewrite(inner)?),
5725 list: list.iter().map(rewrite).collect::<Result<Vec<_>, _>>()?,
5726 negated: *negated,
5727 },
5728 TypedExprKind::IsNull {
5729 expr: inner,
5730 negated,
5731 } => TypedExprKind::IsNull {
5732 expr: Box::new(rewrite(inner)?),
5733 negated: *negated,
5734 },
5735 _ => return Ok(expr.clone()),
5736 };
5737 Ok(TypedExpr {
5738 kind,
5739 resolved_type: expr.resolved_type.clone(),
5740 span: expr.span,
5741 })
5742}
5743
5744fn rewrite_expr_for_projected_output(
5750 expr: &TypedExpr,
5751 projection: &Projection,
5752 output_schema: &[ColumnMetadata],
5753) -> Result<TypedExpr, PlannerError> {
5754 let index = match projection {
5755 Projection::Columns(columns) => columns
5756 .iter()
5757 .position(|column| expr_key(&column.expr) == expr_key(expr)),
5758 Projection::All(_) => match &expr.kind {
5759 TypedExprKind::ColumnRef { column_index, .. }
5760 if *column_index < output_schema.len() =>
5761 {
5762 Some(*column_index)
5763 }
5764 _ => None,
5765 },
5766 };
5767 let Some(index) = index else {
5768 return Err(PlannerError::invalid_expression(
5769 "ORDER BY expression must appear in the SELECT projection for window queries"
5770 .to_string(),
5771 ));
5772 };
5773 let column = output_schema.get(index).ok_or_else(|| {
5774 PlannerError::invalid_expression(
5775 "ORDER BY projection index is outside the output schema".to_string(),
5776 )
5777 })?;
5778 Ok(TypedExpr::column_ref(
5779 "__project__".to_string(),
5780 column.name.clone(),
5781 index,
5782 column.data_type.clone(),
5783 expr.span,
5784 ))
5785}
5786
5787fn map_join_type(join_type: crate::ast::dml::JoinType) -> JoinType {
5788 match join_type {
5789 crate::ast::dml::JoinType::Inner => JoinType::Inner,
5790 crate::ast::dml::JoinType::Left => JoinType::Left,
5791 crate::ast::dml::JoinType::Right => JoinType::Right,
5792 crate::ast::dml::JoinType::Full => JoinType::Full,
5793 crate::ast::dml::JoinType::Cross => JoinType::Cross,
5794 }
5795}
5796
5797struct FoundScopedColumn {
5798 table: String,
5799 index: usize,
5800 ty: ResolvedType,
5801 partner_indices: Vec<usize>,
5802}
5803
5804fn find_scoped_column(
5805 scope: &[ScopedTable],
5806 column: &str,
5807 span: crate::ast::Span,
5808) -> Result<FoundScopedColumn, PlannerError> {
5809 let mut matches = Vec::new();
5810 for table in scope {
5811 if table.hidden_unqualified_columns.contains(column) {
5812 continue;
5813 }
5814 if let Some(local_idx) = table.table.get_column_index(column) {
5815 let meta = &table.table.columns[local_idx];
5816 matches.push(FoundScopedColumn {
5817 table: table.table.name.clone(),
5818 index: table.start_index + local_idx,
5819 ty: meta.data_type.clone(),
5820 partner_indices: table
5821 .merged_column_partners
5822 .get(column)
5823 .cloned()
5824 .unwrap_or_default(),
5825 });
5826 }
5827 }
5828 match matches.len() {
5829 0 => Err(PlannerError::column_not_found(column, "JOIN input", span)),
5830 1 => Ok(matches.remove(0)),
5831 _ => Err(PlannerError::ambiguous_column(
5832 column,
5833 scope.iter().map(|s| s.table.name.clone()).collect(),
5834 span,
5835 )),
5836 }
5837}
5838
5839fn merged_scoped_column_expr(
5840 found: &FoundScopedColumn,
5841 column: &str,
5842 span: crate::ast::Span,
5843) -> TypedExpr {
5844 let own = TypedExpr::column_ref(
5845 found.table.clone(),
5846 column.to_string(),
5847 found.index,
5848 found.ty.clone(),
5849 span,
5850 );
5851 if found.partner_indices.is_empty() {
5852 return own;
5853 }
5854
5855 let mut args = Vec::with_capacity(found.partner_indices.len() + 1);
5856 args.push(own);
5857 args.extend(found.partner_indices.iter().map(|&index| {
5858 TypedExpr::column_ref(
5859 found.table.clone(),
5860 column.to_string(),
5861 index,
5862 found.ty.clone(),
5863 span,
5864 )
5865 }));
5866 TypedExpr {
5867 kind: TypedExprKind::FunctionCall {
5868 name: "coalesce".to_string(),
5869 args,
5870 distinct: false,
5871 star: false,
5872 filter: None,
5873 order_by: Vec::new(),
5874 over: None,
5875 },
5876 resolved_type: found.ty.clone(),
5877 span,
5878 }
5879}
5880
5881fn projection_schema(
5882 projection: &Projection,
5883 input_schema: &[ColumnMetadata],
5884) -> Vec<ColumnMetadata> {
5885 match projection {
5886 Projection::All(names) => names
5887 .iter()
5888 .enumerate()
5889 .map(|(idx, name)| {
5890 let ty = (names.len() == input_schema.len())
5891 .then(|| input_schema.get(idx))
5892 .flatten()
5893 .or_else(|| input_schema.iter().find(|col| &col.name == name))
5894 .map(|col| col.data_type.clone())
5895 .unwrap_or(ResolvedType::Null);
5896 ColumnMetadata::new(name.clone(), ty)
5897 })
5898 .collect(),
5899 Projection::Columns(columns) => columns
5900 .iter()
5901 .enumerate()
5902 .map(|(idx, col)| {
5903 let name = col
5904 .alias
5905 .clone()
5906 .or_else(|| match &col.expr.kind {
5907 TypedExprKind::ColumnRef { column, .. } => Some(column.clone()),
5908 TypedExprKind::FunctionCall { name, args, .. }
5912 if name == "coalesce" && !args.is_empty() =>
5913 {
5914 let first_column = match &args[0].kind {
5915 TypedExprKind::ColumnRef { column, .. } => Some(column),
5916 _ => None,
5917 };
5918 first_column
5919 .filter(|column| {
5920 args.iter().all(|arg| {
5921 matches!(
5922 &arg.kind,
5923 TypedExprKind::ColumnRef { column: other, .. }
5924 if other == *column
5925 )
5926 })
5927 })
5928 .cloned()
5929 }
5930 _ => None,
5931 })
5932 .unwrap_or_else(|| format!("col_{idx}"));
5933 ColumnMetadata::new(name, col.expr.resolved_type.clone())
5934 })
5935 .collect(),
5936 }
5937}
5938
5939fn visible_wildcard_columns(schema: &[ColumnMetadata], scope: &[ScopedTable]) -> Vec<String> {
5940 schema
5941 .iter()
5942 .enumerate()
5943 .filter(|(index, column)| {
5944 !scope.iter().any(|table| {
5945 *index >= table.start_index
5946 && *index < table.start_index + table.table.columns.len()
5947 && table.hidden_unqualified_columns.contains(&column.name)
5948 })
5949 })
5950 .map(|(_, column)| column.name.clone())
5951 .collect()
5952}
5953
5954fn combine_join_shape(
5959 left: &PlannedRelation,
5960 right: &PlannedRelation,
5961 using: Option<&[String]>,
5962) -> (Vec<ColumnMetadata>, Vec<ScopedTable>) {
5963 let mut schema = left.schema.clone();
5964 schema.extend(right.schema.clone());
5965 let mut scope = left.scope.clone();
5966 let mut right_scope = right.scope.clone();
5967 if let Some(columns) = using {
5968 for column in columns {
5972 let right_index = right_scope.iter().find_map(|table| {
5973 table
5974 .table
5975 .get_column_index(column)
5976 .map(|index| table.start_index + index)
5977 });
5978 let Some(right_index) = right_index else {
5979 continue;
5980 };
5981 for table in &mut scope {
5982 if table.table.get_column_index(column).is_some() {
5983 table.merge_column_with(column, right_index);
5984 }
5985 }
5986 }
5987 for table in &mut right_scope {
5988 table.hide_unqualified_columns(columns);
5989 }
5990 }
5991 scope.extend(right_scope);
5992 (schema, scope)
5993}
5994
5995fn from_item_is_lateral(item: &FromItem) -> bool {
6000 match item {
6001 FromItem::Derived { lateral, .. } => *lateral,
6002 FromItem::Function { .. } => true,
6003 FromItem::Table { .. } | FromItem::Join { .. } => false,
6004 }
6005}
6006
6007fn lateral_outer_scope(
6015 left_scope: &[ScopedTable],
6016 base: usize,
6017 left_width: usize,
6018 outer_scope: &[ScopedTable],
6019) -> Vec<ScopedTable> {
6020 debug_assert!(
6021 left_scope.iter().all(|table| table.start_index >= base),
6022 "a FROM item's left sibling scope must start at the join's own base"
6023 );
6024 left_scope
6025 .iter()
6026 .cloned()
6027 .map(|mut table| {
6028 table.start_index -= base;
6029 table
6030 })
6031 .chain(outer_scope.iter().cloned().map(|mut table| {
6032 table.start_index += left_width;
6033 table
6034 }))
6035 .collect()
6036}
6037
6038fn apply_alias_columns(
6043 alias: &str,
6044 columns: &[String],
6045 schema: &mut [ColumnMetadata],
6046 span: crate::ast::Span,
6047) -> Result<(), PlannerError> {
6048 if columns.is_empty() {
6049 return Ok(());
6050 }
6051 if columns.len() != schema.len() {
6052 return Err(PlannerError::table_alias_column_count_mismatch(
6053 alias,
6054 columns.len(),
6055 schema.len(),
6056 span,
6057 ));
6058 }
6059 let mut names = HashSet::new();
6060 for name in columns {
6061 if !names.insert(name) {
6062 return Err(PlannerError::invalid_expression(format!(
6063 "relation alias '{alias}' declares column '{name}' more than once"
6064 )));
6065 }
6066 }
6067 for (column, name) in schema.iter_mut().zip(columns) {
6068 column.name.clone_from(name);
6069 }
6070 Ok(())
6071}
6072
6073fn offset_scope(scope: &[ScopedTable], offset: usize) -> Vec<ScopedTable> {
6074 scope
6075 .iter()
6076 .cloned()
6077 .map(|mut table| {
6078 table.start_index += offset;
6079 table.scope_level += 1;
6080 table
6081 })
6082 .collect()
6083}
6084
6085fn natural_join_columns(
6086 left_schema: &[ColumnMetadata],
6087 right_schema: &[ColumnMetadata],
6088) -> Vec<String> {
6089 let right_names = right_schema
6093 .iter()
6094 .map(|column| column.name.as_str())
6095 .collect::<HashSet<_>>();
6096 left_schema
6097 .iter()
6098 .filter(|left| right_names.contains(left.name.as_str()))
6099 .map(|column| column.name.clone())
6100 .collect()
6101}
6102
6103fn install_base_projection(plan: &mut LogicalPlan, projection: &Projection) {
6104 match plan {
6105 LogicalPlan::Scan {
6106 projection: scan_projection,
6107 ..
6108 } => *scan_projection = projection.clone(),
6109 LogicalPlan::Filter { input, .. } => install_base_projection(input, projection),
6110 _ => {}
6111 }
6112}
6113
6114fn is_aggregate_function(name: &str) -> bool {
6115 matches!(
6116 name.to_ascii_lowercase().as_str(),
6117 "count"
6118 | "sum"
6119 | "total"
6120 | "avg"
6121 | "min"
6122 | "max"
6123 | "group_concat"
6124 | "string_agg"
6125 | "percentile_disc"
6126 | "percentile_cont"
6127 ) || type_checker::is_portable_aggregate_name(&name.to_ascii_lowercase())
6128}
6129
6130fn expr_key(expr: &TypedExpr) -> String {
6131 format!("{:?}", expr.kind)
6132}
6133
6134fn distinct_on_expr_signature(expr: &TypedExpr) -> String {
6144 const MARKER: &str = "span: Span { start: Location { line: ";
6145 let rendered = format!("{:?}", expr.kind);
6146 let mut result = String::with_capacity(rendered.len());
6147 let mut rest = rendered.as_str();
6148 while let Some(position) = rest.find(MARKER) {
6149 let after = &rest[position + MARKER.len()..];
6150 match debug_span_tail_length(after) {
6151 Some(consumed) => {
6152 result.push_str(&rest[..position]);
6153 result.push_str("span: _");
6154 rest = &after[consumed..];
6155 }
6156 None => {
6157 let keep = position + MARKER.len();
6158 result.push_str(&rest[..keep]);
6159 rest = &rest[keep..];
6160 }
6161 }
6162 }
6163 result.push_str(rest);
6164 result
6165}
6166
6167fn debug_span_tail_length(input: &str) -> Option<usize> {
6171 fn digits(input: &str, offset: &mut usize) -> bool {
6172 let start = *offset;
6173 while input
6174 .as_bytes()
6175 .get(*offset)
6176 .is_some_and(u8::is_ascii_digit)
6177 {
6178 *offset += 1;
6179 }
6180 *offset > start
6181 }
6182 fn literal(input: &str, offset: &mut usize, expected: &str) -> bool {
6183 if input[*offset..].starts_with(expected) {
6184 *offset += expected.len();
6185 true
6186 } else {
6187 false
6188 }
6189 }
6190 let mut offset = 0;
6191 (digits(input, &mut offset)
6192 && literal(input, &mut offset, ", column: ")
6193 && digits(input, &mut offset)
6194 && literal(input, &mut offset, " }, end: Location { line: ")
6195 && digits(input, &mut offset)
6196 && literal(input, &mut offset, ", column: ")
6197 && digits(input, &mut offset)
6198 && literal(input, &mut offset, " } }"))
6199 .then_some(offset)
6200}
6201
6202fn build_distinct_on_sort_spec(
6219 key_exprs: Vec<TypedExpr>,
6220 user_order_by: Vec<SortExpr>,
6221 base_schema: &[ColumnMetadata],
6222 fallback_span: crate::ast::Span,
6223) -> Result<(usize, Vec<SortExpr>), PlannerError> {
6224 let key_signatures: Vec<String> = key_exprs.iter().map(distinct_on_expr_signature).collect();
6225 let mut consumed = vec![false; key_exprs.len()];
6226 let mut prefix: Vec<SortExpr> = Vec::new();
6227 let mut tail: Vec<SortExpr> = Vec::new();
6228 let mut prefix_ended = false;
6229 for sort in user_order_by {
6230 let signature = distinct_on_expr_signature(&sort.expr);
6231 if let Some(index) = key_signatures
6232 .iter()
6233 .position(|candidate| candidate == &signature)
6234 {
6235 if prefix_ended {
6236 return Err(PlannerError::distinct_on_order_by_mismatch(sort.expr.span));
6239 }
6240 consumed[index] = true;
6241 prefix.push(sort);
6242 } else {
6243 prefix_ended = true;
6244 tail.push(sort);
6245 }
6246 }
6247 let mut implicit: Vec<SortExpr> = Vec::new();
6248 for (index, key) in key_exprs.into_iter().enumerate() {
6249 if consumed[index] {
6250 continue;
6251 }
6252 if prefix_ended {
6253 return Err(PlannerError::distinct_on_order_by_mismatch(key.span));
6256 }
6257 implicit.push(SortExpr::new(key, true, false));
6258 }
6259 let key_count = prefix.len() + implicit.len();
6260 let mut order_by = prefix;
6261 order_by.append(&mut implicit);
6262 order_by.append(&mut tail);
6263 let mut seen_columns: HashSet<usize> = order_by
6264 .iter()
6265 .filter_map(|sort| match &sort.expr.kind {
6266 TypedExprKind::ColumnRef { column_index, .. } => Some(*column_index),
6267 _ => None,
6268 })
6269 .collect();
6270 for (index, column) in base_schema.iter().enumerate() {
6271 if seen_columns.insert(index) {
6272 order_by.push(SortExpr::new(
6273 TypedExpr::column_ref(
6274 String::new(),
6275 column.name.clone(),
6276 index,
6277 column.data_type.clone(),
6278 fallback_span,
6279 ),
6280 true,
6281 false,
6282 ));
6283 }
6284 }
6285 Ok((key_count, order_by))
6286}
6287
6288fn is_order_sensitive_aggregate(name: &str) -> bool {
6290 matches!(
6291 name.to_ascii_lowercase().as_str(),
6292 "group_concat"
6293 | "string_agg"
6294 | "percentile_disc"
6295 | "percentile_cont"
6296 | "mode"
6297 | "first"
6298 | "last"
6299 )
6300}
6301
6302fn sort_exprs_key(order_by: &[SortExpr]) -> Option<String> {
6303 if order_by.is_empty() {
6304 return None;
6305 }
6306 Some(
6307 order_by
6308 .iter()
6309 .map(|sort| format!("{}|{}|{}", expr_key(&sort.expr), sort.asc, sort.nulls_first))
6310 .collect::<Vec<_>>()
6311 .join(","),
6312 )
6313}
6314
6315#[allow(clippy::too_many_arguments)]
6316fn aggregate_signature(
6317 name: &str,
6318 distinct: bool,
6319 star: bool,
6320 arg: Option<&TypedExpr>,
6321 separator: Option<&String>,
6322 _expr: &TypedExpr,
6323 filter: Option<&TypedExpr>,
6324 order_by: &[SortExpr],
6325) -> AggregateSignature {
6326 AggregateSignature {
6327 name: canonical_aggregate_name(name),
6328 distinct,
6329 star,
6330 arg_key: arg.map(expr_key),
6331 extra_arg_keys: Vec::new(),
6332 separator: separator.cloned(),
6333 filter_key: filter.map(expr_key),
6334 order_key: sort_exprs_key(order_by),
6335 }
6336}
6337
6338fn canonical_aggregate_name(name: &str) -> String {
6339 match name.to_ascii_lowercase().as_str() {
6340 "variance" | "var_samp" => "var_samp".into(),
6341 "stddev" | "stddev_samp" => "stddev_samp".into(),
6342 "min_by" => "arg_min".into(),
6343 "max_by" => "arg_max".into(),
6344 lower => lower.into(),
6345 }
6346}
6347
6348fn build_group_key_map(group_keys: &[TypedExpr]) -> HashMap<String, usize> {
6349 let mut map = HashMap::new();
6350 for (idx, key) in group_keys.iter().enumerate() {
6351 map.insert(expr_key(key), idx);
6352 }
6353 map
6354}
6355
6356fn build_aggregate_map(aggregates: &[AggregateExpr]) -> HashMap<AggregateSignature, usize> {
6357 let mut map = HashMap::new();
6358 for (idx, agg) in aggregates.iter().enumerate() {
6359 let (name, separator, star, arg) = match &agg.function {
6360 AggregateFunction::Count => (
6361 "count".to_string(),
6362 None,
6363 agg.arg.is_none(),
6364 agg.arg.as_ref(),
6365 ),
6366 AggregateFunction::Sum => ("sum".to_string(), None, false, agg.arg.as_ref()),
6367 AggregateFunction::Total => ("total".to_string(), None, false, agg.arg.as_ref()),
6368 AggregateFunction::Avg => ("avg".to_string(), None, false, agg.arg.as_ref()),
6369 AggregateFunction::Min => ("min".to_string(), None, false, agg.arg.as_ref()),
6370 AggregateFunction::Max => ("max".to_string(), None, false, agg.arg.as_ref()),
6371 AggregateFunction::GroupConcat { separator } => (
6372 "group_concat".to_string(),
6373 separator.clone(),
6374 false,
6375 agg.arg.as_ref(),
6376 ),
6377 AggregateFunction::StringAgg { separator } => (
6378 "string_agg".to_string(),
6379 separator.clone(),
6380 false,
6381 agg.arg.as_ref(),
6382 ),
6383 AggregateFunction::PercentileDisc { fraction } => (
6384 "percentile_disc".to_string(),
6385 Some(format!("{fraction:?}")),
6386 false,
6387 None,
6388 ),
6389 AggregateFunction::PercentileCont { fraction } => (
6390 "percentile_cont".to_string(),
6391 Some(format!("{fraction:?}")),
6392 false,
6393 None,
6394 ),
6395 AggregateFunction::QuantileCont { fraction } => (
6396 "quantile_cont".to_string(),
6397 Some(format!("{fraction:?}")),
6398 false,
6399 agg.arg.as_ref(),
6400 ),
6401 AggregateFunction::Variance { sample } => (
6402 if *sample { "var_samp" } else { "var_pop" }.to_string(),
6403 None,
6404 false,
6405 agg.arg.as_ref(),
6406 ),
6407 AggregateFunction::Stddev { sample } => (
6408 if *sample { "stddev_samp" } else { "stddev_pop" }.to_string(),
6409 None,
6410 false,
6411 agg.arg.as_ref(),
6412 ),
6413 AggregateFunction::Covariance { sample } => (
6414 if *sample { "covar_samp" } else { "covar_pop" }.to_string(),
6415 None,
6416 false,
6417 agg.arg.as_ref(),
6418 ),
6419 AggregateFunction::Corr => ("corr".into(), None, false, agg.arg.as_ref()),
6420 AggregateFunction::Median => ("median".into(), None, false, agg.arg.as_ref()),
6421 AggregateFunction::Mode => (
6422 "mode".into(),
6423 None,
6424 false,
6425 agg.order_by.is_empty().then_some(()).and(agg.arg.as_ref()),
6426 ),
6427 AggregateFunction::RegrCount => ("regr_count".into(), None, false, agg.arg.as_ref()),
6428 AggregateFunction::RegrAvgX => ("regr_avgx".into(), None, false, agg.arg.as_ref()),
6429 AggregateFunction::RegrAvgY => ("regr_avgy".into(), None, false, agg.arg.as_ref()),
6430 AggregateFunction::RegrSxx => ("regr_sxx".into(), None, false, agg.arg.as_ref()),
6431 AggregateFunction::RegrSyy => ("regr_syy".into(), None, false, agg.arg.as_ref()),
6432 AggregateFunction::RegrSxy => ("regr_sxy".into(), None, false, agg.arg.as_ref()),
6433 AggregateFunction::RegrSlope => ("regr_slope".into(), None, false, agg.arg.as_ref()),
6434 AggregateFunction::RegrIntercept => {
6435 ("regr_intercept".into(), None, false, agg.arg.as_ref())
6436 }
6437 AggregateFunction::RegrR2 => ("regr_r2".into(), None, false, agg.arg.as_ref()),
6438 AggregateFunction::AnyValue => ("any_value".into(), None, false, agg.arg.as_ref()),
6439 AggregateFunction::First => ("first".into(), None, false, agg.arg.as_ref()),
6440 AggregateFunction::Last => ("last".into(), None, false, agg.arg.as_ref()),
6441 AggregateFunction::ArgMin => ("arg_min".into(), None, false, agg.arg.as_ref()),
6442 AggregateFunction::ArgMax => ("arg_max".into(), None, false, agg.arg.as_ref()),
6443 AggregateFunction::BitAnd => ("bit_and".into(), None, false, agg.arg.as_ref()),
6444 AggregateFunction::BitOr => ("bit_or".into(), None, false, agg.arg.as_ref()),
6445 AggregateFunction::BitXor => ("bit_xor".into(), None, false, agg.arg.as_ref()),
6446 AggregateFunction::BoolAnd => ("bool_and".into(), None, false, agg.arg.as_ref()),
6447 AggregateFunction::BoolOr => ("bool_or".into(), None, false, agg.arg.as_ref()),
6448 };
6449 let signature = AggregateSignature {
6450 name,
6451 distinct: agg.distinct,
6452 star,
6453 arg_key: arg.map(expr_key),
6454 extra_arg_keys: agg.extra_args.iter().map(expr_key).collect(),
6455 separator,
6456 filter_key: agg.filter.as_ref().map(expr_key),
6457 order_key: sort_exprs_key(&agg.order_by),
6458 };
6459 map.insert(signature, idx);
6460 }
6461 map
6462}
6463
6464pub(crate) const GROUPING_ID_COLUMN: &str = "__grouping_id";
6466const MAX_GROUPING_SETS: usize = 4096;
6468const MAX_CUBE_COLUMNS: usize = 12;
6470const MAX_GROUPING_KEYS: usize = 63;
6472
6473struct ExpandedGroupBy {
6475 group_keys: Vec<TypedExpr>,
6476 grouping_sets: Option<Vec<u64>>,
6477}
6478
6479fn grouping_full_mask(key_count: usize) -> u64 {
6480 if key_count == 0 {
6481 0
6482 } else {
6483 (1u64 << key_count) - 1
6484 }
6485}
6486
6487fn is_grouping_function(name: &str) -> bool {
6488 name.eq_ignore_ascii_case("grouping") || name.eq_ignore_ascii_case("grouping_id")
6489}
6490
6491struct GroupingRewrite {
6493 key_index: HashMap<String, usize>,
6495 key_count: usize,
6496 gid_index: usize,
6498 sets_present: bool,
6501}
6502
6503impl GroupingRewrite {
6504 fn new(
6505 group_keys: &[TypedExpr],
6506 aggregates: &[AggregateExpr],
6507 grouping_sets: &Option<Vec<u64>>,
6508 ) -> Self {
6509 let key_index = group_keys
6510 .iter()
6511 .enumerate()
6512 .map(|(index, key)| (expr_key(key), index))
6513 .collect();
6514 Self {
6515 key_index,
6516 key_count: group_keys.len(),
6517 gid_index: group_keys.len() + aggregates.len(),
6518 sets_present: grouping_sets.is_some(),
6519 }
6520 }
6521}
6522
6523fn expr_contains_grouping(expr: &crate::ast::expr::Expr) -> bool {
6525 use crate::ast::expr::ExprKind;
6526
6527 match &expr.kind {
6528 ExprKind::FunctionCall {
6529 name,
6530 args,
6531 order_by,
6532 within_group,
6533 filter,
6534 over,
6535 ..
6536 } => {
6537 is_grouping_function(name)
6538 || args.iter().any(expr_contains_grouping)
6539 || order_by
6540 .iter()
6541 .any(|sort| expr_contains_grouping(&sort.expr))
6542 || within_group
6543 .iter()
6544 .any(|sort| expr_contains_grouping(&sort.expr))
6545 || filter.as_deref().is_some_and(expr_contains_grouping)
6546 || over.as_ref().is_some_and(|window| {
6547 window.partition_by.iter().any(expr_contains_grouping)
6548 || window
6549 .order_by
6550 .iter()
6551 .any(|sort| expr_contains_grouping(&sort.expr))
6552 })
6553 }
6554 ExprKind::BinaryOp { left, right, .. } => {
6555 expr_contains_grouping(left) || expr_contains_grouping(right)
6556 }
6557 ExprKind::UnaryOp { operand, .. } => expr_contains_grouping(operand),
6558 ExprKind::TruthPredicate { expr, .. } => expr_contains_grouping(expr),
6559 ExprKind::IsDistinctFrom { left, right, .. } => {
6560 expr_contains_grouping(left) || expr_contains_grouping(right)
6561 }
6562 ExprKind::Row { items } => items.iter().any(expr_contains_grouping),
6563 ExprKind::Case {
6564 operand,
6565 branches,
6566 else_expr,
6567 } => {
6568 operand.as_deref().is_some_and(expr_contains_grouping)
6569 || branches.iter().any(|branch| {
6570 expr_contains_grouping(&branch.when) || expr_contains_grouping(&branch.then)
6571 })
6572 || else_expr.as_deref().is_some_and(expr_contains_grouping)
6573 }
6574 ExprKind::Cast { expr, .. } | ExprKind::TryCast { expr, .. } => {
6575 expr_contains_grouping(expr)
6576 }
6577 ExprKind::Between {
6578 expr, low, high, ..
6579 } => {
6580 expr_contains_grouping(expr)
6581 || expr_contains_grouping(low)
6582 || expr_contains_grouping(high)
6583 }
6584 ExprKind::Like {
6585 expr,
6586 pattern,
6587 escape,
6588 ..
6589 } => {
6590 expr_contains_grouping(expr)
6591 || expr_contains_grouping(pattern)
6592 || escape.as_deref().is_some_and(expr_contains_grouping)
6593 }
6594 ExprKind::InList { expr, list, .. } => {
6595 expr_contains_grouping(expr) || list.iter().any(expr_contains_grouping)
6596 }
6597 ExprKind::IsNull { expr, .. } => expr_contains_grouping(expr),
6598 ExprKind::ScalarSubquery { .. }
6599 | ExprKind::InSubquery { .. }
6600 | ExprKind::Exists { .. }
6601 | ExprKind::Quantified { .. }
6602 | ExprKind::Literal { .. }
6603 | ExprKind::VectorLiteral { .. }
6604 | ExprKind::ColumnRef { .. } => false,
6605 }
6606}
6607
6608fn typed_expr_contains_grouping(expr: &TypedExpr) -> bool {
6609 match &expr.kind {
6610 TypedExprKind::FunctionCall {
6611 name,
6612 args,
6613 filter,
6614 order_by,
6615 over,
6616 ..
6617 } => {
6618 is_grouping_function(name)
6619 || args.iter().any(typed_expr_contains_grouping)
6620 || filter.as_deref().is_some_and(typed_expr_contains_grouping)
6621 || order_by
6622 .iter()
6623 .any(|sort| typed_expr_contains_grouping(&sort.expr))
6624 || over.as_ref().is_some_and(|window| {
6625 window.partition_by.iter().any(typed_expr_contains_grouping)
6626 || window
6627 .order_by
6628 .iter()
6629 .any(|sort| typed_expr_contains_grouping(&sort.expr))
6630 })
6631 }
6632 TypedExprKind::BinaryOp { left, right, .. } => {
6633 typed_expr_contains_grouping(left) || typed_expr_contains_grouping(right)
6634 }
6635 TypedExprKind::UnaryOp { operand, .. } => typed_expr_contains_grouping(operand),
6636 TypedExprKind::Cast { expr, .. } | TypedExprKind::TryCast { expr, .. } => {
6637 typed_expr_contains_grouping(expr)
6638 }
6639 TypedExprKind::Case {
6640 operand,
6641 branches,
6642 else_expr,
6643 } => {
6644 operand.as_deref().is_some_and(typed_expr_contains_grouping)
6645 || branches.iter().any(|branch| {
6646 typed_expr_contains_grouping(&branch.when)
6647 || typed_expr_contains_grouping(&branch.then)
6648 })
6649 || else_expr
6650 .as_deref()
6651 .is_some_and(typed_expr_contains_grouping)
6652 }
6653 TypedExprKind::Between {
6654 expr, low, high, ..
6655 } => {
6656 typed_expr_contains_grouping(expr)
6657 || typed_expr_contains_grouping(low)
6658 || typed_expr_contains_grouping(high)
6659 }
6660 TypedExprKind::Like {
6661 expr,
6662 pattern,
6663 escape,
6664 ..
6665 } => {
6666 typed_expr_contains_grouping(expr)
6667 || typed_expr_contains_grouping(pattern)
6668 || escape
6669 .as_ref()
6670 .is_some_and(|inner| typed_expr_contains_grouping(inner))
6671 }
6672 TypedExprKind::InList { expr, list, .. } => {
6673 typed_expr_contains_grouping(expr) || list.iter().any(typed_expr_contains_grouping)
6674 }
6675 TypedExprKind::IsNull { expr, .. } => typed_expr_contains_grouping(expr),
6676 TypedExprKind::InSubquery { expr, .. } => typed_expr_contains_grouping(expr),
6677 TypedExprKind::Quantified { expr, .. } => typed_expr_contains_grouping(expr),
6678 TypedExprKind::Literal(_)
6679 | TypedExprKind::VectorLiteral(_)
6680 | TypedExprKind::ColumnRef { .. }
6681 | TypedExprKind::ScalarSubquery(_)
6682 | TypedExprKind::Exists { .. } => false,
6683 }
6684}
6685
6686fn bigint_literal(value: u64, span: crate::ast::Span) -> TypedExpr {
6687 TypedExpr {
6688 kind: TypedExprKind::Literal(Literal::Number(value.to_string())),
6689 resolved_type: ResolvedType::BigInt,
6690 span,
6691 }
6692}
6693
6694fn bigint_binary_op(
6695 left: TypedExpr,
6696 op: crate::ast::expr::BinaryOp,
6697 right: TypedExpr,
6698 span: crate::ast::Span,
6699) -> TypedExpr {
6700 TypedExpr {
6701 kind: TypedExprKind::BinaryOp {
6702 left: Box::new(left),
6703 op,
6704 right: Box::new(right),
6705 },
6706 resolved_type: ResolvedType::BigInt,
6707 span,
6708 }
6709}
6710
6711fn lower_grouping_call(
6719 args: &[TypedExpr],
6720 context: &GroupingRewrite,
6721 span: crate::ast::Span,
6722) -> Result<TypedExpr, PlannerError> {
6723 use crate::ast::expr::BinaryOp as AstBinaryOp;
6724
6725 if args.is_empty() {
6726 return Err(PlannerError::invalid_expression(
6727 "GROUPING requires at least one argument".to_string(),
6728 ));
6729 }
6730 if args.len() > MAX_GROUPING_KEYS {
6731 return Err(PlannerError::invalid_expression(format!(
6732 "GROUPING accepts at most {MAX_GROUPING_KEYS} arguments"
6733 )));
6734 }
6735 let mut key_positions = Vec::with_capacity(args.len());
6736 for arg in args {
6737 let Some(&position) = context.key_index.get(&expr_key(arg)) else {
6738 return Err(PlannerError::invalid_expression(
6739 "arguments to GROUPING must be grouping expressions of the query".to_string(),
6740 ));
6741 };
6742 key_positions.push(position);
6743 }
6744
6745 if !context.sets_present {
6746 return Ok(bigint_literal(0, span));
6748 }
6749
6750 let argument_count = key_positions.len();
6751 let mut sum: Option<TypedExpr> = None;
6752 for (argument, key_position) in key_positions.into_iter().enumerate() {
6753 let gid_ref = TypedExpr::column_ref(
6754 "__agg__".to_string(),
6755 GROUPING_ID_COLUMN.to_string(),
6756 context.gid_index,
6757 ResolvedType::BigInt,
6758 span,
6759 );
6760 let excluded_shift = (context.key_count - 1 - key_position) as u32;
6761 let bit = bigint_binary_op(
6762 bigint_binary_op(
6763 gid_ref,
6764 AstBinaryOp::Div,
6765 bigint_literal(1u64 << excluded_shift, span),
6766 span,
6767 ),
6768 AstBinaryOp::Mod,
6769 bigint_literal(2, span),
6770 span,
6771 );
6772 let weight = 1u64 << (argument_count - 1 - argument);
6773 let term = if weight == 1 {
6774 bit
6775 } else {
6776 bigint_binary_op(bit, AstBinaryOp::Mul, bigint_literal(weight, span), span)
6777 };
6778 sum = Some(match sum {
6779 None => term,
6780 Some(current) => bigint_binary_op(current, AstBinaryOp::Add, term, span),
6781 });
6782 }
6783 Ok(sum.expect("GROUPING argument list is non-empty"))
6784}
6785
6786fn rewrite_grouping_calls(
6793 expr: &TypedExpr,
6794 context: &GroupingRewrite,
6795) -> Result<TypedExpr, PlannerError> {
6796 let rebuild = |inner: &TypedExpr| rewrite_grouping_calls(inner, context);
6797 let rebuild_box = |inner: &TypedExpr| -> Result<Box<TypedExpr>, PlannerError> {
6798 Ok(Box::new(rewrite_grouping_calls(inner, context)?))
6799 };
6800 let kind = match &expr.kind {
6801 TypedExprKind::FunctionCall {
6802 name,
6803 args,
6804 distinct,
6805 star,
6806 filter,
6807 order_by,
6808 over,
6809 } => {
6810 if is_grouping_function(name) {
6811 if over.is_some() {
6812 return Err(PlannerError::invalid_expression(
6813 "GROUPING cannot be used as a window function".to_string(),
6814 ));
6815 }
6816 return lower_grouping_call(args, context, expr.span);
6817 }
6818 if over.is_none() && is_aggregate_function(name) {
6819 if args.iter().any(typed_expr_contains_grouping)
6820 || filter.as_deref().is_some_and(typed_expr_contains_grouping)
6821 || order_by
6822 .iter()
6823 .any(|sort| typed_expr_contains_grouping(&sort.expr))
6824 {
6825 return Err(PlannerError::invalid_expression(
6826 "GROUPING cannot appear inside aggregate function arguments".to_string(),
6827 ));
6828 }
6829 return Ok(expr.clone());
6830 }
6831 TypedExprKind::FunctionCall {
6832 name: name.clone(),
6833 args: args.iter().map(rebuild).collect::<Result<Vec<_>, _>>()?,
6834 distinct: *distinct,
6835 star: *star,
6836 filter: filter.as_deref().map(rebuild_box).transpose()?,
6837 order_by: order_by
6838 .iter()
6839 .map(|sort| {
6840 Ok(SortExpr::new(
6841 rebuild(&sort.expr)?,
6842 sort.asc,
6843 sort.nulls_first,
6844 ))
6845 })
6846 .collect::<Result<Vec<_>, PlannerError>>()?,
6847 over: over
6848 .as_ref()
6849 .map(|window| {
6850 Ok(typed_expr::TypedWindowSpec {
6851 partition_by: window
6852 .partition_by
6853 .iter()
6854 .map(rebuild)
6855 .collect::<Result<Vec<_>, _>>()?,
6856 order_by: window
6857 .order_by
6858 .iter()
6859 .map(|sort| {
6860 Ok(SortExpr::new(
6861 rebuild(&sort.expr)?,
6862 sort.asc,
6863 sort.nulls_first,
6864 ))
6865 })
6866 .collect::<Result<Vec<_>, PlannerError>>()?,
6867 frame: window.frame.clone(),
6868 })
6869 })
6870 .transpose()
6871 .map_err(|error: PlannerError| error)?,
6872 }
6873 }
6874 TypedExprKind::BinaryOp { left, op, right } => TypedExprKind::BinaryOp {
6875 left: rebuild_box(left)?,
6876 op: *op,
6877 right: rebuild_box(right)?,
6878 },
6879 TypedExprKind::UnaryOp { op, operand } => TypedExprKind::UnaryOp {
6880 op: *op,
6881 operand: rebuild_box(operand)?,
6882 },
6883 TypedExprKind::Case {
6884 operand,
6885 branches,
6886 else_expr,
6887 } => TypedExprKind::Case {
6888 operand: operand.as_deref().map(rebuild_box).transpose()?,
6889 branches: branches
6890 .iter()
6891 .map(|branch| {
6892 Ok(TypedCaseWhen {
6893 when: rebuild(&branch.when)?,
6894 then: rebuild(&branch.then)?,
6895 })
6896 })
6897 .collect::<Result<Vec<_>, PlannerError>>()?,
6898 else_expr: else_expr.as_deref().map(rebuild_box).transpose()?,
6899 },
6900 TypedExprKind::Cast {
6901 expr: inner,
6902 target_type,
6903 } => TypedExprKind::Cast {
6904 expr: rebuild_box(inner)?,
6905 target_type: target_type.clone(),
6906 },
6907 TypedExprKind::TryCast {
6908 expr: inner,
6909 target_type,
6910 } => TypedExprKind::TryCast {
6911 expr: rebuild_box(inner)?,
6912 target_type: target_type.clone(),
6913 },
6914 TypedExprKind::Between {
6915 expr: inner,
6916 low,
6917 high,
6918 negated,
6919 } => TypedExprKind::Between {
6920 expr: rebuild_box(inner)?,
6921 low: rebuild_box(low)?,
6922 high: rebuild_box(high)?,
6923 negated: *negated,
6924 },
6925 TypedExprKind::Like {
6926 expr: inner,
6927 pattern,
6928 escape,
6929 negated,
6930 kind,
6931 } => TypedExprKind::Like {
6932 expr: rebuild_box(inner)?,
6933 pattern: rebuild_box(pattern)?,
6934 escape: escape.as_deref().map(rebuild_box).transpose()?,
6935 negated: *negated,
6936 kind: *kind,
6937 },
6938 TypedExprKind::InList {
6939 expr: inner,
6940 list,
6941 negated,
6942 } => TypedExprKind::InList {
6943 expr: rebuild_box(inner)?,
6944 list: list.iter().map(rebuild).collect::<Result<Vec<_>, _>>()?,
6945 negated: *negated,
6946 },
6947 TypedExprKind::IsNull {
6948 expr: inner,
6949 negated,
6950 } => TypedExprKind::IsNull {
6951 expr: rebuild_box(inner)?,
6952 negated: *negated,
6953 },
6954 TypedExprKind::InSubquery {
6955 expr: inner,
6956 subquery,
6957 negated,
6958 } => TypedExprKind::InSubquery {
6959 expr: rebuild_box(inner)?,
6960 subquery: subquery.clone(),
6961 negated: *negated,
6962 },
6963 TypedExprKind::Quantified {
6964 expr: inner,
6965 op,
6966 quantifier,
6967 subquery,
6968 } => TypedExprKind::Quantified {
6969 expr: rebuild_box(inner)?,
6970 op: *op,
6971 quantifier: *quantifier,
6972 subquery: subquery.clone(),
6973 },
6974 TypedExprKind::Literal(_)
6975 | TypedExprKind::VectorLiteral(_)
6976 | TypedExprKind::ColumnRef { .. }
6977 | TypedExprKind::ScalarSubquery(_)
6978 | TypedExprKind::Exists { .. } => return Ok(expr.clone()),
6979 };
6980 Ok(TypedExpr {
6981 kind,
6982 resolved_type: expr.resolved_type.clone(),
6983 span: expr.span,
6984 })
6985}
6986
6987fn build_aggregate_schema(
6988 group_keys: &[TypedExpr],
6989 aggregates: &[AggregateExpr],
6990) -> Vec<ColumnMetadata> {
6991 let mut schema = Vec::new();
6992 for (idx, key) in group_keys.iter().enumerate() {
6993 let name = match &key.kind {
6994 TypedExprKind::ColumnRef { column, .. } => column.clone(),
6995 _ => format!("group_{idx}"),
6996 };
6997 schema.push(ColumnMetadata::new(name, key.resolved_type.clone()));
6998 }
6999 for (idx, agg) in aggregates.iter().enumerate() {
7000 let name = match &agg.function {
7001 AggregateFunction::Count => format!("count_{idx}"),
7002 AggregateFunction::Sum => format!("sum_{idx}"),
7003 AggregateFunction::Total => format!("total_{idx}"),
7004 AggregateFunction::Avg => format!("avg_{idx}"),
7005 AggregateFunction::Min => format!("min_{idx}"),
7006 AggregateFunction::Max => format!("max_{idx}"),
7007 AggregateFunction::GroupConcat { .. } => format!("group_concat_{idx}"),
7008 AggregateFunction::StringAgg { .. } => format!("string_agg_{idx}"),
7009 AggregateFunction::PercentileDisc { .. } => format!("percentile_disc_{idx}"),
7010 _ => format!("aggregate_{idx}"),
7011 };
7012 schema.push(ColumnMetadata::new(name, agg.result_type.clone()));
7013 }
7014 schema
7015}
7016
7017fn rewrite_expr_with_maps(
7018 expr: &TypedExpr,
7019 group_key_map: &HashMap<String, usize>,
7020 aggregate_map: &HashMap<AggregateSignature, usize>,
7021 output_names: &[String],
7022) -> Result<TypedExpr, PlannerError> {
7023 let group_key_count = output_names.len().saturating_sub(aggregate_map.len());
7024 let key = expr_key(expr);
7025 if let Some(idx) = group_key_map.get(&key) {
7026 return Ok(make_output_column_ref(
7027 *idx,
7028 output_names,
7029 expr.resolved_type.clone(),
7030 expr.span,
7031 ));
7032 }
7033
7034 match &expr.kind {
7035 TypedExprKind::FunctionCall {
7036 name,
7037 args,
7038 distinct,
7039 star,
7040 filter,
7041 order_by,
7042 over: None,
7043 } if is_aggregate_function(name) => {
7044 let lower = name.to_ascii_lowercase();
7045 let is_percentile = matches!(lower.as_str(), "percentile_disc" | "percentile_cont");
7046 let separator = if name.eq_ignore_ascii_case("group_concat") && args.len() == 2 {
7047 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
7048 Some(value.clone())
7049 } else {
7050 return Err(PlannerError::invalid_expression(
7051 "GROUP_CONCAT separator must be a string literal".to_string(),
7052 ));
7053 }
7054 } else if name.eq_ignore_ascii_case("string_agg") && args.len() == 2 {
7055 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
7056 Some(value.clone())
7057 } else {
7058 return Err(PlannerError::invalid_expression(
7059 "STRING_AGG separator must be a string literal".to_string(),
7060 ));
7061 }
7062 } else if is_percentile && args.len() == 1 {
7063 Some(format!(
7064 "{:?}",
7065 type_checker::percentile_fraction_named(&lower, &args[0])?
7066 ))
7067 } else if lower == "quantile_cont" && args.len() == 2 {
7068 Some(format!(
7069 "{:?}",
7070 type_checker::percentile_fraction_named(&lower, &args[1])?
7071 ))
7072 } else {
7073 None
7074 };
7075 let signature = AggregateSignature {
7076 name: canonical_aggregate_name(name),
7077 distinct: *distinct,
7078 star: *star,
7079 arg_key: if is_percentile {
7080 None
7081 } else {
7082 args.first().map(expr_key)
7083 },
7084 extra_arg_keys: if matches!(
7085 lower.as_str(),
7086 "group_concat"
7087 | "string_agg"
7088 | "percentile_disc"
7089 | "percentile_cont"
7090 | "quantile_cont"
7091 ) {
7092 Vec::new()
7093 } else {
7094 args.iter().skip(1).map(expr_key).collect()
7095 },
7096 separator,
7097 filter_key: filter.as_deref().map(expr_key),
7098 order_key: if is_order_sensitive_aggregate(name) {
7099 sort_exprs_key(order_by)
7100 } else {
7101 None
7102 },
7103 };
7104 let idx = aggregate_map.get(&signature).ok_or_else(|| {
7105 PlannerError::invalid_expression(
7106 "aggregate in expression is not part of plan".to_string(),
7107 )
7108 })?;
7109 let output_index = group_key_count + idx;
7110 Ok(make_output_column_ref(
7111 output_index,
7112 output_names,
7113 expr.resolved_type.clone(),
7114 expr.span,
7115 ))
7116 }
7117 TypedExprKind::FunctionCall {
7118 name,
7119 args,
7120 distinct,
7121 star,
7122 filter,
7123 order_by,
7124 over,
7125 } => {
7126 if over.is_none() && (*distinct || *star) {
7127 return Err(PlannerError::invalid_expression(
7128 "DISTINCT/STAR modifiers are only supported for aggregates".to_string(),
7129 ));
7130 }
7131 if filter.is_some() || !order_by.is_empty() {
7132 return Err(PlannerError::invalid_expression(
7136 "aggregate in expression is not part of plan".to_string(),
7137 ));
7138 }
7139 let mut rewritten_args = Vec::with_capacity(args.len());
7140 for arg in args {
7141 rewritten_args.push(rewrite_expr_with_maps(
7142 arg,
7143 group_key_map,
7144 aggregate_map,
7145 output_names,
7146 )?);
7147 }
7148 let over = over
7149 .as_ref()
7150 .map(|window| {
7151 let partition_by = window
7152 .partition_by
7153 .iter()
7154 .map(|expr| {
7155 rewrite_expr_with_maps(expr, group_key_map, aggregate_map, output_names)
7156 })
7157 .collect::<Result<Vec<_>, PlannerError>>()?;
7158 let order_by = window
7159 .order_by
7160 .iter()
7161 .map(|sort| {
7162 Ok(SortExpr::new(
7163 rewrite_expr_with_maps(
7164 &sort.expr,
7165 group_key_map,
7166 aggregate_map,
7167 output_names,
7168 )?,
7169 sort.asc,
7170 sort.nulls_first,
7171 ))
7172 })
7173 .collect::<Result<Vec<_>, PlannerError>>()?;
7174 Ok(crate::planner::typed_expr::TypedWindowSpec {
7175 partition_by,
7176 order_by,
7177 frame: window.frame.clone(),
7178 })
7179 })
7180 .transpose()?;
7181 Ok(TypedExpr {
7182 kind: TypedExprKind::FunctionCall {
7183 name: name.clone(),
7184 args: rewritten_args,
7185 distinct: *distinct,
7186 star: *star,
7187 filter: None,
7188 order_by: Vec::new(),
7189 over,
7190 },
7191 resolved_type: expr.resolved_type.clone(),
7192 span: expr.span,
7193 })
7194 }
7195 TypedExprKind::BinaryOp { left, op, right } => {
7196 let left = rewrite_expr_with_maps(left, group_key_map, aggregate_map, output_names)?;
7197 let right = rewrite_expr_with_maps(right, group_key_map, aggregate_map, output_names)?;
7198 Ok(TypedExpr {
7199 kind: TypedExprKind::BinaryOp {
7200 left: Box::new(left),
7201 op: *op,
7202 right: Box::new(right),
7203 },
7204 resolved_type: expr.resolved_type.clone(),
7205 span: expr.span,
7206 })
7207 }
7208 TypedExprKind::UnaryOp { op, operand } => {
7209 let operand =
7210 rewrite_expr_with_maps(operand, group_key_map, aggregate_map, output_names)?;
7211 Ok(TypedExpr {
7212 kind: TypedExprKind::UnaryOp {
7213 op: *op,
7214 operand: Box::new(operand),
7215 },
7216 resolved_type: expr.resolved_type.clone(),
7217 span: expr.span,
7218 })
7219 }
7220 TypedExprKind::Case {
7221 operand,
7222 branches,
7223 else_expr,
7224 } => {
7225 let operand = operand
7226 .as_deref()
7227 .map(|operand| {
7228 rewrite_expr_with_maps(operand, group_key_map, aggregate_map, output_names)
7229 .map(Box::new)
7230 })
7231 .transpose()?;
7232 let mut rewritten_branches = Vec::with_capacity(branches.len());
7233 for branch in branches {
7234 rewritten_branches.push(TypedCaseWhen {
7235 when: rewrite_expr_with_maps(
7236 &branch.when,
7237 group_key_map,
7238 aggregate_map,
7239 output_names,
7240 )?,
7241 then: rewrite_expr_with_maps(
7242 &branch.then,
7243 group_key_map,
7244 aggregate_map,
7245 output_names,
7246 )?,
7247 });
7248 }
7249 let else_expr = else_expr
7250 .as_deref()
7251 .map(|else_expr| {
7252 rewrite_expr_with_maps(else_expr, group_key_map, aggregate_map, output_names)
7253 .map(Box::new)
7254 })
7255 .transpose()?;
7256 Ok(TypedExpr {
7257 kind: TypedExprKind::Case {
7258 operand,
7259 branches: rewritten_branches,
7260 else_expr,
7261 },
7262 resolved_type: expr.resolved_type.clone(),
7263 span: expr.span,
7264 })
7265 }
7266 TypedExprKind::Between {
7267 expr: inner,
7268 low,
7269 high,
7270 negated,
7271 } => {
7272 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
7273 let low = rewrite_expr_with_maps(low, group_key_map, aggregate_map, output_names)?;
7274 let high = rewrite_expr_with_maps(high, group_key_map, aggregate_map, output_names)?;
7275 Ok(TypedExpr {
7276 kind: TypedExprKind::Between {
7277 expr: Box::new(inner),
7278 low: Box::new(low),
7279 high: Box::new(high),
7280 negated: *negated,
7281 },
7282 resolved_type: expr.resolved_type.clone(),
7283 span: expr.span,
7284 })
7285 }
7286 TypedExprKind::Like {
7287 expr: inner,
7288 pattern,
7289 escape,
7290 negated,
7291 kind,
7292 } => {
7293 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
7294 let pattern =
7295 rewrite_expr_with_maps(pattern, group_key_map, aggregate_map, output_names)?;
7296 let escape = if let Some(esc) = escape {
7297 Some(Box::new(rewrite_expr_with_maps(
7298 esc,
7299 group_key_map,
7300 aggregate_map,
7301 output_names,
7302 )?))
7303 } else {
7304 None
7305 };
7306 Ok(TypedExpr {
7307 kind: TypedExprKind::Like {
7308 expr: Box::new(inner),
7309 pattern: Box::new(pattern),
7310 escape,
7311 negated: *negated,
7312 kind: *kind,
7313 },
7314 resolved_type: expr.resolved_type.clone(),
7315 span: expr.span,
7316 })
7317 }
7318 TypedExprKind::InList {
7319 expr: inner,
7320 list,
7321 negated,
7322 } => {
7323 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
7324 let mut rewritten_list = Vec::with_capacity(list.len());
7325 for item in list {
7326 rewritten_list.push(rewrite_expr_with_maps(
7327 item,
7328 group_key_map,
7329 aggregate_map,
7330 output_names,
7331 )?);
7332 }
7333 Ok(TypedExpr {
7334 kind: TypedExprKind::InList {
7335 expr: Box::new(inner),
7336 list: rewritten_list,
7337 negated: *negated,
7338 },
7339 resolved_type: expr.resolved_type.clone(),
7340 span: expr.span,
7341 })
7342 }
7343 TypedExprKind::IsNull {
7344 expr: inner,
7345 negated,
7346 } => {
7347 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
7348 Ok(TypedExpr {
7349 kind: TypedExprKind::IsNull {
7350 expr: Box::new(inner),
7351 negated: *negated,
7352 },
7353 resolved_type: expr.resolved_type.clone(),
7354 span: expr.span,
7355 })
7356 }
7357 TypedExprKind::Literal(_) | TypedExprKind::VectorLiteral(_) => Ok(expr.clone()),
7358 TypedExprKind::ColumnRef { table, .. } if table == "__agg__" => Ok(expr.clone()),
7361 TypedExprKind::ColumnRef { .. } => Err(PlannerError::invalid_expression(
7362 "column reference must appear in GROUP BY or be aggregated".to_string(),
7363 )),
7364 TypedExprKind::Cast {
7365 expr: inner,
7366 target_type,
7367 } => {
7368 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
7369 Ok(TypedExpr {
7370 kind: TypedExprKind::Cast {
7371 expr: Box::new(inner),
7372 target_type: target_type.clone(),
7373 },
7374 resolved_type: expr.resolved_type.clone(),
7375 span: expr.span,
7376 })
7377 }
7378 TypedExprKind::TryCast {
7379 expr: inner,
7380 target_type,
7381 } => {
7382 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
7383 Ok(TypedExpr {
7384 kind: TypedExprKind::TryCast {
7385 expr: Box::new(inner),
7386 target_type: target_type.clone(),
7387 },
7388 resolved_type: expr.resolved_type.clone(),
7389 span: expr.span,
7390 })
7391 }
7392 TypedExprKind::ScalarSubquery(_)
7393 | TypedExprKind::InSubquery { .. }
7394 | TypedExprKind::Exists { .. }
7395 | TypedExprKind::Quantified { .. } => Ok(expr.clone()),
7396 }
7397}
7398
7399fn make_output_column_ref(
7400 index: usize,
7401 output_names: &[String],
7402 resolved_type: ResolvedType,
7403 span: crate::ast::Span,
7404) -> TypedExpr {
7405 let name = output_names
7406 .get(index)
7407 .cloned()
7408 .unwrap_or_else(|| format!("col_{index}"));
7409 TypedExpr::column_ref("__agg__".to_string(), name, index, resolved_type, span)
7410}