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 value.resolved_type != *target_type
2360 && value.resolved_type != ResolvedType::Null
2361 && matches!(
2362 target_type,
2363 ResolvedType::Integer
2364 | ResolvedType::BigInt
2365 | ResolvedType::Float
2366 | ResolvedType::Double
2367 | ResolvedType::Timestamp
2368 )
2369 {
2370 TypedExpr::cast(value, target_type.clone(), span)
2371 } else {
2372 value
2373 }
2374 }
2375
2376 fn plan_update(&self, stmt: &Update) -> Result<LogicalPlan, PlannerError> {
2380 let table = self.name_resolver.resolve_table(&stmt.table, stmt.span)?;
2382
2383 let mut typed_assignments = Vec::new();
2385
2386 for assignment in &stmt.assignments {
2387 let column_meta =
2389 self.name_resolver
2390 .resolve_column(table, &assignment.column, assignment.span)?;
2391 let column_index = table.get_column_index(&assignment.column).unwrap();
2392
2393 let typed_value = self.type_checker.infer_type(&assignment.value, table)?;
2395
2396 if column_meta.not_null
2398 && matches!(&typed_value.kind, TypedExprKind::Literal(Literal::Null))
2399 {
2400 return Err(PlannerError::null_constraint_violation(
2401 &assignment.column,
2402 assignment.value.span,
2403 ));
2404 }
2405
2406 self.validate_type_assignment(
2408 &typed_value,
2409 &column_meta.data_type,
2410 assignment.value.span,
2411 )?;
2412
2413 let typed_value = self.coerce_assignment_value(
2414 typed_value,
2415 &column_meta.data_type,
2416 assignment.value.span,
2417 );
2418
2419 typed_assignments.push(TypedAssignment::new(
2420 assignment.column.clone(),
2421 column_index,
2422 typed_value,
2423 ));
2424 }
2425
2426 let filter = if let Some(ref selection) = stmt.selection {
2428 let predicate = self.type_checker.infer_type(selection, table)?;
2429
2430 if predicate.resolved_type != ResolvedType::Boolean {
2432 return Err(PlannerError::type_mismatch(
2433 "Boolean",
2434 predicate.resolved_type.to_string(),
2435 selection.span,
2436 ));
2437 }
2438
2439 Some(predicate)
2440 } else {
2441 None
2442 };
2443
2444 Ok(LogicalPlan::Update {
2445 table: table.name.clone(),
2446 assignments: typed_assignments,
2447 filter,
2448 })
2449 }
2450
2451 fn plan_delete(&self, stmt: &Delete) -> Result<LogicalPlan, PlannerError> {
2455 let table = self.name_resolver.resolve_table(&stmt.table, stmt.span)?;
2457
2458 let filter = if let Some(ref selection) = stmt.selection {
2460 let predicate = self.type_checker.infer_type(selection, table)?;
2461
2462 if predicate.resolved_type != ResolvedType::Boolean {
2464 return Err(PlannerError::type_mismatch(
2465 "Boolean",
2466 predicate.resolved_type.to_string(),
2467 selection.span,
2468 ));
2469 }
2470
2471 Some(predicate)
2472 } else {
2473 None
2474 };
2475
2476 Ok(LogicalPlan::Delete {
2477 table: table.name.clone(),
2478 filter,
2479 })
2480 }
2481}
2482
2483#[derive(Debug, Clone, PartialEq, Eq, Hash)]
2484struct AggregateSignature {
2485 name: String,
2486 distinct: bool,
2487 star: bool,
2488 arg_key: Option<String>,
2489 separator: Option<String>,
2490}
2491
2492fn expr_contains_aggregate(expr: &crate::ast::expr::Expr) -> bool {
2493 use crate::ast::expr::ExprKind;
2494
2495 match &expr.kind {
2496 ExprKind::FunctionCall { name, args, .. } => {
2497 if is_aggregate_function(name) {
2498 return true;
2499 }
2500 args.iter().any(expr_contains_aggregate)
2501 }
2502 ExprKind::BinaryOp { left, right, .. } => {
2503 expr_contains_aggregate(left) || expr_contains_aggregate(right)
2504 }
2505 ExprKind::UnaryOp { operand, .. } => expr_contains_aggregate(operand),
2506 ExprKind::Cast { expr, .. } => expr_contains_aggregate(expr),
2507 ExprKind::Between {
2508 expr, low, high, ..
2509 } => {
2510 expr_contains_aggregate(expr)
2511 || expr_contains_aggregate(low)
2512 || expr_contains_aggregate(high)
2513 }
2514 ExprKind::Like {
2515 expr,
2516 pattern,
2517 escape,
2518 ..
2519 } => {
2520 expr_contains_aggregate(expr)
2521 || expr_contains_aggregate(pattern)
2522 || escape.as_deref().is_some_and(expr_contains_aggregate)
2523 }
2524 ExprKind::InList { expr, list, .. } => {
2525 expr_contains_aggregate(expr) || list.iter().any(expr_contains_aggregate)
2526 }
2527 ExprKind::IsNull { expr, .. } => expr_contains_aggregate(expr),
2528 ExprKind::ScalarSubquery { .. }
2529 | ExprKind::InSubquery { .. }
2530 | ExprKind::Exists { .. }
2531 | ExprKind::Quantified { .. }
2532 | ExprKind::Literal { .. }
2533 | ExprKind::VectorLiteral { .. }
2534 | ExprKind::ColumnRef { .. } => false,
2535 }
2536}
2537
2538fn typed_expr_contains_aggregate(expr: &TypedExpr) -> bool {
2539 match &expr.kind {
2540 TypedExprKind::FunctionCall { name, args, .. } => {
2541 if is_aggregate_function(name) {
2542 return true;
2543 }
2544 args.iter().any(typed_expr_contains_aggregate)
2545 }
2546 TypedExprKind::BinaryOp { left, right, .. } => {
2547 typed_expr_contains_aggregate(left) || typed_expr_contains_aggregate(right)
2548 }
2549 TypedExprKind::UnaryOp { operand, .. } => typed_expr_contains_aggregate(operand),
2550 TypedExprKind::Between {
2551 expr, low, high, ..
2552 } => {
2553 typed_expr_contains_aggregate(expr)
2554 || typed_expr_contains_aggregate(low)
2555 || typed_expr_contains_aggregate(high)
2556 }
2557 TypedExprKind::Like {
2558 expr,
2559 pattern,
2560 escape,
2561 ..
2562 } => {
2563 typed_expr_contains_aggregate(expr)
2564 || typed_expr_contains_aggregate(pattern)
2565 || escape
2566 .as_ref()
2567 .is_some_and(|inner| typed_expr_contains_aggregate(inner))
2568 }
2569 TypedExprKind::InList { expr, list, .. } => {
2570 typed_expr_contains_aggregate(expr) || list.iter().any(typed_expr_contains_aggregate)
2571 }
2572 TypedExprKind::IsNull { expr, .. } => typed_expr_contains_aggregate(expr),
2573 TypedExprKind::InSubquery { expr, .. } => typed_expr_contains_aggregate(expr),
2574 TypedExprKind::Quantified { expr, .. } => typed_expr_contains_aggregate(expr),
2575 TypedExprKind::ScalarSubquery(_) | TypedExprKind::Exists { .. } => false,
2576 _ => false,
2577 }
2578}
2579
2580fn map_join_type(join_type: crate::ast::dml::JoinType) -> JoinType {
2581 match join_type {
2582 crate::ast::dml::JoinType::Inner => JoinType::Inner,
2583 crate::ast::dml::JoinType::Left => JoinType::Left,
2584 crate::ast::dml::JoinType::Right => JoinType::Right,
2585 crate::ast::dml::JoinType::Full => JoinType::Full,
2586 crate::ast::dml::JoinType::Cross => JoinType::Cross,
2587 }
2588}
2589
2590struct FoundScopedColumn {
2591 table: String,
2592 index: usize,
2593 ty: ResolvedType,
2594 partner_indices: Vec<usize>,
2595}
2596
2597fn find_scoped_column(
2598 scope: &[ScopedTable],
2599 column: &str,
2600 span: crate::ast::Span,
2601) -> Result<FoundScopedColumn, PlannerError> {
2602 let mut matches = Vec::new();
2603 for table in scope {
2604 if table.hidden_unqualified_columns.contains(column) {
2605 continue;
2606 }
2607 if let Some(local_idx) = table.table.get_column_index(column) {
2608 let meta = &table.table.columns[local_idx];
2609 matches.push(FoundScopedColumn {
2610 table: table.table.name.clone(),
2611 index: table.start_index + local_idx,
2612 ty: meta.data_type.clone(),
2613 partner_indices: table
2614 .merged_column_partners
2615 .get(column)
2616 .cloned()
2617 .unwrap_or_default(),
2618 });
2619 }
2620 }
2621 match matches.len() {
2622 0 => Err(PlannerError::column_not_found(column, "JOIN input", span)),
2623 1 => Ok(matches.remove(0)),
2624 _ => Err(PlannerError::ambiguous_column(
2625 column,
2626 scope.iter().map(|s| s.table.name.clone()).collect(),
2627 span,
2628 )),
2629 }
2630}
2631
2632fn merged_scoped_column_expr(
2633 found: &FoundScopedColumn,
2634 column: &str,
2635 span: crate::ast::Span,
2636) -> TypedExpr {
2637 let own = TypedExpr::column_ref(
2638 found.table.clone(),
2639 column.to_string(),
2640 found.index,
2641 found.ty.clone(),
2642 span,
2643 );
2644 if found.partner_indices.is_empty() {
2645 return own;
2646 }
2647
2648 let mut args = Vec::with_capacity(found.partner_indices.len() + 1);
2649 args.push(own);
2650 args.extend(found.partner_indices.iter().map(|&index| {
2651 TypedExpr::column_ref(
2652 found.table.clone(),
2653 column.to_string(),
2654 index,
2655 found.ty.clone(),
2656 span,
2657 )
2658 }));
2659 TypedExpr {
2660 kind: TypedExprKind::FunctionCall {
2661 name: "coalesce".to_string(),
2662 args,
2663 distinct: false,
2664 star: false,
2665 },
2666 resolved_type: found.ty.clone(),
2667 span,
2668 }
2669}
2670
2671fn projection_schema(
2672 projection: &Projection,
2673 input_schema: &[ColumnMetadata],
2674) -> Vec<ColumnMetadata> {
2675 match projection {
2676 Projection::All(names) => names
2677 .iter()
2678 .enumerate()
2679 .map(|(idx, name)| {
2680 let ty = (names.len() == input_schema.len())
2681 .then(|| input_schema.get(idx))
2682 .flatten()
2683 .or_else(|| input_schema.iter().find(|col| &col.name == name))
2684 .map(|col| col.data_type.clone())
2685 .unwrap_or(ResolvedType::Null);
2686 ColumnMetadata::new(name.clone(), ty)
2687 })
2688 .collect(),
2689 Projection::Columns(columns) => columns
2690 .iter()
2691 .enumerate()
2692 .map(|(idx, col)| {
2693 let name = col
2694 .alias
2695 .clone()
2696 .or_else(|| match &col.expr.kind {
2697 TypedExprKind::ColumnRef { column, .. } => Some(column.clone()),
2698 TypedExprKind::FunctionCall { name, args, .. }
2702 if name == "coalesce" && !args.is_empty() =>
2703 {
2704 let first_column = match &args[0].kind {
2705 TypedExprKind::ColumnRef { column, .. } => Some(column),
2706 _ => None,
2707 };
2708 first_column
2709 .filter(|column| {
2710 args.iter().all(|arg| {
2711 matches!(
2712 &arg.kind,
2713 TypedExprKind::ColumnRef { column: other, .. }
2714 if other == *column
2715 )
2716 })
2717 })
2718 .cloned()
2719 }
2720 _ => None,
2721 })
2722 .unwrap_or_else(|| format!("col_{idx}"));
2723 ColumnMetadata::new(name, col.expr.resolved_type.clone())
2724 })
2725 .collect(),
2726 }
2727}
2728
2729fn visible_wildcard_columns(schema: &[ColumnMetadata], scope: &[ScopedTable]) -> Vec<String> {
2730 schema
2731 .iter()
2732 .enumerate()
2733 .filter(|(index, column)| {
2734 !scope.iter().any(|table| {
2735 *index >= table.start_index
2736 && *index < table.start_index + table.table.columns.len()
2737 && table.hidden_unqualified_columns.contains(&column.name)
2738 })
2739 })
2740 .map(|(_, column)| column.name.clone())
2741 .collect()
2742}
2743
2744fn offset_scope(scope: &[ScopedTable], offset: usize) -> Vec<ScopedTable> {
2745 scope
2746 .iter()
2747 .cloned()
2748 .map(|mut table| {
2749 table.start_index += offset;
2750 table.scope_level += 1;
2751 table
2752 })
2753 .collect()
2754}
2755
2756fn natural_join_columns(
2757 left_schema: &[ColumnMetadata],
2758 right_schema: &[ColumnMetadata],
2759) -> Vec<String> {
2760 let right_names = right_schema
2764 .iter()
2765 .map(|column| column.name.as_str())
2766 .collect::<HashSet<_>>();
2767 left_schema
2768 .iter()
2769 .filter(|left| right_names.contains(left.name.as_str()))
2770 .map(|column| column.name.clone())
2771 .collect()
2772}
2773
2774fn install_base_projection(plan: &mut LogicalPlan, projection: &Projection) {
2775 match plan {
2776 LogicalPlan::Scan {
2777 projection: scan_projection,
2778 ..
2779 } => *scan_projection = projection.clone(),
2780 LogicalPlan::Filter { input, .. } => install_base_projection(input, projection),
2781 _ => {}
2782 }
2783}
2784
2785fn is_aggregate_function(name: &str) -> bool {
2786 matches!(
2787 name.to_ascii_lowercase().as_str(),
2788 "count" | "sum" | "total" | "avg" | "min" | "max" | "group_concat" | "string_agg"
2789 )
2790}
2791
2792fn expr_key(expr: &TypedExpr) -> String {
2793 format!("{:?}", expr.kind)
2794}
2795
2796fn aggregate_signature(
2797 name: &str,
2798 distinct: bool,
2799 star: bool,
2800 arg: Option<&TypedExpr>,
2801 separator: Option<&String>,
2802 _expr: &TypedExpr,
2803) -> AggregateSignature {
2804 AggregateSignature {
2805 name: name.to_ascii_lowercase(),
2806 distinct,
2807 star,
2808 arg_key: arg.map(expr_key),
2809 separator: separator.cloned(),
2810 }
2811}
2812
2813fn build_group_key_map(group_keys: &[TypedExpr]) -> HashMap<String, usize> {
2814 let mut map = HashMap::new();
2815 for (idx, key) in group_keys.iter().enumerate() {
2816 map.insert(expr_key(key), idx);
2817 }
2818 map
2819}
2820
2821fn build_aggregate_map(aggregates: &[AggregateExpr]) -> HashMap<AggregateSignature, usize> {
2822 let mut map = HashMap::new();
2823 for (idx, agg) in aggregates.iter().enumerate() {
2824 let (name, separator, star, arg) = match &agg.function {
2825 AggregateFunction::Count => (
2826 "count".to_string(),
2827 None,
2828 agg.arg.is_none(),
2829 agg.arg.as_ref(),
2830 ),
2831 AggregateFunction::Sum => ("sum".to_string(), None, false, agg.arg.as_ref()),
2832 AggregateFunction::Total => ("total".to_string(), None, false, agg.arg.as_ref()),
2833 AggregateFunction::Avg => ("avg".to_string(), None, false, agg.arg.as_ref()),
2834 AggregateFunction::Min => ("min".to_string(), None, false, agg.arg.as_ref()),
2835 AggregateFunction::Max => ("max".to_string(), None, false, agg.arg.as_ref()),
2836 AggregateFunction::GroupConcat { separator } => (
2837 "group_concat".to_string(),
2838 separator.clone(),
2839 false,
2840 agg.arg.as_ref(),
2841 ),
2842 AggregateFunction::StringAgg { separator } => (
2843 "string_agg".to_string(),
2844 separator.clone(),
2845 false,
2846 agg.arg.as_ref(),
2847 ),
2848 };
2849 let signature = AggregateSignature {
2850 name,
2851 distinct: agg.distinct,
2852 star,
2853 arg_key: arg.map(expr_key),
2854 separator,
2855 };
2856 map.insert(signature, idx);
2857 }
2858 map
2859}
2860
2861fn build_aggregate_schema(
2862 group_keys: &[TypedExpr],
2863 aggregates: &[AggregateExpr],
2864) -> Vec<ColumnMetadata> {
2865 let mut schema = Vec::new();
2866 for (idx, key) in group_keys.iter().enumerate() {
2867 let name = match &key.kind {
2868 TypedExprKind::ColumnRef { column, .. } => column.clone(),
2869 _ => format!("group_{idx}"),
2870 };
2871 schema.push(ColumnMetadata::new(name, key.resolved_type.clone()));
2872 }
2873 for (idx, agg) in aggregates.iter().enumerate() {
2874 let name = match &agg.function {
2875 AggregateFunction::Count => format!("count_{idx}"),
2876 AggregateFunction::Sum => format!("sum_{idx}"),
2877 AggregateFunction::Total => format!("total_{idx}"),
2878 AggregateFunction::Avg => format!("avg_{idx}"),
2879 AggregateFunction::Min => format!("min_{idx}"),
2880 AggregateFunction::Max => format!("max_{idx}"),
2881 AggregateFunction::GroupConcat { .. } => format!("group_concat_{idx}"),
2882 AggregateFunction::StringAgg { .. } => format!("string_agg_{idx}"),
2883 };
2884 schema.push(ColumnMetadata::new(name, agg.result_type.clone()));
2885 }
2886 schema
2887}
2888
2889fn rewrite_expr_with_maps(
2890 expr: &TypedExpr,
2891 group_key_map: &HashMap<String, usize>,
2892 aggregate_map: &HashMap<AggregateSignature, usize>,
2893 output_names: &[String],
2894) -> Result<TypedExpr, PlannerError> {
2895 let group_key_count = output_names.len().saturating_sub(aggregate_map.len());
2896 let key = expr_key(expr);
2897 if let Some(idx) = group_key_map.get(&key) {
2898 return Ok(make_output_column_ref(
2899 *idx,
2900 output_names,
2901 expr.resolved_type.clone(),
2902 expr.span,
2903 ));
2904 }
2905
2906 match &expr.kind {
2907 TypedExprKind::FunctionCall {
2908 name,
2909 args,
2910 distinct,
2911 star,
2912 } if is_aggregate_function(name) => {
2913 let separator = if name.eq_ignore_ascii_case("group_concat") && args.len() == 2 {
2914 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
2915 Some(value.clone())
2916 } else {
2917 return Err(PlannerError::invalid_expression(
2918 "GROUP_CONCAT separator must be a string literal".to_string(),
2919 ));
2920 }
2921 } else if name.eq_ignore_ascii_case("string_agg") && args.len() == 2 {
2922 if let TypedExprKind::Literal(Literal::String(value)) = &args[1].kind {
2923 Some(value.clone())
2924 } else {
2925 return Err(PlannerError::invalid_expression(
2926 "STRING_AGG separator must be a string literal".to_string(),
2927 ));
2928 }
2929 } else {
2930 None
2931 };
2932 let signature = AggregateSignature {
2933 name: name.to_ascii_lowercase(),
2934 distinct: *distinct,
2935 star: *star,
2936 arg_key: args.first().map(expr_key),
2937 separator,
2938 };
2939 let idx = aggregate_map.get(&signature).ok_or_else(|| {
2940 PlannerError::invalid_expression(
2941 "aggregate in expression is not part of plan".to_string(),
2942 )
2943 })?;
2944 let output_index = group_key_count + idx;
2945 Ok(make_output_column_ref(
2946 output_index,
2947 output_names,
2948 expr.resolved_type.clone(),
2949 expr.span,
2950 ))
2951 }
2952 TypedExprKind::FunctionCall {
2953 name,
2954 args,
2955 distinct,
2956 star,
2957 } => {
2958 if *distinct || *star {
2959 return Err(PlannerError::invalid_expression(
2960 "DISTINCT/STAR modifiers are only supported for aggregates".to_string(),
2961 ));
2962 }
2963 let mut rewritten_args = Vec::with_capacity(args.len());
2964 for arg in args {
2965 rewritten_args.push(rewrite_expr_with_maps(
2966 arg,
2967 group_key_map,
2968 aggregate_map,
2969 output_names,
2970 )?);
2971 }
2972 Ok(TypedExpr {
2973 kind: TypedExprKind::FunctionCall {
2974 name: name.clone(),
2975 args: rewritten_args,
2976 distinct: false,
2977 star: false,
2978 },
2979 resolved_type: expr.resolved_type.clone(),
2980 span: expr.span,
2981 })
2982 }
2983 TypedExprKind::BinaryOp { left, op, right } => {
2984 let left = rewrite_expr_with_maps(left, group_key_map, aggregate_map, output_names)?;
2985 let right = rewrite_expr_with_maps(right, group_key_map, aggregate_map, output_names)?;
2986 Ok(TypedExpr {
2987 kind: TypedExprKind::BinaryOp {
2988 left: Box::new(left),
2989 op: *op,
2990 right: Box::new(right),
2991 },
2992 resolved_type: expr.resolved_type.clone(),
2993 span: expr.span,
2994 })
2995 }
2996 TypedExprKind::UnaryOp { op, operand } => {
2997 let operand =
2998 rewrite_expr_with_maps(operand, group_key_map, aggregate_map, output_names)?;
2999 Ok(TypedExpr {
3000 kind: TypedExprKind::UnaryOp {
3001 op: *op,
3002 operand: Box::new(operand),
3003 },
3004 resolved_type: expr.resolved_type.clone(),
3005 span: expr.span,
3006 })
3007 }
3008 TypedExprKind::Between {
3009 expr: inner,
3010 low,
3011 high,
3012 negated,
3013 } => {
3014 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
3015 let low = rewrite_expr_with_maps(low, group_key_map, aggregate_map, output_names)?;
3016 let high = rewrite_expr_with_maps(high, group_key_map, aggregate_map, output_names)?;
3017 Ok(TypedExpr {
3018 kind: TypedExprKind::Between {
3019 expr: Box::new(inner),
3020 low: Box::new(low),
3021 high: Box::new(high),
3022 negated: *negated,
3023 },
3024 resolved_type: expr.resolved_type.clone(),
3025 span: expr.span,
3026 })
3027 }
3028 TypedExprKind::Like {
3029 expr: inner,
3030 pattern,
3031 escape,
3032 negated,
3033 kind,
3034 } => {
3035 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
3036 let pattern =
3037 rewrite_expr_with_maps(pattern, group_key_map, aggregate_map, output_names)?;
3038 let escape = if let Some(esc) = escape {
3039 Some(Box::new(rewrite_expr_with_maps(
3040 esc,
3041 group_key_map,
3042 aggregate_map,
3043 output_names,
3044 )?))
3045 } else {
3046 None
3047 };
3048 Ok(TypedExpr {
3049 kind: TypedExprKind::Like {
3050 expr: Box::new(inner),
3051 pattern: Box::new(pattern),
3052 escape,
3053 negated: *negated,
3054 kind: *kind,
3055 },
3056 resolved_type: expr.resolved_type.clone(),
3057 span: expr.span,
3058 })
3059 }
3060 TypedExprKind::InList {
3061 expr: inner,
3062 list,
3063 negated,
3064 } => {
3065 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
3066 let mut rewritten_list = Vec::with_capacity(list.len());
3067 for item in list {
3068 rewritten_list.push(rewrite_expr_with_maps(
3069 item,
3070 group_key_map,
3071 aggregate_map,
3072 output_names,
3073 )?);
3074 }
3075 Ok(TypedExpr {
3076 kind: TypedExprKind::InList {
3077 expr: Box::new(inner),
3078 list: rewritten_list,
3079 negated: *negated,
3080 },
3081 resolved_type: expr.resolved_type.clone(),
3082 span: expr.span,
3083 })
3084 }
3085 TypedExprKind::IsNull {
3086 expr: inner,
3087 negated,
3088 } => {
3089 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
3090 Ok(TypedExpr {
3091 kind: TypedExprKind::IsNull {
3092 expr: Box::new(inner),
3093 negated: *negated,
3094 },
3095 resolved_type: expr.resolved_type.clone(),
3096 span: expr.span,
3097 })
3098 }
3099 TypedExprKind::Literal(_) | TypedExprKind::VectorLiteral(_) => Ok(expr.clone()),
3100 TypedExprKind::ColumnRef { .. } => Err(PlannerError::invalid_expression(
3101 "column reference must appear in GROUP BY or be aggregated".to_string(),
3102 )),
3103 TypedExprKind::Cast {
3104 expr: inner,
3105 target_type,
3106 } => {
3107 let inner = rewrite_expr_with_maps(inner, group_key_map, aggregate_map, output_names)?;
3108 Ok(TypedExpr {
3109 kind: TypedExprKind::Cast {
3110 expr: Box::new(inner),
3111 target_type: target_type.clone(),
3112 },
3113 resolved_type: expr.resolved_type.clone(),
3114 span: expr.span,
3115 })
3116 }
3117 TypedExprKind::ScalarSubquery(_)
3118 | TypedExprKind::InSubquery { .. }
3119 | TypedExprKind::Exists { .. }
3120 | TypedExprKind::Quantified { .. } => Ok(expr.clone()),
3121 }
3122}
3123
3124fn make_output_column_ref(
3125 index: usize,
3126 output_names: &[String],
3127 resolved_type: ResolvedType,
3128 span: crate::ast::Span,
3129) -> TypedExpr {
3130 let name = output_names
3131 .get(index)
3132 .cloned()
3133 .unwrap_or_else(|| format!("col_{index}"));
3134 TypedExpr::column_ref("__agg__".to_string(), name, index, resolved_type, span)
3135}