1pub mod aggregate_expr;
13mod error;
14pub mod knn_optimizer;
15pub mod logical_plan;
16pub mod name_resolver;
17pub mod type_checker;
18pub mod typed_expr;
19pub mod types;
20
21#[cfg(test)]
22mod planner_tests;
23
24pub use aggregate_expr::{AggregateExpr, AggregateFunction};
25pub use error::PlannerError;
26pub use knn_optimizer::{KnnPattern, SortDirection, detect_knn_pattern};
27pub use logical_plan::{JoinType, LogicalPlan};
28pub use name_resolver::{NameResolver, ResolvedColumn};
29pub use type_checker::{ScopedTable, TypeChecker};
30pub use typed_expr::{
31 ProjectedColumn, Projection, SortExpr, TypedAssignment, TypedExpr, TypedExprKind,
32};
33pub use types::ResolvedType;
34
35use crate::ast::ddl::{
36 ColumnConstraint, ColumnDef, CreateIndex, CreateTable, DropIndex, DropTable,
37};
38use crate::ast::dml::{
39 Delete, FromItem, Insert, InsertSource, LITERAL_TABLE, OrderByExpr, Select, SelectItem, Update,
40};
41use crate::ast::expr::Literal;
42use crate::ast::{PragmaValue, Spanned, Statement, StatementKind};
43use crate::catalog::{Catalog, ColumnMetadata, IndexMetadata, TableMetadata};
44use crate::{AlopexDialect, DataSourceFormat, Parser, SqlError, TableType};
45use std::collections::{HashMap, HashSet};
46
47struct PlannedRelation {
48 plan: LogicalPlan,
49 schema: Vec<ColumnMetadata>,
50 scope: Vec<ScopedTable>,
51}
52
53#[derive(Debug, Clone)]
59pub struct PlannedStatement {
60 pub plan: LogicalPlan,
62 pub routing_input: RoutingInput,
64}
65
66impl PlannedStatement {
67 pub fn statement_kind(&self) -> &StatementKind {
69 &self.routing_input.statement_kind
70 }
71
72 pub fn table_references(&self) -> &[TableReference] {
74 &self.routing_input.table_references
75 }
76
77 pub fn diagnostics(&self) -> &[PlanningDiagnostic] {
80 &self.routing_input.diagnostics
81 }
82}
83
84#[derive(Debug, Clone)]
86pub struct RoutingInput {
87 pub statement_kind: StatementKind,
90 pub table_references: Vec<TableReference>,
92 pub diagnostics: Vec<PlanningDiagnostic>,
94}
95
96#[derive(Debug, Clone, PartialEq, Eq)]
98pub struct TableReference {
99 pub table_name: String,
101 pub access: TableReferenceAccess,
103 pub source: TableReferenceSource,
105}
106
107impl TableReference {
108 pub fn new(
109 table_name: impl Into<String>,
110 access: TableReferenceAccess,
111 source: TableReferenceSource,
112 ) -> Self {
113 Self {
114 table_name: table_name.into(),
115 access,
116 source,
117 }
118 }
119}
120
121#[derive(Debug, Clone, Copy, PartialEq, Eq)]
123pub enum TableReferenceAccess {
124 Read,
126 Write,
128 Create,
130 Drop,
132 Metadata,
134}
135
136#[derive(Debug, Clone, Copy, PartialEq, Eq)]
138pub enum TableReferenceSource {
139 TopLevelPlanTableName,
141 LogicalPlanScan,
143 LogicalPlanMutationTarget,
145 LogicalPlanDdlTarget,
147 LogicalPlanIndexTarget,
149 TypedExprSubquery,
151}
152
153#[derive(Debug, Clone, Copy, PartialEq, Eq)]
155pub enum PlanningDiagnosticSeverity {
156 Info,
157 Warning,
158}
159
160#[derive(Debug, Clone, PartialEq, Eq)]
162pub struct PlanningDiagnostic {
163 pub code: &'static str,
165 pub severity: PlanningDiagnosticSeverity,
167 pub message: String,
169}
170
171impl PlanningDiagnostic {
172 pub fn info(code: &'static str, message: impl Into<String>) -> Self {
173 Self {
174 code,
175 severity: PlanningDiagnosticSeverity::Info,
176 message: message.into(),
177 }
178 }
179
180 pub fn warning(code: &'static str, message: impl Into<String>) -> Self {
181 Self {
182 code,
183 severity: PlanningDiagnosticSeverity::Warning,
184 message: message.into(),
185 }
186 }
187}
188
189pub fn plan_sql_for_routing<C: Catalog + ?Sized>(
191 catalog: &C,
192 sql: &str,
193) -> Result<Vec<PlannedStatement>, SqlError> {
194 let statements = Parser::parse_sql(&AlopexDialect, sql).map_err(SqlError::from)?;
195 statements
196 .iter()
197 .map(|statement| plan_statement_for_routing(catalog, statement).map_err(SqlError::from))
198 .collect()
199}
200
201pub fn plan_statement_for_routing<C: Catalog + ?Sized>(
203 catalog: &C,
204 statement: &Statement,
205) -> Result<PlannedStatement, PlannerError> {
206 let planner = Planner::new(catalog);
207 let plan = planner.plan(statement)?;
208 let routing_input = routing_input_for_plan(statement, &plan)?;
209 Ok(PlannedStatement {
210 plan,
211 routing_input,
212 })
213}
214
215fn routing_input_for_plan(
216 statement: &Statement,
217 plan: &LogicalPlan,
218) -> Result<RoutingInput, PlannerError> {
219 let mut diagnostics = Vec::new();
220 let extractor = TableReferenceExtractor::new();
221 let table_references = extractor.extract_from_logical_plan(
222 plan,
223 table_reference_access(statement)?,
224 &mut diagnostics,
225 );
226
227 Ok(RoutingInput {
228 statement_kind: statement.kind.clone(),
229 table_references,
230 diagnostics,
231 })
232}
233
234#[derive(Debug, Default, Clone, Copy)]
236pub struct TableReferenceExtractor;
237
238impl TableReferenceExtractor {
239 pub fn new() -> Self {
240 Self
241 }
242
243 pub fn extract_from_logical_plan(
246 &self,
247 plan: &LogicalPlan,
248 root_access: TableReferenceAccess,
249 diagnostics: &mut Vec<PlanningDiagnostic>,
250 ) -> Vec<TableReference> {
251 let mut references = Vec::new();
252 self.extract_plan(
253 plan,
254 root_access,
255 TableReferenceSource::LogicalPlanScan,
256 diagnostics,
257 &mut references,
258 );
259 if references.is_empty() {
260 diagnostics.push(PlanningDiagnostic::info(
261 "ALOPEX-PLAN-ROUTE-001",
262 "statement has no physical table reference",
263 ));
264 }
265 references
266 }
267
268 pub fn extract_from_subquery_context(
270 &self,
271 plan: &LogicalPlan,
272 diagnostics: &mut Vec<PlanningDiagnostic>,
273 ) -> Vec<TableReference> {
274 let mut references = Vec::new();
275 self.extract_plan(
276 plan,
277 TableReferenceAccess::Read,
278 TableReferenceSource::TypedExprSubquery,
279 diagnostics,
280 &mut references,
281 );
282 references
283 }
284
285 fn extract_plan(
286 &self,
287 plan: &LogicalPlan,
288 root_access: TableReferenceAccess,
289 scan_source: TableReferenceSource,
290 diagnostics: &mut Vec<PlanningDiagnostic>,
291 references: &mut Vec<TableReference>,
292 ) {
293 match plan {
294 LogicalPlan::Scan { table, projection } => {
295 if table != LITERAL_TABLE {
296 push_table_reference(
297 references,
298 table,
299 TableReferenceAccess::Read,
300 scan_source,
301 );
302 }
303 self.extract_projection(projection, diagnostics, references);
304 }
305 LogicalPlan::Filter { input, predicate } => {
306 self.extract_plan(input, root_access, scan_source, diagnostics, references);
307 self.extract_typed_expr(predicate, diagnostics, references);
308 }
309 LogicalPlan::Project { input, projection } => {
310 self.extract_plan(input, root_access, scan_source, diagnostics, references);
311 self.extract_projection(projection, diagnostics, references);
312 }
313 LogicalPlan::Join {
314 left,
315 right,
316 condition,
317 ..
318 } => {
319 self.extract_plan(
320 left,
321 TableReferenceAccess::Read,
322 scan_source,
323 diagnostics,
324 references,
325 );
326 self.extract_plan(
327 right,
328 TableReferenceAccess::Read,
329 scan_source,
330 diagnostics,
331 references,
332 );
333 if let Some(condition) = condition {
334 self.extract_typed_expr(condition, diagnostics, references);
335 }
336 }
337 LogicalPlan::Aggregate {
338 input,
339 group_keys,
340 aggregates,
341 having,
342 projection,
343 } => {
344 self.extract_plan(input, root_access, scan_source, diagnostics, references);
345 for expr in group_keys {
346 self.extract_typed_expr(expr, diagnostics, references);
347 }
348 for aggregate in aggregates {
349 if let Some(arg) = &aggregate.arg {
350 self.extract_typed_expr(arg, diagnostics, references);
351 }
352 }
353 if let Some(having) = having {
354 self.extract_typed_expr(having, diagnostics, references);
355 }
356 self.extract_projection(projection, diagnostics, references);
357 }
358 LogicalPlan::Sort { input, order_by } => {
359 self.extract_plan(input, root_access, scan_source, diagnostics, references);
360 for sort_expr in order_by {
361 self.extract_typed_expr(&sort_expr.expr, diagnostics, references);
362 }
363 }
364 LogicalPlan::Limit { input, .. } => {
365 self.extract_plan(input, root_access, scan_source, diagnostics, references);
366 }
367 LogicalPlan::Insert { table, values, .. } => {
368 push_table_reference(
369 references,
370 table,
371 root_access,
372 TableReferenceSource::LogicalPlanMutationTarget,
373 );
374 for row in values {
375 for value in row {
376 self.extract_typed_expr(value, diagnostics, references);
377 }
378 }
379 }
380 LogicalPlan::InsertSelect { table, source, .. } => {
381 push_table_reference(
382 references,
383 table,
384 root_access,
385 TableReferenceSource::LogicalPlanMutationTarget,
386 );
387 self.extract_plan(
388 source,
389 TableReferenceAccess::Read,
390 scan_source,
391 diagnostics,
392 references,
393 );
394 }
395 LogicalPlan::Update {
396 table,
397 assignments,
398 filter,
399 } => {
400 push_table_reference(
401 references,
402 table,
403 root_access,
404 TableReferenceSource::LogicalPlanMutationTarget,
405 );
406 for assignment in assignments {
407 self.extract_typed_expr(&assignment.value, diagnostics, references);
408 }
409 if let Some(filter) = filter {
410 self.extract_typed_expr(filter, diagnostics, references);
411 }
412 }
413 LogicalPlan::Delete { table, filter } => {
414 push_table_reference(
415 references,
416 table,
417 root_access,
418 TableReferenceSource::LogicalPlanMutationTarget,
419 );
420 if let Some(filter) = filter {
421 self.extract_typed_expr(filter, diagnostics, references);
422 }
423 }
424 LogicalPlan::CreateTable { table, .. } => push_table_reference(
425 references,
426 &table.name,
427 root_access,
428 TableReferenceSource::LogicalPlanDdlTarget,
429 ),
430 LogicalPlan::DropTable { name, .. } => push_table_reference(
431 references,
432 name,
433 root_access,
434 TableReferenceSource::LogicalPlanDdlTarget,
435 ),
436 LogicalPlan::CreateIndex { index, .. } => push_table_reference(
437 references,
438 &index.table,
439 root_access,
440 TableReferenceSource::LogicalPlanIndexTarget,
441 ),
442 LogicalPlan::DropIndex { name, .. } => diagnostics.push(PlanningDiagnostic::warning(
443 "ALOPEX-PLAN-ROUTE-003",
444 format!(
445 "DROP INDEX {name} does not expose a target table in the current logical plan"
446 ),
447 )),
448 LogicalPlan::Pragma { .. } => {}
449 }
450 }
451
452 fn extract_projection(
453 &self,
454 projection: &Projection,
455 diagnostics: &mut Vec<PlanningDiagnostic>,
456 references: &mut Vec<TableReference>,
457 ) {
458 if let Projection::Columns(columns) = projection {
459 for column in columns {
460 self.extract_typed_expr(&column.expr, diagnostics, references);
461 }
462 }
463 }
464
465 fn extract_typed_expr(
466 &self,
467 expr: &TypedExpr,
468 diagnostics: &mut Vec<PlanningDiagnostic>,
469 references: &mut Vec<TableReference>,
470 ) {
471 match &expr.kind {
472 TypedExprKind::Literal(_)
473 | TypedExprKind::ColumnRef { .. }
474 | TypedExprKind::VectorLiteral(_) => {}
475 TypedExprKind::BinaryOp { left, right, .. } => {
476 self.extract_typed_expr(left, diagnostics, references);
477 self.extract_typed_expr(right, diagnostics, references);
478 }
479 TypedExprKind::UnaryOp { operand, .. }
480 | TypedExprKind::Cast { expr: operand, .. }
481 | TypedExprKind::IsNull { expr: operand, .. } => {
482 self.extract_typed_expr(operand, diagnostics, references);
483 }
484 TypedExprKind::FunctionCall { args, .. } => {
485 for arg in args {
486 self.extract_typed_expr(arg, diagnostics, references);
487 }
488 }
489 TypedExprKind::Between {
490 expr, low, high, ..
491 } => {
492 self.extract_typed_expr(expr, diagnostics, references);
493 self.extract_typed_expr(low, diagnostics, references);
494 self.extract_typed_expr(high, diagnostics, references);
495 }
496 TypedExprKind::Like {
497 expr,
498 pattern,
499 escape,
500 ..
501 } => {
502 self.extract_typed_expr(expr, diagnostics, references);
503 self.extract_typed_expr(pattern, diagnostics, references);
504 if let Some(escape) = escape {
505 self.extract_typed_expr(escape, diagnostics, references);
506 }
507 }
508 TypedExprKind::InList { expr, list, .. } => {
509 self.extract_typed_expr(expr, diagnostics, references);
510 for item in list {
511 self.extract_typed_expr(item, diagnostics, references);
512 }
513 }
514 TypedExprKind::ScalarSubquery(subquery) => self.extract_plan(
515 subquery,
516 TableReferenceAccess::Read,
517 TableReferenceSource::TypedExprSubquery,
518 diagnostics,
519 references,
520 ),
521 TypedExprKind::InSubquery { expr, subquery, .. } => {
522 self.extract_typed_expr(expr, diagnostics, references);
523 self.extract_plan(
524 subquery,
525 TableReferenceAccess::Read,
526 TableReferenceSource::TypedExprSubquery,
527 diagnostics,
528 references,
529 );
530 }
531 TypedExprKind::Exists { subquery, .. } => self.extract_plan(
532 subquery,
533 TableReferenceAccess::Read,
534 TableReferenceSource::TypedExprSubquery,
535 diagnostics,
536 references,
537 ),
538 TypedExprKind::Quantified { expr, subquery, .. } => {
539 self.extract_typed_expr(expr, diagnostics, references);
540 self.extract_plan(
541 subquery,
542 TableReferenceAccess::Read,
543 TableReferenceSource::TypedExprSubquery,
544 diagnostics,
545 references,
546 );
547 }
548 }
549 }
550}
551
552fn push_table_reference(
553 references: &mut Vec<TableReference>,
554 table_name: &str,
555 access: TableReferenceAccess,
556 source: TableReferenceSource,
557) {
558 if !references.iter().any(|reference| {
559 reference.table_name == table_name
560 && reference.access == access
561 && reference.source == source
562 }) {
563 references.push(TableReference::new(table_name, access, source));
564 }
565}
566
567#[derive(Debug)]
568enum GenericHostStatement<'a> {
569 CreateTable(&'a CreateTable),
570 DropTable(&'a DropTable),
571 CreateIndex(&'a CreateIndex),
572 DropIndex(&'a DropIndex),
573 Pragma {
574 name: &'a str,
575 value: &'a Option<PragmaValue>,
576 },
577 Select(&'a Select),
578 Insert(&'a Insert),
579 Update(&'a Update),
580 Delete(&'a Delete),
581 Unsupported,
582}
583
584fn classify_generic_host_statement(statement_kind: &StatementKind) -> GenericHostStatement<'_> {
585 #[allow(unreachable_patterns)]
588 match statement_kind {
589 StatementKind::CreateTable(statement) => GenericHostStatement::CreateTable(statement),
590 StatementKind::DropTable(statement) => GenericHostStatement::DropTable(statement),
591 StatementKind::CreateIndex(statement) => GenericHostStatement::CreateIndex(statement),
592 StatementKind::DropIndex(statement) => GenericHostStatement::DropIndex(statement),
593 StatementKind::Pragma { name, value } => GenericHostStatement::Pragma { name, value },
594 StatementKind::Select(statement) => GenericHostStatement::Select(statement),
595 StatementKind::Insert(statement) => GenericHostStatement::Insert(statement),
596 StatementKind::Update(statement) => GenericHostStatement::Update(statement),
597 StatementKind::Delete(statement) => GenericHostStatement::Delete(statement),
598 _ => GenericHostStatement::Unsupported,
599 }
600}
601
602fn unsupported_generic_statement(statement: &Statement) -> PlannerError {
603 PlannerError::unsupported_feature(
604 "statement kind for the generic SQL planner",
605 "a statement-specific planner",
606 statement.span,
607 )
608}
609
610fn table_reference_access(statement: &Statement) -> Result<TableReferenceAccess, PlannerError> {
611 table_reference_access_for_classified(
612 statement,
613 classify_generic_host_statement(&statement.kind),
614 )
615}
616
617fn table_reference_access_for_classified(
618 statement: &Statement,
619 classified: GenericHostStatement<'_>,
620) -> Result<TableReferenceAccess, PlannerError> {
621 match classified {
622 GenericHostStatement::Select(_) => Ok(TableReferenceAccess::Read),
623 GenericHostStatement::Insert(_)
624 | GenericHostStatement::Update(_)
625 | GenericHostStatement::Delete(_) => Ok(TableReferenceAccess::Write),
626 GenericHostStatement::CreateTable(_) => Ok(TableReferenceAccess::Create),
627 GenericHostStatement::DropTable(_) => Ok(TableReferenceAccess::Drop),
628 GenericHostStatement::CreateIndex(_)
629 | GenericHostStatement::DropIndex(_)
630 | GenericHostStatement::Pragma { .. } => Ok(TableReferenceAccess::Metadata),
631 GenericHostStatement::Unsupported => Err(unsupported_generic_statement(statement)),
632 }
633}
634
635pub struct Planner<'a, C: Catalog + ?Sized> {
662 catalog: &'a C,
663 name_resolver: NameResolver<'a, C>,
664 type_checker: TypeChecker<'a, C>,
665}
666
667impl<'a, C: Catalog + ?Sized> Planner<'a, C> {
668 pub fn new(catalog: &'a C) -> Self {
670 Self {
671 catalog,
672 name_resolver: NameResolver::new(catalog),
673 type_checker: TypeChecker::new(catalog),
674 }
675 }
676
677 pub fn plan(&self, stmt: &Statement) -> Result<LogicalPlan, PlannerError> {
688 self.plan_classified_statement(stmt, classify_generic_host_statement(&stmt.kind))
689 }
690
691 fn plan_classified_statement(
692 &self,
693 stmt: &Statement,
694 classified: GenericHostStatement<'_>,
695 ) -> Result<LogicalPlan, PlannerError> {
696 match classified {
697 GenericHostStatement::CreateTable(statement) => self.plan_create_table(statement),
699 GenericHostStatement::DropTable(statement) => self.plan_drop_table(statement),
700 GenericHostStatement::CreateIndex(statement) => self.plan_create_index(statement),
701 GenericHostStatement::DropIndex(statement) => self.plan_drop_index(statement),
702 GenericHostStatement::Pragma { name, value } => self.plan_pragma(name, value),
703
704 GenericHostStatement::Select(statement) => self.plan_select(statement),
706 GenericHostStatement::Insert(statement) => self.plan_insert(statement),
707 GenericHostStatement::Update(statement) => self.plan_update(statement),
708 GenericHostStatement::Delete(statement) => self.plan_delete(statement),
709 GenericHostStatement::Unsupported => Err(unsupported_generic_statement(stmt)),
710 }
711 }
712
713 fn plan_pragma(
714 &self,
715 raw_name: &str,
716 value: &Option<PragmaValue>,
717 ) -> Result<LogicalPlan, PlannerError> {
718 let name = raw_name.to_ascii_lowercase();
719 if !matches!(name.as_str(), "cache_size" | "memory_limit" | "io_stats") {
720 return Err(PlannerError::InvalidPragma {
721 name,
722 reason: "supported names are cache_size, memory_limit, and io_stats".to_string(),
723 });
724 }
725 match name.as_str() {
726 "cache_size" => match value {
727 Some(PragmaValue::Int(v)) if *v > 0 => {}
728 Some(PragmaValue::Int(_)) => {
729 return Err(PlannerError::InvalidPragma {
730 name,
731 reason: "cache_size must be a positive page count".to_string(),
732 });
733 }
734 Some(PragmaValue::Text(_)) => {
735 return Err(PlannerError::InvalidPragma {
736 name,
737 reason: "cache_size requires an integer page count".to_string(),
738 });
739 }
740 None => {}
741 },
742 "memory_limit" => {
743 if let Some(PragmaValue::Int(v)) = value
744 && *v < 0
745 {
746 return Err(PlannerError::InvalidPragma {
747 name,
748 reason: "memory_limit cannot be negative".to_string(),
749 });
750 }
751 }
752 "io_stats" => {
753 if value.is_some() {
754 return Err(PlannerError::InvalidPragma {
755 name,
756 reason: "io_stats does not accept a value".to_string(),
757 });
758 }
759 }
760 _ => unreachable!(),
761 }
762 Ok(LogicalPlan::Pragma {
763 name,
764 value: value.clone(),
765 })
766 }
767
768 fn plan_create_table(&self, stmt: &CreateTable) -> Result<LogicalPlan, PlannerError> {
777 if !stmt.if_not_exists && self.catalog.table_exists(&stmt.name) {
779 return Err(PlannerError::table_already_exists(&stmt.name));
780 }
781
782 let columns: Vec<ColumnMetadata> = stmt
784 .columns
785 .iter()
786 .map(|col| self.convert_column_def(col))
787 .collect();
788
789 let primary_key = Self::extract_primary_key(stmt);
791
792 let mut table = TableMetadata::new(stmt.name.clone(), columns);
795 if let Some(pk) = primary_key {
796 table = table.with_primary_key(pk);
797 }
798 table.catalog_name = "default".to_string();
799 table.namespace_name = "default".to_string();
800 table.table_type = TableType::Managed;
801 table.data_source_format = DataSourceFormat::Alopex;
802 table.properties = HashMap::new();
803
804 Ok(LogicalPlan::CreateTable {
805 table,
806 if_not_exists: stmt.if_not_exists,
807 with_options: stmt
808 .with_options
809 .iter()
810 .map(|opt| (opt.key.clone(), opt.value.clone()))
811 .collect(),
812 })
813 }
814
815 fn convert_column_def(&self, col: &ColumnDef) -> ColumnMetadata {
817 let data_type = ResolvedType::from_ast(&col.data_type);
818 let mut meta = ColumnMetadata::new(col.name.clone(), data_type);
819
820 for constraint in &col.constraints {
822 meta = Self::apply_column_constraint(meta, constraint);
823 }
824
825 meta
826 }
827
828 fn apply_column_constraint(
830 mut meta: ColumnMetadata,
831 constraint: &ColumnConstraint,
832 ) -> ColumnMetadata {
833 match constraint {
834 ColumnConstraint::NotNull { .. } => {
835 meta.not_null = true;
836 }
837 ColumnConstraint::PrimaryKey { .. } => {
838 meta.primary_key = true;
839 meta.not_null = true; }
841 ColumnConstraint::Unique { .. } => {
842 meta.unique = true;
843 }
844 ColumnConstraint::Default { value: expr, .. } => {
845 meta.default = Some(expr.clone());
846 }
847 }
848 meta
849 }
850
851 fn extract_primary_key(stmt: &CreateTable) -> Option<Vec<String>> {
853 use crate::ast::ddl::TableConstraint;
854
855 if let Some(TableConstraint::PrimaryKey { columns, .. }) = stmt.constraints.first() {
859 return Some(columns.clone());
860 }
861
862 let pk_columns: Vec<String> = stmt
864 .columns
865 .iter()
866 .filter(|col| col.constraints.iter().any(Self::is_primary_key_constraint))
867 .map(|col| col.name.clone())
868 .collect();
869
870 if pk_columns.is_empty() {
871 None
872 } else {
873 Some(pk_columns)
874 }
875 }
876
877 fn is_primary_key_constraint(constraint: &ColumnConstraint) -> bool {
879 matches!(constraint, ColumnConstraint::PrimaryKey { .. })
880 }
881
882 fn plan_drop_table(&self, stmt: &DropTable) -> Result<LogicalPlan, PlannerError> {
886 if !stmt.if_exists && !self.table_exists_in_default(&stmt.name) {
888 return Err(PlannerError::TableNotFound {
889 name: stmt.name.clone(),
890 line: stmt.span.start.line,
891 column: stmt.span.start.column,
892 });
893 }
894
895 Ok(LogicalPlan::DropTable {
896 name: stmt.name.clone(),
897 if_exists: stmt.if_exists,
898 })
899 }
900
901 fn table_exists_in_default(&self, name: &str) -> bool {
902 match self.catalog.get_table(name) {
903 Some(table) => table.catalog_name == "default" && table.namespace_name == "default",
904 None => false,
905 }
906 }
907
908 fn plan_create_index(&self, stmt: &CreateIndex) -> Result<LogicalPlan, PlannerError> {
915 if !stmt.if_not_exists && self.catalog.index_exists(&stmt.name) {
917 return Err(PlannerError::index_already_exists(&stmt.name));
918 }
919
920 let table = self.name_resolver.resolve_table(&stmt.table, stmt.span)?;
922
923 self.name_resolver
925 .resolve_column(table, &stmt.column, stmt.span)?;
926
927 let mut index = IndexMetadata::new(
931 0,
932 stmt.name.clone(),
933 stmt.table.clone(),
934 vec![stmt.column.clone()],
935 );
936
937 if let Some(method) = stmt.method {
938 index = index.with_method(method);
939 }
940
941 let options: Vec<(String, String)> = stmt
942 .options
943 .iter()
944 .map(|opt| (opt.key.clone(), opt.value.clone()))
945 .collect();
946 if !options.is_empty() {
947 index = index.with_options(options);
948 }
949
950 Ok(LogicalPlan::CreateIndex {
951 index,
952 if_not_exists: stmt.if_not_exists,
953 })
954 }
955
956 fn plan_drop_index(&self, stmt: &DropIndex) -> Result<LogicalPlan, PlannerError> {
960 if !stmt.if_exists && !self.index_exists_in_default(&stmt.name) {
962 return Err(PlannerError::index_not_found(&stmt.name));
963 }
964
965 Ok(LogicalPlan::DropIndex {
966 name: stmt.name.clone(),
967 if_exists: stmt.if_exists,
968 })
969 }
970
971 fn index_exists_in_default(&self, name: &str) -> bool {
972 match self.catalog.get_index(name) {
973 Some(index) => index.catalog_name == "default" && index.namespace_name == "default",
974 None => false,
975 }
976 }
977
978 fn plan_select(&self, stmt: &Select) -> Result<LogicalPlan, PlannerError> {
987 self.plan_select_relation(stmt, &[])
988 .map(|relation| relation.plan)
989 }
990
991 fn plan_select_relation(
992 &self,
993 stmt: &Select,
994 outer_scope: &[ScopedTable],
995 ) -> Result<PlannedRelation, PlannerError> {
996 let mut relation = self.plan_from_items(&stmt.from, stmt.span, outer_scope)?;
997 let expr_scope = relation
998 .scope
999 .iter()
1000 .cloned()
1001 .chain(offset_scope(outer_scope, relation.schema.len()))
1002 .collect::<Vec<_>>();
1003
1004 let has_group_by = stmt
1005 .group_by
1006 .as_ref()
1007 .is_some_and(|items| !items.is_empty());
1008 let has_aggregate = self.select_contains_aggregate(stmt);
1009 let distinct_only =
1010 stmt.distinct && !has_group_by && !has_aggregate && stmt.having.is_none();
1011
1012 let final_projection =
1013 self.build_projection_with_scope(&stmt.projection, &relation.schema, &expr_scope)?;
1014 install_base_projection(&mut relation.plan, &final_projection);
1015 let needs_project_boundary = !matches!(relation.plan, LogicalPlan::Scan { .. });
1016 let mut plan = relation.plan;
1017
1018 if let Some(ref selection) = stmt.selection {
1020 let predicate = self.infer_expr_with_scope(selection, &expr_scope)?;
1021
1022 if predicate.resolved_type != ResolvedType::Boolean {
1024 return Err(PlannerError::type_mismatch(
1025 "Boolean",
1026 predicate.resolved_type.to_string(),
1027 selection.span,
1028 ));
1029 }
1030
1031 plan = LogicalPlan::Filter {
1032 input: Box::new(plan),
1033 predicate,
1034 };
1035 }
1036
1037 if has_group_by || has_aggregate || stmt.having.is_some() || stmt.distinct {
1038 if !has_group_by && !has_aggregate && stmt.having.is_some() {
1039 return Err(PlannerError::invalid_expression(
1040 "HAVING requires GROUP BY or aggregate functions".to_string(),
1041 ));
1042 }
1043
1044 let (group_keys, projected) = if distinct_only {
1045 let projected = self.build_projected_columns_for_distinct_with_scope(
1046 &stmt.projection,
1047 &relation.schema,
1048 &expr_scope,
1049 )?;
1050 let group_keys = projected.iter().map(|col| col.expr.clone()).collect();
1051 (group_keys, projected)
1052 } else {
1053 let group_keys = self.build_group_keys_with_scope(stmt, &expr_scope)?;
1054 let projected = self.build_projected_columns_for_aggregate_with_scope(
1055 &stmt.projection,
1056 &expr_scope,
1057 )?;
1058 (group_keys, projected)
1059 };
1060 let mut aggregates = Vec::new();
1061 let mut agg_map = HashMap::new();
1062
1063 for col in &projected {
1064 self.collect_aggregates_from_typed_expr(&col.expr, &mut aggregates, &mut agg_map)?;
1065 }
1066
1067 let having_typed = if let Some(having) = &stmt.having {
1068 let typed = self.infer_expr_with_scope(having, &expr_scope)?;
1069 if typed.resolved_type != ResolvedType::Boolean {
1070 return Err(PlannerError::type_mismatch(
1071 "Boolean",
1072 typed.resolved_type.type_name().to_string(),
1073 typed.span,
1074 ));
1075 }
1076 self.collect_aggregates_from_typed_expr(&typed, &mut aggregates, &mut agg_map)?;
1077 Some(typed)
1078 } else {
1079 None
1080 };
1081
1082 let mut order_by = Vec::new();
1083 if !stmt.order_by.is_empty() {
1084 for order_expr in &stmt.order_by {
1085 let typed = self.infer_expr_with_scope(&order_expr.expr, &expr_scope)?;
1086 self.collect_aggregates_from_typed_expr(&typed, &mut aggregates, &mut agg_map)?;
1087 let asc = order_expr.asc.unwrap_or(true);
1088 let nulls_first = order_expr.nulls_first.unwrap_or(false);
1089 order_by.push(SortExpr::new(typed, asc, nulls_first));
1090 }
1091 }
1092
1093 if let Some(ref having) = having_typed {
1094 self.type_checker
1095 .validate_having_expr(having, &group_keys, &aggregates)?;
1096 }
1097
1098 let output_schema = build_aggregate_schema(&group_keys, &aggregates);
1099 let output_names: Vec<String> = output_schema.iter().map(|c| c.name.clone()).collect();
1100
1101 let projection = self.build_aggregate_projection(
1102 projected,
1103 &group_keys,
1104 &aggregates,
1105 &output_names,
1106 )?;
1107
1108 let having = if let Some(having) = having_typed {
1109 Some(self.rewrite_expr_for_aggregate(
1110 &having,
1111 &group_keys,
1112 &aggregates,
1113 &output_names,
1114 )?)
1115 } else {
1116 None
1117 };
1118
1119 let order_by = order_by
1120 .into_iter()
1121 .map(|expr| {
1122 let rewritten = self.rewrite_expr_for_aggregate(
1123 &expr.expr,
1124 &group_keys,
1125 &aggregates,
1126 &output_names,
1127 )?;
1128 Ok(SortExpr::new(rewritten, expr.asc, expr.nulls_first))
1129 })
1130 .collect::<Result<Vec<_>, PlannerError>>()?;
1131
1132 let schema = projection_schema(&projection, &output_schema);
1133 plan = LogicalPlan::Aggregate {
1134 input: Box::new(plan),
1135 group_keys,
1136 aggregates,
1137 having,
1138 projection,
1139 };
1140
1141 if !order_by.is_empty() {
1142 plan = LogicalPlan::Sort {
1143 input: Box::new(plan),
1144 order_by,
1145 };
1146 }
1147
1148 if stmt.limit.is_some() || stmt.offset.is_some() {
1149 let limit = self.extract_limit_value(&stmt.limit, stmt.span)?;
1150 let offset = self.extract_limit_value(&stmt.offset, stmt.span)?;
1151 plan = LogicalPlan::Limit {
1152 input: Box::new(plan),
1153 limit,
1154 offset,
1155 };
1156 }
1157
1158 return Ok(PlannedRelation {
1159 plan,
1160 schema: schema.clone(),
1161 scope: vec![ScopedTable::new(
1162 TableMetadata::new(LITERAL_TABLE, schema),
1163 0,
1164 )],
1165 });
1166 }
1167
1168 if !stmt.order_by.is_empty() {
1170 let order_by = self.build_sort_exprs_with_scope(&stmt.order_by, &expr_scope)?;
1171 plan = LogicalPlan::Sort {
1172 input: Box::new(plan),
1173 order_by,
1174 };
1175 }
1176
1177 if stmt.limit.is_some() || stmt.offset.is_some() {
1178 let limit = self.extract_limit_value(&stmt.limit, stmt.span)?;
1179 let offset = self.extract_limit_value(&stmt.offset, stmt.span)?;
1180 plan = LogicalPlan::Limit {
1181 input: Box::new(plan),
1182 limit,
1183 offset,
1184 };
1185 }
1186
1187 let output_schema = projection_schema(&final_projection, &relation.schema);
1188 if needs_project_boundary {
1189 plan = LogicalPlan::Project {
1190 input: Box::new(plan),
1191 projection: final_projection,
1192 };
1193 }
1194 Ok(PlannedRelation {
1195 plan,
1196 schema: output_schema.clone(),
1197 scope: vec![ScopedTable::new(
1198 TableMetadata::new(LITERAL_TABLE, output_schema),
1199 0,
1200 )],
1201 })
1202 }
1203
1204 fn plan_from_items(
1208 &self,
1209 items: &[FromItem],
1210 select_span: crate::ast::Span,
1211 outer_scope: &[ScopedTable],
1212 ) -> Result<PlannedRelation, PlannerError> {
1213 match items {
1214 [] => {
1215 let schema = Vec::new();
1216 Ok(PlannedRelation {
1217 plan: LogicalPlan::Scan {
1218 table: LITERAL_TABLE.to_string(),
1219 projection: Projection::All(Vec::new()),
1220 },
1221 schema: schema.clone(),
1222 scope: vec![ScopedTable::new(
1223 TableMetadata::new(LITERAL_TABLE, schema),
1224 0,
1225 )],
1226 })
1227 }
1228 [single] => self.plan_from_item(single, 0, outer_scope),
1229 [first, rest @ ..] => {
1230 let mut relation = self.plan_from_item(first, 0, outer_scope)?;
1231 for item in rest {
1232 let right = self.plan_from_item(item, relation.schema.len(), outer_scope)?;
1233 relation = self.combine_join_relation(
1234 relation,
1235 right,
1236 JoinType::Cross,
1237 None,
1238 None,
1239 select_span,
1240 )?;
1241 }
1242 Ok(relation)
1243 }
1244 }
1245 }
1246
1247 fn plan_from_item(
1248 &self,
1249 item: &FromItem,
1250 start_index: usize,
1251 outer_scope: &[ScopedTable],
1252 ) -> Result<PlannedRelation, PlannerError> {
1253 match item {
1254 FromItem::Table { name, alias, span } => {
1255 let table = self.name_resolver.resolve_table(name, *span)?.clone();
1256 let mut scope_table = table.clone();
1257 if let Some(alias) = alias {
1258 scope_table.name = alias.clone();
1259 }
1260 let schema = table.columns.clone();
1261 Ok(PlannedRelation {
1262 plan: LogicalPlan::Scan {
1263 table: name.clone(),
1264 projection: Projection::All(
1265 schema.iter().map(|col| col.name.clone()).collect(),
1266 ),
1267 },
1268 schema,
1269 scope: vec![ScopedTable::new(scope_table, start_index)],
1270 })
1271 }
1272 FromItem::Join {
1273 left,
1274 right,
1275 join_type,
1276 condition,
1277 using,
1278 natural,
1279 span,
1280 } => {
1281 let left_relation = self.plan_from_item(left, start_index, outer_scope)?;
1282 let right_relation = self.plan_from_item(
1283 right,
1284 start_index + left_relation.schema.len(),
1285 outer_scope,
1286 )?;
1287 let expr_scope = left_relation
1288 .scope
1289 .iter()
1290 .cloned()
1291 .chain(right_relation.scope.iter().cloned())
1292 .chain(offset_scope(
1293 outer_scope,
1294 left_relation.schema.len() + right_relation.schema.len(),
1295 ))
1296 .collect::<Vec<_>>();
1297 let using = if *natural {
1298 Some(natural_join_columns(
1299 &left_relation.schema,
1300 &right_relation.schema,
1301 ))
1302 } else {
1303 using.clone()
1304 };
1305 let typed_condition = if let Some(expr) = condition {
1306 let typed = self.infer_expr_with_scope(expr, &expr_scope)?;
1307 if typed.resolved_type != ResolvedType::Boolean {
1308 return Err(PlannerError::type_mismatch(
1309 "Boolean",
1310 typed.resolved_type.to_string(),
1311 expr.span,
1312 ));
1313 }
1314 Some(typed)
1315 } else {
1316 self.build_using_condition(
1317 using.as_deref(),
1318 &left_relation,
1319 &right_relation,
1320 *span,
1321 )?
1322 };
1323 self.combine_join_relation(
1324 left_relation,
1325 right_relation,
1326 map_join_type(*join_type),
1327 typed_condition,
1328 using,
1329 *span,
1330 )
1331 }
1332 FromItem::Derived {
1333 subquery,
1334 alias,
1335 span,
1336 } => {
1337 let crate::ast::StatementKind::Select(select) = &subquery.kind else {
1338 return Err(PlannerError::unsupported_feature(
1339 "non-SELECT derived table",
1340 "v0.6.0-subquery Phase 6",
1341 *span,
1342 ));
1343 };
1344 let mut relation = self.plan_select_relation(select, &[])?;
1351 let alias = alias.clone().ok_or_else(|| {
1352 PlannerError::invalid_expression("derived table requires an alias".to_string())
1353 })?;
1354 relation.plan = LogicalPlan::Project {
1355 input: Box::new(relation.plan),
1356 projection: Projection::All(
1357 relation.schema.iter().map(|col| col.name.clone()).collect(),
1358 ),
1359 };
1360 relation.scope = vec![ScopedTable::new(
1361 TableMetadata::new(alias, relation.schema.clone()),
1362 start_index,
1363 )];
1364 Ok(relation)
1365 }
1366 }
1367 }
1368
1369 fn combine_join_relation(
1370 &self,
1371 left: PlannedRelation,
1372 right: PlannedRelation,
1373 join_type: JoinType,
1374 condition: Option<TypedExpr>,
1375 using: Option<Vec<String>>,
1376 _span: crate::ast::Span,
1377 ) -> Result<PlannedRelation, PlannerError> {
1378 let mut schema = left.schema.clone();
1379 schema.extend(right.schema.clone());
1380 let mut scope = left.scope.clone();
1381 let mut right_scope = right.scope.clone();
1382 if let Some(columns) = &using {
1383 for column in columns {
1387 let right_index = right_scope.iter().find_map(|table| {
1388 table
1389 .table
1390 .get_column_index(column)
1391 .map(|index| table.start_index + index)
1392 });
1393 let Some(right_index) = right_index else {
1394 continue;
1395 };
1396 for table in &mut scope {
1397 if table.table.get_column_index(column).is_some() {
1398 table.merge_column_with(column, right_index);
1399 }
1400 }
1401 }
1402 for table in &mut right_scope {
1403 table.hide_unqualified_columns(columns);
1404 }
1405 }
1406 scope.extend(right_scope);
1407 Ok(PlannedRelation {
1408 plan: LogicalPlan::Join {
1409 left: Box::new(left.plan),
1410 right: Box::new(right.plan),
1411 join_type,
1412 condition,
1413 using,
1414 },
1415 schema,
1416 scope,
1417 })
1418 }
1419
1420 fn build_using_condition(
1421 &self,
1422 using: Option<&[String]>,
1423 left: &PlannedRelation,
1424 right: &PlannedRelation,
1425 span: crate::ast::Span,
1426 ) -> Result<Option<TypedExpr>, PlannerError> {
1427 let Some(columns) = using else {
1428 return Ok(None);
1429 };
1430 let mut condition = None;
1431 for column in columns {
1432 let left_col = find_scoped_column(&left.scope, column, span)?;
1433 let right_col = find_scoped_column(&right.scope, column, span)?;
1434 let left_expr = merged_scoped_column_expr(&left_col, column, span);
1435 let right_expr = merged_scoped_column_expr(&right_col, column, span);
1436 self.type_checker
1437 .check_comparison_op(&left_col.ty, &right_col.ty, span)?;
1438 let eq = TypedExpr::binary_op(
1439 left_expr,
1440 crate::ast::expr::BinaryOp::Eq,
1441 right_expr,
1442 ResolvedType::Boolean,
1443 span,
1444 );
1445 condition = Some(match condition {
1446 Some(prev) => TypedExpr::binary_op(
1447 prev,
1448 crate::ast::expr::BinaryOp::And,
1449 eq,
1450 ResolvedType::Boolean,
1451 span,
1452 ),
1453 None => eq,
1454 });
1455 }
1456 Ok(condition)
1457 }
1458
1459 fn infer_expr_with_scope(
1460 &self,
1461 expr: &crate::ast::expr::Expr,
1462 scope: &[ScopedTable],
1463 ) -> Result<TypedExpr, PlannerError> {
1464 self.type_checker
1465 .infer_type_with_scope(expr, scope, &|stmt, outer_scope| {
1466 let crate::ast::StatementKind::Select(select) = &stmt.kind else {
1467 return Err(PlannerError::unsupported_feature(
1468 "non-SELECT subquery",
1469 "v0.6.0-subquery Phase 6",
1470 stmt.span(),
1471 ));
1472 };
1473 let relation = self.plan_select_relation(select, outer_scope)?;
1474 Ok((relation.plan, relation.schema))
1475 })
1476 }
1477
1478 #[allow(dead_code)]
1479 fn build_projection(
1480 &self,
1481 items: &[SelectItem],
1482 table: &TableMetadata,
1483 ) -> Result<Projection, PlannerError> {
1484 if items.len() == 1 && matches!(&items[0], SelectItem::Wildcard { .. }) {
1486 let columns = self.name_resolver.expand_wildcard(table);
1487 return Ok(Projection::All(columns));
1488 }
1489
1490 let mut projected_columns = Vec::new();
1492 for item in items {
1493 match item {
1494 SelectItem::Wildcard { span } => {
1495 for col in &table.columns {
1497 let column_index = table.get_column_index(&col.name).unwrap();
1498 let typed_expr = TypedExpr::column_ref(
1499 table.name.clone(),
1500 col.name.clone(),
1501 column_index,
1502 col.data_type.clone(),
1503 *span,
1504 );
1505 projected_columns.push(ProjectedColumn::new(typed_expr));
1506 }
1507 }
1508 SelectItem::QualifiedWildcard {
1509 table: qualifier,
1510 span,
1511 } => {
1512 if qualifier != &table.name {
1513 return Err(PlannerError::invalid_expression(format!(
1514 "table '{qualifier}' is not available for wildcard projection"
1515 )));
1516 }
1517 for col in &table.columns {
1518 let column_index = table.get_column_index(&col.name).unwrap();
1519 projected_columns.push(ProjectedColumn::new(TypedExpr::column_ref(
1520 table.name.clone(),
1521 col.name.clone(),
1522 column_index,
1523 col.data_type.clone(),
1524 *span,
1525 )));
1526 }
1527 }
1528 SelectItem::Expr { expr, alias, .. } => {
1529 let typed_expr = self.type_checker.infer_type(expr, table)?;
1530 let projected = if let Some(alias) = alias {
1531 ProjectedColumn::with_alias(typed_expr, alias.clone())
1532 } else {
1533 ProjectedColumn::new(typed_expr)
1534 };
1535 projected_columns.push(projected);
1536 }
1537 }
1538 }
1539
1540 Ok(Projection::Columns(projected_columns))
1541 }
1542
1543 fn build_projection_with_scope(
1544 &self,
1545 items: &[SelectItem],
1546 schema: &[ColumnMetadata],
1547 scope: &[ScopedTable],
1548 ) -> Result<Projection, PlannerError> {
1549 if items.len() == 1 && matches!(&items[0], SelectItem::Wildcard { .. }) {
1550 return Ok(Projection::All(visible_wildcard_columns(schema, scope)));
1551 }
1552
1553 let mut projected_columns = Vec::new();
1554 for item in items {
1555 match item {
1556 SelectItem::Wildcard { span } => {
1557 for scoped in scope {
1558 for (local_idx, col) in scoped.table.columns.iter().enumerate() {
1559 projected_columns.push(ProjectedColumn::new(TypedExpr::column_ref(
1560 scoped.table.name.clone(),
1561 col.name.clone(),
1562 scoped.start_index + local_idx,
1563 col.data_type.clone(),
1564 *span,
1565 )));
1566 }
1567 }
1568 }
1569 SelectItem::QualifiedWildcard { table, span } => {
1570 let scoped = scope
1571 .iter()
1572 .filter(|scoped| scoped.table.name == *table)
1573 .collect::<Vec<_>>();
1574 match scoped.as_slice() {
1575 [] => {
1576 return Err(PlannerError::invalid_expression(format!(
1577 "table '{table}' is not available for wildcard projection"
1578 )));
1579 }
1580 [scoped] => {
1581 for (local_idx, col) in scoped.table.columns.iter().enumerate() {
1582 projected_columns.push(ProjectedColumn::new(
1583 TypedExpr::column_ref(
1584 scoped.table.name.clone(),
1585 col.name.clone(),
1586 scoped.start_index + local_idx,
1587 col.data_type.clone(),
1588 *span,
1589 ),
1590 ));
1591 }
1592 }
1593 _ => {
1594 return Err(PlannerError::ambiguous_column(
1595 table,
1596 scoped
1597 .iter()
1598 .map(|scoped| scoped.table.name.clone())
1599 .collect(),
1600 *span,
1601 ));
1602 }
1603 }
1604 }
1605 SelectItem::Expr { expr, alias, .. } => {
1606 let typed_expr = self.infer_expr_with_scope(expr, scope)?;
1607 let projected = if let Some(alias) = alias {
1608 ProjectedColumn::with_alias(typed_expr, alias.clone())
1609 } else {
1610 ProjectedColumn::new(typed_expr)
1611 };
1612 projected_columns.push(projected);
1613 }
1614 }
1615 }
1616
1617 Ok(Projection::Columns(projected_columns))
1618 }
1619
1620 #[allow(dead_code)]
1622 fn build_sort_exprs(
1623 &self,
1624 order_by: &[OrderByExpr],
1625 table: &TableMetadata,
1626 ) -> Result<Vec<SortExpr>, PlannerError> {
1627 let mut sort_exprs = Vec::new();
1628
1629 for order_expr in order_by {
1630 let typed_expr = self.type_checker.infer_type(&order_expr.expr, table)?;
1631
1632 let asc = order_expr.asc.unwrap_or(true);
1634
1635 let nulls_first = order_expr.nulls_first.unwrap_or(false);
1637
1638 sort_exprs.push(SortExpr::new(typed_expr, asc, nulls_first));
1639 }
1640
1641 Ok(sort_exprs)
1642 }
1643
1644 fn build_sort_exprs_with_scope(
1645 &self,
1646 order_by: &[OrderByExpr],
1647 scope: &[ScopedTable],
1648 ) -> Result<Vec<SortExpr>, PlannerError> {
1649 let mut sort_exprs = Vec::new();
1650 for order_expr in order_by {
1651 let typed_expr = self.infer_expr_with_scope(&order_expr.expr, scope)?;
1652 let asc = order_expr.asc.unwrap_or(true);
1653 let nulls_first = order_expr.nulls_first.unwrap_or(false);
1654 sort_exprs.push(SortExpr::new(typed_expr, asc, nulls_first));
1655 }
1656 Ok(sort_exprs)
1657 }
1658
1659 fn select_contains_aggregate(&self, stmt: &Select) -> bool {
1660 stmt.projection.iter().any(|item| match item {
1661 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => false,
1662 SelectItem::Expr { expr, .. } => expr_contains_aggregate(expr),
1663 }) || stmt
1664 .group_by
1665 .as_ref()
1666 .map(|items| items.iter().any(expr_contains_aggregate))
1667 .unwrap_or(false)
1668 || stmt
1669 .having
1670 .as_ref()
1671 .map(expr_contains_aggregate)
1672 .unwrap_or(false)
1673 || stmt
1674 .order_by
1675 .iter()
1676 .any(|order| expr_contains_aggregate(&order.expr))
1677 }
1678
1679 #[allow(dead_code)]
1680 fn build_group_keys(
1681 &self,
1682 stmt: &Select,
1683 table: &TableMetadata,
1684 ) -> Result<Vec<TypedExpr>, PlannerError> {
1685 let mut keys = Vec::new();
1686 if let Some(items) = &stmt.group_by {
1687 for expr in items {
1688 let typed = self.type_checker.infer_type(expr, table)?;
1689 if typed_expr_contains_aggregate(&typed) {
1690 return Err(PlannerError::invalid_expression(
1691 "GROUP BY cannot contain aggregate functions".to_string(),
1692 ));
1693 }
1694 if !matches!(typed.kind, TypedExprKind::ColumnRef { .. }) {
1695 return Err(PlannerError::invalid_expression(
1696 "GROUP BY expressions must be column references".to_string(),
1697 ));
1698 }
1699 keys.push(typed);
1700 }
1701 }
1702 Ok(keys)
1703 }
1704
1705 fn build_group_keys_with_scope(
1706 &self,
1707 stmt: &Select,
1708 scope: &[ScopedTable],
1709 ) -> Result<Vec<TypedExpr>, PlannerError> {
1710 let mut keys = Vec::new();
1711 if let Some(items) = &stmt.group_by {
1712 for expr in items {
1713 let typed = self.infer_expr_with_scope(expr, scope)?;
1714 if typed_expr_contains_aggregate(&typed) {
1715 return Err(PlannerError::invalid_expression(
1716 "GROUP BY cannot contain aggregate functions".to_string(),
1717 ));
1718 }
1719 if !matches!(typed.kind, TypedExprKind::ColumnRef { .. }) {
1720 return Err(PlannerError::invalid_expression(
1721 "GROUP BY expressions must be column references".to_string(),
1722 ));
1723 }
1724 keys.push(typed);
1725 }
1726 }
1727 Ok(keys)
1728 }
1729
1730 #[allow(dead_code)]
1731 fn build_projected_columns_for_aggregate(
1732 &self,
1733 items: &[SelectItem],
1734 table: &TableMetadata,
1735 ) -> Result<Vec<ProjectedColumn>, PlannerError> {
1736 let mut projected = Vec::new();
1737 for item in items {
1738 match item {
1739 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => {
1740 return Err(PlannerError::invalid_expression(
1741 "wildcard projection not supported with GROUP BY/aggregate".to_string(),
1742 ));
1743 }
1744 SelectItem::Expr { expr, alias, .. } => {
1745 let typed = self.type_checker.infer_type(expr, table)?;
1746 projected.push(ProjectedColumn {
1747 expr: typed,
1748 alias: alias.clone(),
1749 });
1750 }
1751 }
1752 }
1753 Ok(projected)
1754 }
1755
1756 fn build_projected_columns_for_aggregate_with_scope(
1757 &self,
1758 items: &[SelectItem],
1759 scope: &[ScopedTable],
1760 ) -> Result<Vec<ProjectedColumn>, PlannerError> {
1761 let mut projected = Vec::new();
1762 for item in items {
1763 match item {
1764 SelectItem::Wildcard { .. } | SelectItem::QualifiedWildcard { .. } => {
1765 return Err(PlannerError::invalid_expression(
1766 "wildcard projection not supported with GROUP BY/aggregate".to_string(),
1767 ));
1768 }
1769 SelectItem::Expr { expr, alias, .. } => {
1770 let typed = self.infer_expr_with_scope(expr, scope)?;
1771 projected.push(ProjectedColumn {
1772 expr: typed,
1773 alias: alias.clone(),
1774 });
1775 }
1776 }
1777 }
1778 Ok(projected)
1779 }
1780
1781 #[allow(dead_code)]
1782 fn build_projected_columns_for_distinct(
1783 &self,
1784 items: &[SelectItem],
1785 table: &TableMetadata,
1786 ) -> Result<Vec<ProjectedColumn>, PlannerError> {
1787 let projection = self.build_projection(items, table)?;
1788 match projection {
1789 Projection::All(columns) => {
1790 let mut projected = Vec::with_capacity(columns.len());
1791 for column in columns {
1792 let column_index = table.get_column_index(&column).ok_or_else(|| {
1793 PlannerError::invalid_expression(format!(
1794 "column '{column}' not found for DISTINCT projection"
1795 ))
1796 })?;
1797 let column_meta = table.get_column(&column).ok_or_else(|| {
1798 PlannerError::invalid_expression(format!(
1799 "column '{column}' not found for DISTINCT projection"
1800 ))
1801 })?;
1802 let typed_expr = TypedExpr::column_ref(
1803 table.name.clone(),
1804 column.clone(),
1805 column_index,
1806 column_meta.data_type.clone(),
1807 crate::ast::Span::default(),
1808 );
1809 projected.push(ProjectedColumn::new(typed_expr));
1810 }
1811 Ok(projected)
1812 }
1813 Projection::Columns(columns) => Ok(columns),
1814 }
1815 }
1816
1817 fn build_projected_columns_for_distinct_with_scope(
1818 &self,
1819 items: &[SelectItem],
1820 schema: &[ColumnMetadata],
1821 scope: &[ScopedTable],
1822 ) -> Result<Vec<ProjectedColumn>, PlannerError> {
1823 let projection = self.build_projection_with_scope(items, schema, scope)?;
1824 match projection {
1825 Projection::All(columns) => {
1826 let mut projected = Vec::with_capacity(columns.len());
1827 for (idx, column) in columns.into_iter().enumerate() {
1828 let column_meta = schema.get(idx).ok_or_else(|| {
1829 PlannerError::invalid_expression(format!(
1830 "column '{column}' not found for DISTINCT projection"
1831 ))
1832 })?;
1833 projected.push(ProjectedColumn::new(TypedExpr::column_ref(
1834 LITERAL_TABLE.to_string(),
1835 column,
1836 idx,
1837 column_meta.data_type.clone(),
1838 crate::ast::Span::default(),
1839 )));
1840 }
1841 Ok(projected)
1842 }
1843 Projection::Columns(columns) => Ok(columns),
1844 }
1845 }
1846
1847 fn collect_aggregates_from_typed_expr(
1848 &self,
1849 expr: &TypedExpr,
1850 aggregates: &mut Vec<AggregateExpr>,
1851 aggregate_map: &mut HashMap<AggregateSignature, usize>,
1852 ) -> Result<(), PlannerError> {
1853 match &expr.kind {
1854 TypedExprKind::FunctionCall {
1855 name,
1856 args,
1857 distinct,
1858 star,
1859 } if is_aggregate_function(name) => {
1860 for arg in args {
1861 if typed_expr_contains_aggregate(arg) {
1862 return Err(PlannerError::invalid_expression(
1863 "nested aggregate functions are not supported".to_string(),
1864 ));
1865 }
1866 }
1867 let (agg, signature) =
1868 self.build_aggregate_expr_from_typed(expr, name, args, *distinct, *star)?;
1869 aggregate_map.entry(signature).or_insert_with(|| {
1870 aggregates.push(agg);
1871 aggregates.len() - 1
1872 });
1873 Ok(())
1874 }
1875 TypedExprKind::BinaryOp { left, right, .. } => {
1876 self.collect_aggregates_from_typed_expr(left, aggregates, aggregate_map)?;
1877 self.collect_aggregates_from_typed_expr(right, aggregates, aggregate_map)?;
1878 Ok(())
1879 }
1880 TypedExprKind::UnaryOp { operand, .. } => {
1881 self.collect_aggregates_from_typed_expr(operand, aggregates, aggregate_map)
1882 }
1883 TypedExprKind::FunctionCall { args, .. } => {
1884 for arg in args {
1885 self.collect_aggregates_from_typed_expr(arg, aggregates, aggregate_map)?;
1886 }
1887 Ok(())
1888 }
1889 TypedExprKind::Between {
1890 expr, low, high, ..
1891 } => {
1892 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)?;
1893 self.collect_aggregates_from_typed_expr(low, aggregates, aggregate_map)?;
1894 self.collect_aggregates_from_typed_expr(high, aggregates, aggregate_map)?;
1895 Ok(())
1896 }
1897 TypedExprKind::Like {
1898 expr,
1899 pattern,
1900 escape,
1901 ..
1902 } => {
1903 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)?;
1904 self.collect_aggregates_from_typed_expr(pattern, aggregates, aggregate_map)?;
1905 if let Some(esc) = escape {
1906 self.collect_aggregates_from_typed_expr(esc, aggregates, aggregate_map)?;
1907 }
1908 Ok(())
1909 }
1910 TypedExprKind::InList { expr, list, .. } => {
1911 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)?;
1912 for item in list {
1913 self.collect_aggregates_from_typed_expr(item, aggregates, aggregate_map)?;
1914 }
1915 Ok(())
1916 }
1917 TypedExprKind::IsNull { expr, .. } => {
1918 self.collect_aggregates_from_typed_expr(expr, aggregates, aggregate_map)
1919 }
1920 _ => Ok(()),
1921 }
1922 }
1923
1924 fn build_aggregate_expr_from_typed(
1925 &self,
1926 expr: &TypedExpr,
1927 name: &str,
1928 args: &[TypedExpr],
1929 distinct: bool,
1930 star: bool,
1931 ) -> Result<(AggregateExpr, AggregateSignature), PlannerError> {
1932 let lower = name.to_lowercase();
1933 match lower.as_str() {
1934 "count" => {
1935 if star {
1936 let agg = AggregateExpr::count_star();
1937 let signature = aggregate_signature(name, distinct, star, None, None, expr);
1938 return Ok((agg, signature));
1939 }
1940 if args.len() != 1 {
1941 return Err(PlannerError::type_mismatch(
1942 "1 argument",
1943 format!("{} arguments", args.len()),
1944 expr.span,
1945 ));
1946 }
1947 let agg = AggregateExpr {
1948 function: AggregateFunction::Count,
1949 arg: Some(args[0].clone()),
1950 distinct,
1951 result_type: ResolvedType::BigInt,
1952 };
1953 let signature =
1954 aggregate_signature(name, distinct, star, Some(&args[0]), None, expr);
1955 Ok((agg, signature))
1956 }
1957 "sum" => {
1958 let arg = self.require_single_aggregate_arg(args, expr.span)?;
1959 let agg = AggregateExpr {
1960 function: AggregateFunction::Sum,
1961 arg: Some(arg.clone()),
1962 distinct,
1963 result_type: crate::planner::aggregate_expr::sum_result_type(
1964 &arg.resolved_type,
1965 ),
1966 };
1967 let signature = aggregate_signature(name, distinct, star, Some(arg), None, expr);
1968 Ok((agg, signature))
1969 }
1970 "total" => {
1971 let arg = self.require_single_aggregate_arg(args, expr.span)?;
1972 let agg = AggregateExpr {
1973 function: AggregateFunction::Total,
1974 arg: Some(arg.clone()),
1975 distinct: false,
1976 result_type: ResolvedType::Double,
1977 };
1978 let signature = aggregate_signature(name, false, star, Some(arg), None, expr);
1979 Ok((agg, signature))
1980 }
1981 "avg" => {
1982 let arg = self.require_single_aggregate_arg(args, expr.span)?;
1983 let agg = AggregateExpr {
1984 function: AggregateFunction::Avg,
1985 arg: Some(arg.clone()),
1986 distinct,
1987 result_type: ResolvedType::Double,
1988 };
1989 let signature = aggregate_signature(name, distinct, star, Some(arg), None, expr);
1990 Ok((agg, signature))
1991 }
1992 "min" => {
1993 let arg = self.require_single_aggregate_arg(args, expr.span)?;
1994 let agg = AggregateExpr {
1995 function: AggregateFunction::Min,
1996 arg: Some(arg.clone()),
1997 distinct,
1998 result_type: arg.resolved_type.clone(),
1999 };
2000 let signature = aggregate_signature(name, distinct, star, Some(arg), None, expr);
2001 Ok((agg, signature))
2002 }
2003 "max" => {
2004 let arg = self.require_single_aggregate_arg(args, expr.span)?;
2005 let agg = AggregateExpr {
2006 function: AggregateFunction::Max,
2007 arg: Some(arg.clone()),
2008 distinct,
2009 result_type: arg.resolved_type.clone(),
2010 };
2011 let signature = aggregate_signature(name, distinct, star, Some(arg), None, expr);
2012 Ok((agg, signature))
2013 }
2014 "group_concat" => {
2015 if args.is_empty() || args.len() > 2 {
2016 return Err(PlannerError::type_mismatch(
2017 "1 or 2 arguments",
2018 format!("{} arguments", args.len()),
2019 expr.span,
2020 ));
2021 }
2022 let arg = &args[0];
2023 let mut separator = None;
2024 if args.len() == 2 {
2025 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
2026 separator = Some(value.clone());
2027 } else {
2028 return Err(PlannerError::invalid_expression(
2029 "GROUP_CONCAT separator must be a string literal".to_string(),
2030 ));
2031 }
2032 }
2033 let agg = AggregateExpr {
2034 function: AggregateFunction::GroupConcat { separator },
2035 arg: Some(arg.clone()),
2036 distinct,
2037 result_type: ResolvedType::Text,
2038 };
2039 let signature = aggregate_signature(
2040 name,
2041 distinct,
2042 star,
2043 Some(arg),
2044 match &agg.function {
2045 AggregateFunction::GroupConcat { separator } => separator.as_ref(),
2046 _ => None,
2047 },
2048 expr,
2049 );
2050 Ok((agg, signature))
2051 }
2052 "string_agg" => {
2053 if args.len() != 2 {
2054 return Err(PlannerError::type_mismatch(
2055 "2 arguments",
2056 format!("{} arguments", args.len()),
2057 expr.span,
2058 ));
2059 }
2060 let arg = &args[0];
2061 let separator =
2062 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
2063 Some(value.clone())
2064 } else {
2065 return Err(PlannerError::invalid_expression(
2066 "STRING_AGG separator must be a string literal".to_string(),
2067 ));
2068 };
2069 let agg = AggregateExpr {
2070 function: AggregateFunction::StringAgg { separator },
2071 arg: Some(arg.clone()),
2072 distinct,
2073 result_type: ResolvedType::Text,
2074 };
2075 let signature = aggregate_signature(
2076 name,
2077 distinct,
2078 star,
2079 Some(arg),
2080 match &agg.function {
2081 AggregateFunction::StringAgg { separator } => separator.as_ref(),
2082 _ => None,
2083 },
2084 expr,
2085 );
2086 Ok((agg, signature))
2087 }
2088 _ => Err(PlannerError::unsupported_feature(
2089 format!("function '{}'", name),
2090 "future",
2091 expr.span,
2092 )),
2093 }
2094 }
2095
2096 fn require_single_aggregate_arg<'b>(
2097 &self,
2098 args: &'b [TypedExpr],
2099 span: crate::ast::Span,
2100 ) -> Result<&'b TypedExpr, PlannerError> {
2101 if args.len() != 1 {
2102 return Err(PlannerError::type_mismatch(
2103 "1 argument",
2104 format!("{} arguments", args.len()),
2105 span,
2106 ));
2107 }
2108 Ok(&args[0])
2109 }
2110
2111 fn build_aggregate_projection(
2112 &self,
2113 projected: Vec<ProjectedColumn>,
2114 group_keys: &[TypedExpr],
2115 aggregates: &[AggregateExpr],
2116 output_names: &[String],
2117 ) -> Result<Projection, PlannerError> {
2118 let mut columns = Vec::new();
2119 for col in projected {
2120 let rewritten =
2121 self.rewrite_expr_for_aggregate(&col.expr, group_keys, aggregates, output_names)?;
2122 columns.push(ProjectedColumn {
2123 expr: rewritten,
2124 alias: col.alias,
2125 });
2126 }
2127 Ok(Projection::Columns(columns))
2128 }
2129
2130 fn rewrite_expr_for_aggregate(
2131 &self,
2132 expr: &TypedExpr,
2133 group_keys: &[TypedExpr],
2134 aggregates: &[AggregateExpr],
2135 output_names: &[String],
2136 ) -> Result<TypedExpr, PlannerError> {
2137 let group_key_map = build_group_key_map(group_keys);
2138 let aggregate_map = build_aggregate_map(aggregates);
2139
2140 rewrite_expr_with_maps(expr, &group_key_map, &aggregate_map, output_names)
2141 }
2142
2143 fn extract_limit_value(
2147 &self,
2148 expr: &Option<crate::ast::expr::Expr>,
2149 stmt_span: crate::ast::Span,
2150 ) -> Result<Option<u64>, PlannerError> {
2151 match expr {
2152 None => Ok(None),
2153 Some(e) => {
2154 if let crate::ast::expr::ExprKind::Literal {
2156 literal: Literal::Number(s),
2157 } = &e.kind
2158 {
2159 s.parse::<u64>().map(Some).map_err(|_| {
2160 PlannerError::type_mismatch("unsigned integer", s.clone(), e.span)
2161 })
2162 } else {
2163 Err(PlannerError::unsupported_feature(
2164 "non-literal LIMIT/OFFSET",
2165 "v0.3.0+",
2166 stmt_span,
2167 ))
2168 }
2169 }
2170 }
2171 }
2172
2173 fn plan_insert(&self, stmt: &Insert) -> Result<LogicalPlan, PlannerError> {
2178 let table = self.name_resolver.resolve_table(&stmt.table, stmt.span)?;
2180
2181 let columns: Vec<String> = if let Some(ref cols) = stmt.columns {
2183 for col in cols {
2185 self.name_resolver.resolve_column(table, col, stmt.span)?;
2186 }
2187 cols.clone()
2188 } else {
2189 table.column_names().into_iter().map(String::from).collect()
2191 };
2192
2193 match &stmt.source {
2194 InsertSource::Values { values } => {
2195 let mut typed_values: Vec<Vec<TypedExpr>> = Vec::new();
2196
2197 for row in values {
2198 if row.len() != columns.len() {
2199 return Err(PlannerError::column_value_count_mismatch(
2200 columns.len(),
2201 row.len(),
2202 stmt.span,
2203 ));
2204 }
2205
2206 typed_values.push(self.type_check_insert_values(row, &columns, table)?);
2207 }
2208
2209 Ok(LogicalPlan::Insert {
2210 table: table.name.clone(),
2211 columns,
2212 values: typed_values,
2213 })
2214 }
2215 InsertSource::Select { select } => {
2216 let source = self.plan_select_relation(select, &[])?;
2217 if source.schema.len() != columns.len() {
2218 return Err(PlannerError::column_value_count_mismatch(
2219 columns.len(),
2220 source.schema.len(),
2221 stmt.span,
2222 ));
2223 }
2224
2225 for (source_column, target_column) in source.schema.iter().zip(&columns) {
2226 let target = table
2227 .get_column(target_column)
2228 .expect("validated target column");
2229 if target.not_null && source_column.data_type == ResolvedType::Null {
2230 return Err(PlannerError::null_constraint_violation(
2231 target_column,
2232 stmt.span,
2233 ));
2234 }
2235 self.validate_resolved_type_assignment(
2236 &source_column.data_type,
2237 &target.data_type,
2238 stmt.span,
2239 )?;
2240 }
2241
2242 Ok(LogicalPlan::InsertSelect {
2243 table: table.name.clone(),
2244 columns,
2245 source: Box::new(source.plan),
2246 })
2247 }
2248 }
2249 }
2250
2251 fn type_check_insert_values(
2253 &self,
2254 values: &[crate::ast::expr::Expr],
2255 columns: &[String],
2256 table: &TableMetadata,
2257 ) -> Result<Vec<TypedExpr>, PlannerError> {
2258 let mut typed_values = Vec::new();
2259
2260 for (i, value) in values.iter().enumerate() {
2261 let column_name = &columns[i];
2262 let column_meta = table.get_column(column_name).ok_or_else(|| {
2263 PlannerError::column_not_found(column_name, &table.name, value.span)
2264 })?;
2265
2266 let typed_value = self.type_checker.infer_type(value, table)?;
2268
2269 if column_meta.not_null
2271 && matches!(&typed_value.kind, TypedExprKind::Literal(Literal::Null))
2272 {
2273 return Err(PlannerError::null_constraint_violation(
2274 column_name,
2275 value.span,
2276 ));
2277 }
2278
2279 self.validate_type_assignment(&typed_value, &column_meta.data_type, value.span)?;
2281
2282 let typed_value =
2283 self.coerce_assignment_value(typed_value, &column_meta.data_type, value.span);
2284
2285 typed_values.push(typed_value);
2286 }
2287
2288 Ok(typed_values)
2289 }
2290
2291 fn validate_type_assignment(
2293 &self,
2294 value: &TypedExpr,
2295 target_type: &ResolvedType,
2296 span: crate::ast::Span,
2297 ) -> Result<(), PlannerError> {
2298 self.validate_resolved_type_assignment(&value.resolved_type, target_type, span)
2299 }
2300
2301 fn validate_resolved_type_assignment(
2302 &self,
2303 source_type: &ResolvedType,
2304 target_type: &ResolvedType,
2305 span: crate::ast::Span,
2306 ) -> Result<(), PlannerError> {
2307 if *source_type == ResolvedType::Null {
2309 return Ok(());
2310 }
2311
2312 if self.types_compatible(source_type, target_type) {
2314 return Ok(());
2315 }
2316
2317 Err(PlannerError::type_mismatch(
2318 target_type.to_string(),
2319 source_type.to_string(),
2320 span,
2321 ))
2322 }
2323
2324 fn types_compatible(&self, source: &ResolvedType, target: &ResolvedType) -> bool {
2326 use ResolvedType::*;
2327
2328 if source == target {
2330 return true;
2331 }
2332
2333 match (source, target) {
2335 (Integer, BigInt) | (Integer, Float) | (Integer, Double) => true,
2337 (BigInt, Float) | (BigInt, Double) => true,
2339 (Float, Double) => true,
2341 (Double, Float) => true,
2344 (Text | Integer | BigInt | Float | Double, Timestamp) => true,
2347 (Vector { dimension: d1, .. }, Vector { dimension: d2, .. }) => d1 == d2,
2349 _ => false,
2350 }
2351 }
2352
2353 fn coerce_assignment_value(
2354 &self,
2355 value: TypedExpr,
2356 target_type: &ResolvedType,
2357 span: crate::ast::Span,
2358 ) -> TypedExpr {
2359 if matches!(target_type, ResolvedType::Timestamp)
2360 && !matches!(
2361 value.resolved_type,
2362 ResolvedType::Timestamp | ResolvedType::Null
2363 )
2364 {
2365 TypedExpr::cast(value, ResolvedType::Timestamp, span)
2366 } else {
2367 value
2368 }
2369 }
2370
2371 fn plan_update(&self, stmt: &Update) -> Result<LogicalPlan, PlannerError> {
2375 let table = self.name_resolver.resolve_table(&stmt.table, stmt.span)?;
2377
2378 let mut typed_assignments = Vec::new();
2380
2381 for assignment in &stmt.assignments {
2382 let column_meta =
2384 self.name_resolver
2385 .resolve_column(table, &assignment.column, assignment.span)?;
2386 let column_index = table.get_column_index(&assignment.column).unwrap();
2387
2388 let typed_value = self.type_checker.infer_type(&assignment.value, table)?;
2390
2391 if column_meta.not_null
2393 && matches!(&typed_value.kind, TypedExprKind::Literal(Literal::Null))
2394 {
2395 return Err(PlannerError::null_constraint_violation(
2396 &assignment.column,
2397 assignment.value.span,
2398 ));
2399 }
2400
2401 self.validate_type_assignment(
2403 &typed_value,
2404 &column_meta.data_type,
2405 assignment.value.span,
2406 )?;
2407
2408 let typed_value = self.coerce_assignment_value(
2409 typed_value,
2410 &column_meta.data_type,
2411 assignment.value.span,
2412 );
2413
2414 typed_assignments.push(TypedAssignment::new(
2415 assignment.column.clone(),
2416 column_index,
2417 typed_value,
2418 ));
2419 }
2420
2421 let filter = if let Some(ref selection) = stmt.selection {
2423 let predicate = self.type_checker.infer_type(selection, table)?;
2424
2425 if predicate.resolved_type != ResolvedType::Boolean {
2427 return Err(PlannerError::type_mismatch(
2428 "Boolean",
2429 predicate.resolved_type.to_string(),
2430 selection.span,
2431 ));
2432 }
2433
2434 Some(predicate)
2435 } else {
2436 None
2437 };
2438
2439 Ok(LogicalPlan::Update {
2440 table: table.name.clone(),
2441 assignments: typed_assignments,
2442 filter,
2443 })
2444 }
2445
2446 fn plan_delete(&self, stmt: &Delete) -> Result<LogicalPlan, PlannerError> {
2450 let table = self.name_resolver.resolve_table(&stmt.table, stmt.span)?;
2452
2453 let filter = if let Some(ref selection) = stmt.selection {
2455 let predicate = self.type_checker.infer_type(selection, table)?;
2456
2457 if predicate.resolved_type != ResolvedType::Boolean {
2459 return Err(PlannerError::type_mismatch(
2460 "Boolean",
2461 predicate.resolved_type.to_string(),
2462 selection.span,
2463 ));
2464 }
2465
2466 Some(predicate)
2467 } else {
2468 None
2469 };
2470
2471 Ok(LogicalPlan::Delete {
2472 table: table.name.clone(),
2473 filter,
2474 })
2475 }
2476}
2477
2478#[derive(Debug, Clone, PartialEq, Eq, Hash)]
2479struct AggregateSignature {
2480 name: String,
2481 distinct: bool,
2482 star: bool,
2483 arg_key: Option<String>,
2484 separator: Option<String>,
2485}
2486
2487fn expr_contains_aggregate(expr: &crate::ast::expr::Expr) -> bool {
2488 use crate::ast::expr::ExprKind;
2489
2490 match &expr.kind {
2491 ExprKind::FunctionCall { name, args, .. } => {
2492 if is_aggregate_function(name) {
2493 return true;
2494 }
2495 args.iter().any(expr_contains_aggregate)
2496 }
2497 ExprKind::BinaryOp { left, right, .. } => {
2498 expr_contains_aggregate(left) || expr_contains_aggregate(right)
2499 }
2500 ExprKind::UnaryOp { operand, .. } => expr_contains_aggregate(operand),
2501 ExprKind::Cast { expr, .. } => expr_contains_aggregate(expr),
2502 ExprKind::Between {
2503 expr, low, high, ..
2504 } => {
2505 expr_contains_aggregate(expr)
2506 || expr_contains_aggregate(low)
2507 || expr_contains_aggregate(high)
2508 }
2509 ExprKind::Like {
2510 expr,
2511 pattern,
2512 escape,
2513 ..
2514 } => {
2515 expr_contains_aggregate(expr)
2516 || expr_contains_aggregate(pattern)
2517 || escape.as_deref().is_some_and(expr_contains_aggregate)
2518 }
2519 ExprKind::InList { expr, list, .. } => {
2520 expr_contains_aggregate(expr) || list.iter().any(expr_contains_aggregate)
2521 }
2522 ExprKind::IsNull { expr, .. } => expr_contains_aggregate(expr),
2523 ExprKind::ScalarSubquery { .. }
2524 | ExprKind::InSubquery { .. }
2525 | ExprKind::Exists { .. }
2526 | ExprKind::Quantified { .. }
2527 | ExprKind::Literal { .. }
2528 | ExprKind::VectorLiteral { .. }
2529 | ExprKind::ColumnRef { .. } => false,
2530 }
2531}
2532
2533fn typed_expr_contains_aggregate(expr: &TypedExpr) -> bool {
2534 match &expr.kind {
2535 TypedExprKind::FunctionCall { name, args, .. } => {
2536 if is_aggregate_function(name) {
2537 return true;
2538 }
2539 args.iter().any(typed_expr_contains_aggregate)
2540 }
2541 TypedExprKind::BinaryOp { left, right, .. } => {
2542 typed_expr_contains_aggregate(left) || typed_expr_contains_aggregate(right)
2543 }
2544 TypedExprKind::UnaryOp { operand, .. } => typed_expr_contains_aggregate(operand),
2545 TypedExprKind::Between {
2546 expr, low, high, ..
2547 } => {
2548 typed_expr_contains_aggregate(expr)
2549 || typed_expr_contains_aggregate(low)
2550 || typed_expr_contains_aggregate(high)
2551 }
2552 TypedExprKind::Like {
2553 expr,
2554 pattern,
2555 escape,
2556 ..
2557 } => {
2558 typed_expr_contains_aggregate(expr)
2559 || typed_expr_contains_aggregate(pattern)
2560 || escape
2561 .as_ref()
2562 .is_some_and(|inner| typed_expr_contains_aggregate(inner))
2563 }
2564 TypedExprKind::InList { expr, list, .. } => {
2565 typed_expr_contains_aggregate(expr) || list.iter().any(typed_expr_contains_aggregate)
2566 }
2567 TypedExprKind::IsNull { expr, .. } => typed_expr_contains_aggregate(expr),
2568 TypedExprKind::InSubquery { expr, .. } => typed_expr_contains_aggregate(expr),
2569 TypedExprKind::Quantified { expr, .. } => typed_expr_contains_aggregate(expr),
2570 TypedExprKind::ScalarSubquery(_) | TypedExprKind::Exists { .. } => false,
2571 _ => false,
2572 }
2573}
2574
2575fn map_join_type(join_type: crate::ast::dml::JoinType) -> JoinType {
2576 match join_type {
2577 crate::ast::dml::JoinType::Inner => JoinType::Inner,
2578 crate::ast::dml::JoinType::Left => JoinType::Left,
2579 crate::ast::dml::JoinType::Right => JoinType::Right,
2580 crate::ast::dml::JoinType::Full => JoinType::Full,
2581 crate::ast::dml::JoinType::Cross => JoinType::Cross,
2582 }
2583}
2584
2585struct FoundScopedColumn {
2586 table: String,
2587 index: usize,
2588 ty: ResolvedType,
2589 partner_indices: Vec<usize>,
2590}
2591
2592fn find_scoped_column(
2593 scope: &[ScopedTable],
2594 column: &str,
2595 span: crate::ast::Span,
2596) -> Result<FoundScopedColumn, PlannerError> {
2597 let mut matches = Vec::new();
2598 for table in scope {
2599 if table.hidden_unqualified_columns.contains(column) {
2600 continue;
2601 }
2602 if let Some(local_idx) = table.table.get_column_index(column) {
2603 let meta = &table.table.columns[local_idx];
2604 matches.push(FoundScopedColumn {
2605 table: table.table.name.clone(),
2606 index: table.start_index + local_idx,
2607 ty: meta.data_type.clone(),
2608 partner_indices: table
2609 .merged_column_partners
2610 .get(column)
2611 .cloned()
2612 .unwrap_or_default(),
2613 });
2614 }
2615 }
2616 match matches.len() {
2617 0 => Err(PlannerError::column_not_found(column, "JOIN input", span)),
2618 1 => Ok(matches.remove(0)),
2619 _ => Err(PlannerError::ambiguous_column(
2620 column,
2621 scope.iter().map(|s| s.table.name.clone()).collect(),
2622 span,
2623 )),
2624 }
2625}
2626
2627fn merged_scoped_column_expr(
2628 found: &FoundScopedColumn,
2629 column: &str,
2630 span: crate::ast::Span,
2631) -> TypedExpr {
2632 let own = TypedExpr::column_ref(
2633 found.table.clone(),
2634 column.to_string(),
2635 found.index,
2636 found.ty.clone(),
2637 span,
2638 );
2639 if found.partner_indices.is_empty() {
2640 return own;
2641 }
2642
2643 let mut args = Vec::with_capacity(found.partner_indices.len() + 1);
2644 args.push(own);
2645 args.extend(found.partner_indices.iter().map(|&index| {
2646 TypedExpr::column_ref(
2647 found.table.clone(),
2648 column.to_string(),
2649 index,
2650 found.ty.clone(),
2651 span,
2652 )
2653 }));
2654 TypedExpr {
2655 kind: TypedExprKind::FunctionCall {
2656 name: "coalesce".to_string(),
2657 args,
2658 distinct: false,
2659 star: false,
2660 },
2661 resolved_type: found.ty.clone(),
2662 span,
2663 }
2664}
2665
2666fn projection_schema(
2667 projection: &Projection,
2668 input_schema: &[ColumnMetadata],
2669) -> Vec<ColumnMetadata> {
2670 match projection {
2671 Projection::All(names) => names
2672 .iter()
2673 .enumerate()
2674 .map(|(idx, name)| {
2675 let ty = (names.len() == input_schema.len())
2676 .then(|| input_schema.get(idx))
2677 .flatten()
2678 .or_else(|| input_schema.iter().find(|col| &col.name == name))
2679 .map(|col| col.data_type.clone())
2680 .unwrap_or(ResolvedType::Null);
2681 ColumnMetadata::new(name.clone(), ty)
2682 })
2683 .collect(),
2684 Projection::Columns(columns) => columns
2685 .iter()
2686 .enumerate()
2687 .map(|(idx, col)| {
2688 let name = col
2689 .alias
2690 .clone()
2691 .or_else(|| match &col.expr.kind {
2692 TypedExprKind::ColumnRef { column, .. } => Some(column.clone()),
2693 TypedExprKind::FunctionCall { name, args, .. }
2697 if name == "coalesce" && !args.is_empty() =>
2698 {
2699 let first_column = match &args[0].kind {
2700 TypedExprKind::ColumnRef { column, .. } => Some(column),
2701 _ => None,
2702 };
2703 first_column
2704 .filter(|column| {
2705 args.iter().all(|arg| {
2706 matches!(
2707 &arg.kind,
2708 TypedExprKind::ColumnRef { column: other, .. }
2709 if other == *column
2710 )
2711 })
2712 })
2713 .cloned()
2714 }
2715 _ => None,
2716 })
2717 .unwrap_or_else(|| format!("col_{idx}"));
2718 ColumnMetadata::new(name, col.expr.resolved_type.clone())
2719 })
2720 .collect(),
2721 }
2722}
2723
2724fn visible_wildcard_columns(schema: &[ColumnMetadata], scope: &[ScopedTable]) -> Vec<String> {
2725 schema
2726 .iter()
2727 .enumerate()
2728 .filter(|(index, column)| {
2729 !scope.iter().any(|table| {
2730 *index >= table.start_index
2731 && *index < table.start_index + table.table.columns.len()
2732 && table.hidden_unqualified_columns.contains(&column.name)
2733 })
2734 })
2735 .map(|(_, column)| column.name.clone())
2736 .collect()
2737}
2738
2739fn offset_scope(scope: &[ScopedTable], offset: usize) -> Vec<ScopedTable> {
2740 scope
2741 .iter()
2742 .cloned()
2743 .map(|mut table| {
2744 table.start_index += offset;
2745 table.scope_level += 1;
2746 table
2747 })
2748 .collect()
2749}
2750
2751fn natural_join_columns(
2752 left_schema: &[ColumnMetadata],
2753 right_schema: &[ColumnMetadata],
2754) -> Vec<String> {
2755 let right_names = right_schema
2759 .iter()
2760 .map(|column| column.name.as_str())
2761 .collect::<HashSet<_>>();
2762 left_schema
2763 .iter()
2764 .filter(|left| right_names.contains(left.name.as_str()))
2765 .map(|column| column.name.clone())
2766 .collect()
2767}
2768
2769fn install_base_projection(plan: &mut LogicalPlan, projection: &Projection) {
2770 match plan {
2771 LogicalPlan::Scan {
2772 projection: scan_projection,
2773 ..
2774 } => *scan_projection = projection.clone(),
2775 LogicalPlan::Filter { input, .. } => install_base_projection(input, projection),
2776 _ => {}
2777 }
2778}
2779
2780fn is_aggregate_function(name: &str) -> bool {
2781 matches!(
2782 name.to_ascii_lowercase().as_str(),
2783 "count" | "sum" | "total" | "avg" | "min" | "max" | "group_concat" | "string_agg"
2784 )
2785}
2786
2787fn expr_key(expr: &TypedExpr) -> String {
2788 format!("{:?}", expr.kind)
2789}
2790
2791fn aggregate_signature(
2792 name: &str,
2793 distinct: bool,
2794 star: bool,
2795 arg: Option<&TypedExpr>,
2796 separator: Option<&String>,
2797 _expr: &TypedExpr,
2798) -> AggregateSignature {
2799 AggregateSignature {
2800 name: name.to_ascii_lowercase(),
2801 distinct,
2802 star,
2803 arg_key: arg.map(expr_key),
2804 separator: separator.cloned(),
2805 }
2806}
2807
2808fn build_group_key_map(group_keys: &[TypedExpr]) -> HashMap<String, usize> {
2809 let mut map = HashMap::new();
2810 for (idx, key) in group_keys.iter().enumerate() {
2811 map.insert(expr_key(key), idx);
2812 }
2813 map
2814}
2815
2816fn build_aggregate_map(aggregates: &[AggregateExpr]) -> HashMap<AggregateSignature, usize> {
2817 let mut map = HashMap::new();
2818 for (idx, agg) in aggregates.iter().enumerate() {
2819 let (name, separator, star, arg) = match &agg.function {
2820 AggregateFunction::Count => (
2821 "count".to_string(),
2822 None,
2823 agg.arg.is_none(),
2824 agg.arg.as_ref(),
2825 ),
2826 AggregateFunction::Sum => ("sum".to_string(), None, false, agg.arg.as_ref()),
2827 AggregateFunction::Total => ("total".to_string(), None, false, agg.arg.as_ref()),
2828 AggregateFunction::Avg => ("avg".to_string(), None, false, agg.arg.as_ref()),
2829 AggregateFunction::Min => ("min".to_string(), None, false, agg.arg.as_ref()),
2830 AggregateFunction::Max => ("max".to_string(), None, false, agg.arg.as_ref()),
2831 AggregateFunction::GroupConcat { separator } => (
2832 "group_concat".to_string(),
2833 separator.clone(),
2834 false,
2835 agg.arg.as_ref(),
2836 ),
2837 AggregateFunction::StringAgg { separator } => (
2838 "string_agg".to_string(),
2839 separator.clone(),
2840 false,
2841 agg.arg.as_ref(),
2842 ),
2843 };
2844 let signature = AggregateSignature {
2845 name,
2846 distinct: agg.distinct,
2847 star,
2848 arg_key: arg.map(expr_key),
2849 separator,
2850 };
2851 map.insert(signature, idx);
2852 }
2853 map
2854}
2855
2856fn build_aggregate_schema(
2857 group_keys: &[TypedExpr],
2858 aggregates: &[AggregateExpr],
2859) -> Vec<ColumnMetadata> {
2860 let mut schema = Vec::new();
2861 for (idx, key) in group_keys.iter().enumerate() {
2862 let name = match &key.kind {
2863 TypedExprKind::ColumnRef { column, .. } => column.clone(),
2864 _ => format!("group_{idx}"),
2865 };
2866 schema.push(ColumnMetadata::new(name, key.resolved_type.clone()));
2867 }
2868 for (idx, agg) in aggregates.iter().enumerate() {
2869 let name = match &agg.function {
2870 AggregateFunction::Count => format!("count_{idx}"),
2871 AggregateFunction::Sum => format!("sum_{idx}"),
2872 AggregateFunction::Total => format!("total_{idx}"),
2873 AggregateFunction::Avg => format!("avg_{idx}"),
2874 AggregateFunction::Min => format!("min_{idx}"),
2875 AggregateFunction::Max => format!("max_{idx}"),
2876 AggregateFunction::GroupConcat { .. } => format!("group_concat_{idx}"),
2877 AggregateFunction::StringAgg { .. } => format!("string_agg_{idx}"),
2878 };
2879 schema.push(ColumnMetadata::new(name, agg.result_type.clone()));
2880 }
2881 schema
2882}
2883
2884fn rewrite_expr_with_maps(
2885 expr: &TypedExpr,
2886 group_key_map: &HashMap<String, usize>,
2887 aggregate_map: &HashMap<AggregateSignature, usize>,
2888 output_names: &[String],
2889) -> Result<TypedExpr, PlannerError> {
2890 let group_key_count = output_names.len().saturating_sub(aggregate_map.len());
2891 let key = expr_key(expr);
2892 if let Some(idx) = group_key_map.get(&key) {
2893 return Ok(make_output_column_ref(
2894 *idx,
2895 output_names,
2896 expr.resolved_type.clone(),
2897 expr.span,
2898 ));
2899 }
2900
2901 match &expr.kind {
2902 TypedExprKind::FunctionCall {
2903 name,
2904 args,
2905 distinct,
2906 star,
2907 } if is_aggregate_function(name) => {
2908 let separator = if name.eq_ignore_ascii_case("group_concat") && args.len() == 2 {
2909 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
2910 Some(value.clone())
2911 } else {
2912 return Err(PlannerError::invalid_expression(
2913 "GROUP_CONCAT separator must be a string literal".to_string(),
2914 ));
2915 }
2916 } else if name.eq_ignore_ascii_case("string_agg") && args.len() == 2 {
2917 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
2918 Some(value.clone())
2919 } else {
2920 return Err(PlannerError::invalid_expression(
2921 "STRING_AGG separator must be a string literal".to_string(),
2922 ));
2923 }
2924 } else {
2925 None
2926 };
2927 let signature = AggregateSignature {
2928 name: name.to_ascii_lowercase(),
2929 distinct: *distinct,
2930 star: *star,
2931 arg_key: args.first().map(expr_key),
2932 separator,
2933 };
2934 let idx = aggregate_map.get(&signature).ok_or_else(|| {
2935 PlannerError::invalid_expression(
2936 "aggregate in expression is not part of plan".to_string(),
2937 )
2938 })?;
2939 let output_index = group_key_count + idx;
2940 Ok(make_output_column_ref(
2941 output_index,
2942 output_names,
2943 expr.resolved_type.clone(),
2944 expr.span,
2945 ))
2946 }
2947 TypedExprKind::FunctionCall {
2948 name,
2949 args,
2950 distinct,
2951 star,
2952 } => {
2953 if *distinct || *star {
2954 return Err(PlannerError::invalid_expression(
2955 "DISTINCT/STAR modifiers are only supported for aggregates".to_string(),
2956 ));
2957 }
2958 let mut rewritten_args = Vec::with_capacity(args.len());
2959 for arg in args {
2960 rewritten_args.push(rewrite_expr_with_maps(
2961 arg,
2962 group_key_map,
2963 aggregate_map,
2964 output_names,
2965 )?);
2966 }
2967 Ok(TypedExpr {
2968 kind: TypedExprKind::FunctionCall {
2969 name: name.clone(),
2970 args: rewritten_args,
2971 distinct: false,
2972 star: false,
2973 },
2974 resolved_type: expr.resolved_type.clone(),
2975 span: expr.span,
2976 })
2977 }
2978 TypedExprKind::BinaryOp { left, op, right } => {
2979 let left = rewrite_expr_with_maps(left, group_key_map, aggregate_map, output_names)?;
2980 let right = rewrite_expr_with_maps(right, group_key_map, aggregate_map, output_names)?;
2981 Ok(TypedExpr {
2982 kind: TypedExprKind::BinaryOp {
2983 left: Box::new(left),
2984 op: *op,
2985 right: Box::new(right),
2986 },
2987 resolved_type: expr.resolved_type.clone(),
2988 span: expr.span,
2989 })
2990 }
2991 TypedExprKind::UnaryOp { op, operand } => {
2992 let operand =
2993 rewrite_expr_with_maps(operand, group_key_map, aggregate_map, output_names)?;
2994 Ok(TypedExpr {
2995 kind: TypedExprKind::UnaryOp {
2996 op: *op,
2997 operand: Box::new(operand),
2998 },
2999 resolved_type: expr.resolved_type.clone(),
3000 span: expr.span,
3001 })
3002 }
3003 TypedExprKind::Between {
3004 expr: inner,
3005 low,
3006 high,
3007 negated,
3008 } => {
3009 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
3010 let low = rewrite_expr_with_maps(low, group_key_map, aggregate_map, output_names)?;
3011 let high = rewrite_expr_with_maps(high, group_key_map, aggregate_map, output_names)?;
3012 Ok(TypedExpr {
3013 kind: TypedExprKind::Between {
3014 expr: Box::new(inner),
3015 low: Box::new(low),
3016 high: Box::new(high),
3017 negated: *negated,
3018 },
3019 resolved_type: expr.resolved_type.clone(),
3020 span: expr.span,
3021 })
3022 }
3023 TypedExprKind::Like {
3024 expr: inner,
3025 pattern,
3026 escape,
3027 negated,
3028 kind,
3029 } => {
3030 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
3031 let pattern =
3032 rewrite_expr_with_maps(pattern, group_key_map, aggregate_map, output_names)?;
3033 let escape = if let Some(esc) = escape {
3034 Some(Box::new(rewrite_expr_with_maps(
3035 esc,
3036 group_key_map,
3037 aggregate_map,
3038 output_names,
3039 )?))
3040 } else {
3041 None
3042 };
3043 Ok(TypedExpr {
3044 kind: TypedExprKind::Like {
3045 expr: Box::new(inner),
3046 pattern: Box::new(pattern),
3047 escape,
3048 negated: *negated,
3049 kind: *kind,
3050 },
3051 resolved_type: expr.resolved_type.clone(),
3052 span: expr.span,
3053 })
3054 }
3055 TypedExprKind::InList {
3056 expr: inner,
3057 list,
3058 negated,
3059 } => {
3060 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
3061 let mut rewritten_list = Vec::with_capacity(list.len());
3062 for item in list {
3063 rewritten_list.push(rewrite_expr_with_maps(
3064 item,
3065 group_key_map,
3066 aggregate_map,
3067 output_names,
3068 )?);
3069 }
3070 Ok(TypedExpr {
3071 kind: TypedExprKind::InList {
3072 expr: Box::new(inner),
3073 list: rewritten_list,
3074 negated: *negated,
3075 },
3076 resolved_type: expr.resolved_type.clone(),
3077 span: expr.span,
3078 })
3079 }
3080 TypedExprKind::IsNull {
3081 expr: inner,
3082 negated,
3083 } => {
3084 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
3085 Ok(TypedExpr {
3086 kind: TypedExprKind::IsNull {
3087 expr: Box::new(inner),
3088 negated: *negated,
3089 },
3090 resolved_type: expr.resolved_type.clone(),
3091 span: expr.span,
3092 })
3093 }
3094 TypedExprKind::Literal(_) | TypedExprKind::VectorLiteral(_) => Ok(expr.clone()),
3095 TypedExprKind::ColumnRef { .. } => Err(PlannerError::invalid_expression(
3096 "column reference must appear in GROUP BY or be aggregated".to_string(),
3097 )),
3098 TypedExprKind::Cast {
3099 expr: inner,
3100 target_type,
3101 } => {
3102 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
3103 Ok(TypedExpr {
3104 kind: TypedExprKind::Cast {
3105 expr: Box::new(inner),
3106 target_type: target_type.clone(),
3107 },
3108 resolved_type: expr.resolved_type.clone(),
3109 span: expr.span,
3110 })
3111 }
3112 TypedExprKind::ScalarSubquery(_)
3113 | TypedExprKind::InSubquery { .. }
3114 | TypedExprKind::Exists { .. }
3115 | TypedExprKind::Quantified { .. } => Ok(expr.clone()),
3116 }
3117}
3118
3119fn make_output_column_ref(
3120 index: usize,
3121 output_names: &[String],
3122 resolved_type: ResolvedType,
3123 span: crate::ast::Span,
3124) -> TypedExpr {
3125 let name = output_names
3126 .get(index)
3127 .cloned()
3128 .unwrap_or_else(|| format!("col_{index}"));
3129 TypedExpr::column_ref("__agg__".to_string(), name, index, resolved_type, span)
3130}