1use rudb_catalog::{Catalog, Entry, QualifiedName, same_name};
16use rudb_common::{
17 Error, Field, LogicalType, Result, Semantics, Session, ShowBehavior, Span, Stat, Value,
18};
19use rudb_functions::{
20 Columns, FILE_ROW_NUMBER, FunctionKind, Given, Resolved, TableFunction, csv_fields, csv_given,
21 files, is_file, is_pattern, kind_of, parquet_footers, resolve, resolve_pragma, resolve_table,
22};
23use rudb_parse::ast::{self, Ast, Distinct, LiteralKind, Nulls, Order, Quantifier, SetOp};
24use rudb_parse::{NONE, identifier_parts, parse_ast_with_case};
25use rudb_plan::{
26 BuildSide, ColumnBinding, Expr, ExprRef, JoinKind, Node, NodeRef, Plan, SetOpKind, SortKey,
27 WindowBound, WindowExclude, WindowFrame, WindowUnit,
28};
29
30use crate::expr::{describe, has_aggregate};
31use crate::parameters::Parameters;
32use crate::scope::{Scope, Visible};
33
34pub fn bind(ast: &Ast, catalog: &Catalog) -> Result<Plan> {
41 bind_with(ast, catalog, &Parameters::new(), &Session::new())
42}
43
44pub fn bind_with(
53 ast: &Ast,
54 catalog: &Catalog,
55 parameters: &Parameters,
56 session: &Session,
57) -> Result<Plan> {
58 let query = match ast.statements.as_slice() {
59 [ast::Statement::Query(query)] => *query,
60 [] => return Err(Error::binder("no statement to bind")),
61 [_] => return Err(Error::not_implemented("a statement that is not a query")),
64 _ => return Err(Error::not_implemented("a script of more than one statement")),
65 };
66 let mut binder = Binder::with(catalog, parameters, session);
67 let (root, _) = binder.bind_query(ast, query)?;
68 let mut plan = binder.into_plan();
69 plan.set_root(root);
70 plan.validate()?;
71 Ok(plan)
72}
73
74pub fn bind_sql(query: &str, catalog: &Catalog) -> Result<Plan> {
80 bind_sql_with(query, catalog, &Session::new())
81}
82
83pub fn bind_sql_with(query: &str, catalog: &Catalog, session: &Session) -> Result<Plan> {
89 let ast = parse_ast_with_case(query, session.semantics().identifier_case())?;
90 bind_with(&ast, catalog, &Parameters::new(), session)
91}
92
93#[derive(Debug)]
95pub(crate) struct Aggregation {
96 pub(crate) index: u32,
98 pub(crate) groups: Vec<ExprRef>,
100 pub(crate) aggregates: Vec<ExprRef>,
102}
103
104#[derive(Debug)]
112pub(crate) struct WindowRun {
113 index: u32,
115 partition: Vec<ExprRef>,
117 order: Vec<SortKey>,
119 frame: WindowFrame,
121 calls: Vec<ExprRef>,
123}
124
125pub(crate) struct WindowCall<'a> {
130 pub(crate) name: &'a str,
132 pub(crate) args: &'a [ast::ExprRef],
134 pub(crate) distinct: bool,
136 pub(crate) filter: ast::ExprRef,
138 pub(crate) ignore_nulls: bool,
140 pub(crate) spec: ast::WindowRef,
142}
143
144struct WindowParts {
146 args: Vec<ExprRef>,
148 partition: Vec<ExprRef>,
150 order: Vec<SortKey>,
152 frame: WindowFrame,
154}
155
156#[derive(Debug)]
163struct Read {
164 fields: Vec<Field>,
166 rows: Stat<u64>,
168 distincts: Vec<(String, u64)>,
170}
171
172impl Read {
173 fn uncounted(fields: Vec<Field>) -> Self {
175 Self { fields, rows: Stat::Unknown, distincts: Vec::new() }
176 }
177}
178
179#[derive(Debug)]
181struct Materialized {
182 written: u32,
184 cte: u32,
186 name: String,
188 fields: Vec<Field>,
190}
191
192#[derive(Debug)]
193pub(crate) struct PendingSubquery {
194 pub(crate) node: NodeRef,
195 pub(crate) kind: JoinKind,
196 pub(crate) conditions: Vec<ExprRef>,
197 pub(crate) dependent: bool,
198 pub(crate) index: u32,
204}
205
206#[derive(Debug)]
208pub(crate) struct Binder<'a> {
209 catalog: &'a Catalog,
210 pub(crate) parameters: &'a Parameters,
212 pub(crate) session: &'a Session,
214 pub(crate) semantics: Semantics,
216 plan: Plan,
217 next_index: u32,
218 pub(crate) current_span: Span,
220 pub(crate) aggregation: Option<Aggregation>,
222 pub(crate) in_aggregate: bool,
224 pub(crate) in_filter: bool,
226 pub(crate) windows: Vec<WindowRun>,
228 pub(crate) in_window: bool,
230 pub(crate) scalar_subqueries: Vec<PendingSubquery>,
232 pub(crate) joined_above: Vec<u32>,
238 pub(crate) outer_scopes: Vec<Scope>,
239 pub(crate) lateral_scopes: Vec<usize>,
246 pub(crate) correlations: Vec<Vec<ColumnBinding>>,
247 pub(crate) clause: &'static str,
249 expanding: Vec<String>,
251 materialized: Vec<Materialized>,
257 next_cte: u32,
259 started: Option<i64>,
261}
262
263impl<'a> Binder<'a> {
264 pub(crate) fn with(
265 catalog: &'a Catalog,
266 parameters: &'a Parameters,
267 session: &'a Session,
268 ) -> Self {
269 Self {
270 catalog,
271 parameters,
272 session,
273 semantics: session.semantics(),
274 plan: Plan::new(),
275 next_index: 0,
276 current_span: Span::new(0, 0),
277 aggregation: None,
278 in_aggregate: false,
279 in_filter: false,
280 windows: Vec::new(),
281 in_window: false,
282 scalar_subqueries: Vec::new(),
283 joined_above: Vec::new(),
284 outer_scopes: Vec::new(),
285 lateral_scopes: Vec::new(),
286 correlations: Vec::new(),
287 clause: "SELECT clause",
288 expanding: Vec::new(),
289 materialized: Vec::new(),
290 next_cte: 0,
291 started: None,
292 }
293 }
294
295 pub(crate) fn catalog(&self) -> &Catalog {
296 self.catalog
297 }
298
299 pub(crate) fn instant(&mut self) -> i64 {
306 *self.started.get_or_insert_with(crate::context::micros_now)
307 }
308
309 pub(crate) fn plan(&self) -> &Plan {
310 &self.plan
311 }
312
313 pub(crate) fn plan_mut(&mut self) -> &mut Plan {
314 &mut self.plan
315 }
316
317 pub(crate) fn add_expr(&mut self, expr: Expr, ty: LogicalType) -> ExprRef {
318 self.plan.add_expr_at(expr, ty, self.current_span)
319 }
320
321 pub(crate) fn add_constant(&mut self, value: Value) -> ExprRef {
322 let ty = value.logical_type();
323 let reference = self.plan.add_value(value);
324 self.plan.add_expr_at(Expr::Constant(reference), ty, self.current_span)
325 }
326
327 pub(crate) fn add_node(&mut self, node: Node) -> NodeRef {
328 self.plan.add_node_at(node, self.current_span)
329 }
330
331 pub(crate) fn into_plan(self) -> Plan {
332 self.plan
333 }
334
335 pub(crate) fn fresh_index(&mut self) -> u32 {
337 let index = self.next_index;
338 self.next_index += 1;
339 index
340 }
341
342 fn column(&mut self, index: u32, position: usize, ty: LogicalType) -> ExprRef {
344 let binding = ColumnBinding::new(index, position as u32);
345 self.plan.add_expr(Expr::Column(binding), ty)
346 }
347
348 fn attach_scalar_subqueries(&mut self, mut input: NodeRef) -> NodeRef {
350 let subqueries = std::mem::take(&mut self.scalar_subqueries);
351 for pending in subqueries {
352 let PendingSubquery { node: mut right, kind, conditions, dependent, index: _ } =
353 pending;
354 if kind == JoinKind::Single && !self.semantics.scalar_subquery_error_on_multiple_rows()
355 {
356 right = self.add_node(Node::Limit { input: right, count: Some(1), offset: 0 });
357 }
358 let conditions = self.plan.add_expr_list(&conditions);
359 input = if dependent {
360 self.add_node(Node::DependentJoin { left: input, right, kind, conditions })
361 } else {
362 self.add_node(Node::Join {
363 left: input,
364 right,
365 kind,
366 conditions,
367 build: BuildSide::default(),
368 })
369 };
370 }
371 input
372 }
373
374 pub(crate) fn bind_query(
377 &mut self,
378 ast: &Ast,
379 query: ast::QueryRef,
380 ) -> Result<(NodeRef, Scope)> {
381 let span = ast.query_span(query);
382 let outer = std::mem::replace(&mut self.current_span, span);
383 let result =
384 self.bind_query_inner(ast, query).map_err(|error| error.with_fallback_span(span));
385 self.current_span = outer;
386 result
387 }
388
389 fn bind_query_inner(&mut self, ast: &Ast, query: ast::QueryRef) -> Result<(NodeRef, Scope)> {
390 let written = ast.query(query);
391 if written.ctes.is_empty() {
392 return self.bind_body(ast, &written);
393 }
394 let depth = self.materialized.len();
398 let result = self.bind_materialized(ast, &written);
399 self.materialized.truncate(depth);
400 result
401 }
402
403 fn bind_materialized(&mut self, ast: &Ast, written: &ast::Query) -> Result<(NodeRef, Scope)> {
409 let depth = self.materialized.len();
410 let held = ast.cte_list(written.ctes).to_vec();
411 let mut definitions = Vec::with_capacity(held.len());
412 for &index in &held {
413 definitions.push(self.bind_definition(ast, index)?);
414 }
415 let (mut node, scope) = self.bind_body(ast, written)?;
416 for (at, definition) in definitions.into_iter().enumerate().rev() {
417 let entry = &self.materialized[depth + at];
418 let cte = entry.cte;
419 let name = entry.name.clone();
420 let fields = entry.fields.clone();
421 let name = self.plan.intern(&name);
422 let columns = self.plan.add_fields(&fields);
423 node =
424 self.add_node(Node::MaterializedCte { definition, body: node, name, cte, columns });
425 }
426 Ok((node, scope))
427 }
428
429 fn bind_definition(&mut self, ast: &Ast, index: u32) -> Result<NodeRef> {
439 let held = ast.cte(index);
440 let name = ast.string(held.name).to_string();
441 let (node, mut scope) = self.bind_query(ast, held.query)?;
442 if !held.columns.is_empty() {
443 let names: Vec<&str> = ast.name(held.columns).collect();
444 scope.rename_prefix(&names);
445 }
446 let table = self.fresh_index();
447 let mut exprs = Vec::with_capacity(scope.len());
448 let mut names = Vec::with_capacity(scope.len());
449 for column in &scope.columns {
450 exprs.push(self.plan.add_expr(Expr::Column(column.binding), column.ty.clone()));
451 names.push(self.plan.intern(&column.name));
452 }
453 let exprs = self.plan.add_expr_list(&exprs);
454 let names = self.plan.add_name_list(&names);
455 let node = self.add_node(Node::Project { input: node, index: table, exprs, names });
456 let cte = self.next_cte;
457 self.next_cte += 1;
458 self.materialized.push(Materialized { written: index, cte, name, fields: scope.fields() });
459 Ok(node)
460 }
461
462 fn bind_body(&mut self, ast: &Ast, written: &ast::Query) -> Result<(NodeRef, Scope)> {
463 match written.body {
464 ast::QueryBody::Select(select) => self.bind_select(ast, select, written),
465 ast::QueryBody::SetOp { op, quantifier, by_name, left, right } => {
466 if by_name {
467 return Err(Error::not_implemented("UNION BY NAME"));
468 }
469 self.bind_set_op(ast, written, op, quantifier, left, right)
470 }
471 ast::QueryBody::Values(rows) => self.bind_values(ast, written, rows),
472 ast::QueryBody::Describe(inner) => self.bind_describe(ast, written, inner),
473 ast::QueryBody::Show { name, relation } => self.bind_show(ast, written, name, relation),
474 }
475 }
476
477 fn bind_show(
479 &mut self,
480 ast: &Ast,
481 query: &ast::Query,
482 name: ast::Slice,
483 relation: ast::QueryRef,
484 ) -> Result<(NodeRef, Scope)> {
485 let text = ast.name_text(name);
486 let parts: Vec<&str> = ast.name(name).collect();
487 let table_exists = self.catalog.resolve(&parts).is_ok();
488 let as_table = match self.semantics.show_behavior() {
489 ShowBehavior::Auto => table_exists,
490 ShowBehavior::Setting => false,
491 ShowBehavior::Table => true,
492 };
493 if as_table {
494 return self.bind_describe(ast, query, relation);
495 }
496 let Some((_, value)) =
497 self.session.iter().find(|(name, _)| name.eq_ignore_ascii_case(&text))
498 else {
499 return Err(Error::catalog(format!("Setting with name \"{text}\" does not exist")));
500 };
501 let field = Field::new(text, LogicalType::Varchar);
502 let expr = self.plan.add_constant(Value::Varchar(value.to_string()));
503 let row = self.plan.add_expr_list(&[expr]);
504 let rows = self.plan.add_rows(&[row]);
505 let columns = self.plan.add_fields(std::slice::from_ref(&field));
506 let index = self.fresh_index();
507 let node = self.add_node(Node::Values { index, columns, rows });
508 let mut scope = Scope::empty();
509 scope.push(Visible {
510 table: String::new(),
511 name: field.name,
512 binding: ColumnBinding::new(index, 0),
513 ty: LogicalType::Varchar,
514 not_null: false,
515 });
516 Ok((node, scope))
517 }
518
519 fn bind_describe(
535 &mut self,
536 ast: &Ast,
537 query: &ast::Query,
538 inner: ast::QueryRef,
539 ) -> Result<(NodeRef, Scope)> {
540 let (_, described) = self.bind_query(ast, inner)?;
541 let fields: Vec<Field> = ["column_name", "column_type", "null", "key", "default", "extra"]
542 .iter()
543 .map(|name| Field::new(*name, LogicalType::Varchar))
544 .collect();
545 let mut slices = Vec::with_capacity(described.columns.len());
546 for column in described.columns.clone() {
547 let written = [
550 column.name.clone(),
551 column.ty.to_string(),
552 if column.not_null { "NO" } else { "YES" }.to_owned(),
553 ];
554 let mut items: Vec<ExprRef> = written
555 .into_iter()
556 .map(|text| self.plan.add_constant(Value::Varchar(text)))
557 .collect();
558 for _ in 0..3 {
559 let empty = self.plan.add_constant(Value::Null);
560 items.push(self.cast_to(empty, &LogicalType::Varchar));
561 }
562 slices.push(self.plan.add_expr_list(&items));
563 }
564 let rows = self.plan.add_rows(&slices);
565 let columns = self.plan.add_fields(&fields);
566 let index = self.fresh_index();
567 let mut node = self.add_node(Node::Values { index, columns, rows });
568 let mut scope = Scope::empty();
569 for (at, field) in fields.iter().enumerate() {
570 scope.push(Visible {
571 table: String::new(),
572 name: field.name.clone(),
573 binding: ColumnBinding::new(index, at as u32),
574 ty: field.ty.clone(),
575 not_null: false,
576 });
577 }
578 let keys = self.sort_keys(ast, query, &scope, &[])?;
579 if !keys.is_empty() {
580 let keys = self.plan.add_sort_keys(&keys);
581 node = self.add_node(Node::Sort { input: node, keys });
582 }
583 node = self.apply_limit(ast, query, node)?;
584 Ok((node, scope))
585 }
586
587 fn passes_through(&self, expr: ExprRef, input: &Scope) -> bool {
593 let Expr::Column(binding) = *self.plan.expr(expr) else { return false };
594 input.columns.iter().any(|column| column.binding == binding && column.not_null)
595 }
596
597 fn bind_values(
604 &mut self,
605 ast: &Ast,
606 query: &ast::Query,
607 rows: ast::Slice,
608 ) -> Result<(NodeRef, Scope)> {
609 let written = ast.rows(rows).to_vec();
610 let Some(first) = written.first() else {
611 return Err(Error::binder("VALUES needs at least one row"));
612 };
613 let width = first.len as usize;
614 for (at, row) in written.iter().enumerate() {
615 if row.len as usize != width {
616 return Err(Error::binder(format!(
617 "VALUES lists must all be the same length, expected {width} columns but row {} has {}",
618 at + 1,
619 row.len
620 )));
621 }
622 }
623 let empty = Scope::empty();
625 let previous = std::mem::replace(&mut self.clause, "VALUES clause");
626 let mut bound: Vec<Vec<ExprRef>> = Vec::with_capacity(written.len());
627 for row in &written {
628 let mut items = Vec::with_capacity(width);
629 for &expr in ast.expr_list(*row) {
630 items.push(self.bind_expr(ast, expr, &empty)?);
631 }
632 bound.push(items);
633 }
634 self.clause = previous;
635 let mut types = Vec::with_capacity(width);
636 for at in 0..width {
637 let mut ty = self.plan.expr_type(bound[0][at]).clone();
638 for row in &bound[1..] {
639 let other = self.plan.expr_type(row[at]).clone();
640 ty = ty.promote(&other).ok_or_else(|| {
641 Error::binder(format!(
642 "Cannot combine a value of type {ty} with a value of type {other} in column {} of a VALUES",
643 at + 1
644 ))
645 })?;
646 }
647 types.push(ty);
648 }
649 let mut slices = Vec::with_capacity(bound.len());
650 for row in &bound {
651 let items: Vec<ExprRef> = row
652 .iter()
653 .zip(&types)
654 .map(|(&expr, ty)| self.checked_cast_to(expr, ty, false))
655 .collect::<Result<_>>()?;
656 slices.push(self.plan.add_expr_list(&items));
657 }
658 let rows = self.plan.add_rows(&slices);
659 let fields: Vec<Field> = types
660 .iter()
661 .enumerate()
662 .map(|(at, ty)| Field::new(format!("col{at}"), ty.clone()))
663 .collect();
664 let columns = self.plan.add_fields(&fields);
665 let index = self.fresh_index();
666 let mut node = self.add_node(Node::Values { index, columns, rows });
667 let mut scope = Scope::empty();
668 for (at, field) in fields.iter().enumerate() {
669 scope.push(Visible {
670 table: String::new(),
671 name: field.name.clone(),
672 binding: ColumnBinding::new(index, at as u32),
673 ty: field.ty.clone(),
674 not_null: false,
675 });
676 }
677 let keys = self.sort_keys(ast, query, &scope, &[])?;
678 if !keys.is_empty() {
679 let keys = self.plan.add_sort_keys(&keys);
680 node = self.add_node(Node::Sort { input: node, keys });
681 }
682 node = self.apply_limit(ast, query, node)?;
683 Ok((node, scope))
684 }
685
686 fn bind_set_op(
687 &mut self,
688 ast: &Ast,
689 query: &ast::Query,
690 op: SetOp,
691 quantifier: Quantifier,
692 left: ast::QueryRef,
693 right: ast::QueryRef,
694 ) -> Result<(NodeRef, Scope)> {
695 let (left_node, left_scope) = self.bind_query(ast, left)?;
696 let (right_node, right_scope) = self.bind_query(ast, right)?;
697 if left_scope.len() != right_scope.len() {
698 return Err(Error::binder(format!(
699 "Set operations can only apply to expressions with the same number of result columns, but left side has {} and right side has {}",
700 left_scope.len(),
701 right_scope.len()
702 )));
703 }
704 let mut types = Vec::with_capacity(left_scope.len());
706 for (left, right) in left_scope.columns.iter().zip(&right_scope.columns) {
707 let common = left.ty.promote(&right.ty).ok_or_else(|| {
708 Error::binder(format!(
709 "Cannot combine a column of type {} with a column of type {} in a set operation",
710 left.ty, right.ty
711 ))
712 })?;
713 types.push(common);
714 }
715 let left_node = self.conform(left_node, &left_scope, &types)?;
716 let right_node = self.conform(right_node, &right_scope, &types)?;
717 let index = self.fresh_index();
718 let kind = match op {
719 SetOp::Union => SetOpKind::Union,
720 SetOp::Except => SetOpKind::Except,
721 SetOp::Intersect => SetOpKind::Intersect,
722 };
723 let all = quantifier == Quantifier::All;
726 let mut node =
727 self.add_node(Node::SetOp { left: left_node, right: right_node, kind, all, index });
728 let mut scope = Scope::empty();
729 for (at, (column, ty)) in left_scope.columns.iter().zip(&types).enumerate() {
730 scope.push(Visible {
731 table: String::new(),
732 name: column.name.clone(),
733 binding: ColumnBinding::new(index, at as u32),
734 ty: ty.clone(),
735 not_null: false,
738 });
739 }
740 let keys = self.sort_keys(ast, query, &scope, &[])?;
744 if !keys.is_empty() {
745 let keys = self.plan.add_sort_keys(&keys);
746 node = self.add_node(Node::Sort { input: node, keys });
747 }
748 node = self.apply_limit(ast, query, node)?;
749 Ok((node, scope))
750 }
751
752 fn conform(&mut self, node: NodeRef, scope: &Scope, types: &[LogicalType]) -> Result<NodeRef> {
754 if scope.columns.iter().zip(types).all(|(column, ty)| &column.ty == ty) {
755 return Ok(node);
756 }
757 let index = self.fresh_index();
758 let mut exprs = Vec::with_capacity(types.len());
759 let mut names = Vec::with_capacity(types.len());
760 for (column, ty) in scope.columns.iter().zip(types) {
761 let expr = self.plan.add_expr(Expr::Column(column.binding), column.ty.clone());
762 exprs.push(self.checked_cast_to(expr, ty, false)?);
763 names.push(self.plan.intern(&column.name));
764 }
765 let exprs = self.plan.add_expr_list(&exprs);
766 let names = self.plan.add_name_list(&names);
767 Ok(self.add_node(Node::Project { input: node, index, exprs, names }))
768 }
769
770 fn bind_select(
773 &mut self,
774 ast: &Ast,
775 select: ast::SelectRef,
776 query: &ast::Query,
777 ) -> Result<(NodeRef, Scope)> {
778 let written = ast.select(select);
779 let outer_windows = std::mem::take(&mut self.windows);
783 let (mut node, input) = self.bind_from(ast, written.from)?;
784 node = self.attach_scalar_subqueries(node);
785
786 if written.filter != NONE {
787 self.clause = "WHERE clause";
788 let predicate = self.bind_expr(ast, written.filter, &input)?;
789 let predicate = self.as_boolean(predicate, "WHERE")?;
790 node = self.attach_scalar_subqueries(node);
791 node = self.add_node(Node::Filter { input: node, predicate });
792 }
793
794 let targets = ast.target_list(written.targets).to_vec();
795 if targets.is_empty() {
796 return Err(Error::binder("a SELECT needs at least one expression to select"));
797 }
798
799 let group_items = self.group_items(ast, &written, &targets)?;
800 let aggregating = !group_items.is_empty()
801 || written.having != NONE
802 || targets.iter().any(|target| has_aggregate(ast, target.expr));
803 if aggregating {
804 self.clause = "GROUP BY clause";
805 let mut groups = Vec::with_capacity(group_items.len());
806 for item in &group_items {
807 groups.push(self.bind_expr(ast, *item, &input)?);
808 }
809 let index = self.fresh_index();
810 self.aggregation = Some(Aggregation { index, groups, aggregates: Vec::new() });
811 }
812
813 self.clause = "SELECT clause";
814 let (mut exprs, mut names) = self.bind_targets(ast, &targets, &input)?;
815 let visible = exprs.len();
816
817 let mut having = None;
818 let mut above = Vec::new();
825 if written.having != NONE {
826 self.clause = "HAVING clause";
827 let before = self.scalar_subqueries.len();
828 let predicate = self.bind_expr(ast, written.having, &input)?;
829 for pending in self.scalar_subqueries.split_off(before) {
832 if pending.dependent {
833 self.scalar_subqueries.push(pending);
834 } else {
835 above.push(pending);
836 }
837 }
838 self.joined_above = above.iter().map(|pending| pending.index).collect();
839 let predicate = self.over_aggregate(predicate, &input)?;
840 let mut rewritten = Vec::with_capacity(above.len());
843 for mut pending in above {
844 let conditions = std::mem::take(&mut pending.conditions);
845 let mut over = Vec::with_capacity(conditions.len());
846 for condition in conditions {
847 over.push(self.over_aggregate(condition, &input)?);
848 }
849 pending.conditions = over;
850 rewritten.push(pending);
851 }
852 above = rewritten;
853 self.joined_above.clear();
854 having = Some(self.as_boolean(predicate, "HAVING")?);
855 }
856
857 let project = self.fresh_index();
860 let mut output = Scope::empty();
861 for (at, (expr, name)) in exprs.iter().zip(&names).enumerate() {
862 output.push(Visible {
863 table: String::new(),
864 name: name.clone(),
865 binding: ColumnBinding::new(project, at as u32),
866 ty: self.plan.expr_type(*expr).clone(),
867 not_null: self.passes_through(*expr, &input),
868 });
869 }
870
871 self.clause = "ORDER BY clause";
872 let mut extra = Vec::new();
873 let keys = self.select_sort_keys(
874 ast, query, &input, &output, project, &mut exprs, &mut names, &mut extra,
875 )?;
876 if !extra.is_empty() && written.distinct != Distinct::No {
877 return Err(Error::binder(
878 "For SELECT DISTINCT, ORDER BY expressions must appear in the select list",
879 ));
880 }
881 let on = self.distinct_on(ast, written.distinct, &output)?;
882
883 node = self.attach_scalar_subqueries(node);
884
885 if let Some(aggregation) = self.aggregation.take() {
886 let index = aggregation.index;
887 let groups = self.plan.add_expr_list(&aggregation.groups);
888 let aggregates = self.plan.add_expr_list(&aggregation.aggregates);
889 node = self.add_node(Node::Aggregate { input: node, index, groups, aggregates });
890 }
891 if !above.is_empty() {
892 debug_assert!(self.scalar_subqueries.is_empty(), "a query is waiting to be joined");
893 self.scalar_subqueries = above;
894 node = self.attach_scalar_subqueries(node);
895 }
896 if let Some(predicate) = having {
897 node = self.add_node(Node::Filter { input: node, predicate });
898 }
899
900 for run in std::mem::replace(&mut self.windows, outer_windows) {
904 let partition = self.plan.add_expr_list(&run.partition);
905 let order = self.plan.add_sort_keys(&run.order);
906 let expressions = self.plan.add_expr_list(&run.calls);
907 node = self.add_node(Node::Window {
908 input: node,
909 index: run.index,
910 partition,
911 order,
912 frame: run.frame,
913 expressions,
914 });
915 }
916
917 let interned: Vec<u32> = names.iter().map(|name| self.plan.intern(name)).collect();
918 let exprs_slice = self.plan.add_expr_list(&exprs);
919 let names_slice = self.plan.add_name_list(&interned);
920 node = self.add_node(Node::Project {
921 input: node,
922 index: project,
923 exprs: exprs_slice,
924 names: names_slice,
925 });
926
927 if written.distinct != Distinct::No {
928 let on = self.plan.add_expr_list(&on);
929 node = self.add_node(Node::Distinct { input: node, on });
930 }
931 if !keys.is_empty() {
932 let keys = self.plan.add_sort_keys(&keys);
933 node = self.add_node(Node::Sort { input: node, keys });
934 }
935 node = self.apply_limit(ast, query, node)?;
936
937 if extra.is_empty() {
938 output.columns.truncate(visible);
939 return Ok((node, output));
940 }
941 let index = self.fresh_index();
944 let mut kept = Vec::with_capacity(visible);
945 let mut kept_names = Vec::with_capacity(visible);
946 let mut scope = Scope::empty();
947 for (at, name) in names.iter().enumerate().take(visible) {
948 let ty = output.columns[at].ty.clone();
949 kept.push(self.column(project, at, ty.clone()));
950 kept_names.push(self.plan.intern(name));
951 scope.push(Visible {
952 table: String::new(),
953 name: name.clone(),
954 binding: ColumnBinding::new(index, at as u32),
955 ty,
956 not_null: output.columns[at].not_null,
957 });
958 }
959 let exprs = self.plan.add_expr_list(&kept);
960 let names = self.plan.add_name_list(&kept_names);
961 node = self.add_node(Node::Project { input: node, index, exprs, names });
962 Ok((node, scope))
963 }
964
965 fn bind_targets(
967 &mut self,
968 ast: &Ast,
969 targets: &[ast::Target],
970 input: &Scope,
971 ) -> Result<(Vec<ExprRef>, Vec<String>)> {
972 let mut exprs = Vec::with_capacity(targets.len());
973 let mut names = Vec::with_capacity(targets.len());
974 for target in targets {
975 if let ast::Expr::Star { qualifier, replacements } = ast.expr(target.expr) {
976 let table = ast.name(qualifier).last().map(str::to_string);
977 let expanded: Vec<Visible> =
978 input.star(table.as_deref())?.into_iter().cloned().collect();
979 let replacements = ast.target_list(replacements).to_vec();
980 let mut used = vec![false; replacements.len()];
981 for column in expanded {
982 let found = replacements.iter().zip(&mut used).find(|(replacement, _)| {
983 same_name(ast.string(replacement.alias), &column.name)
984 });
985 let (expr, name) = match found {
990 Some((replacement, used)) => {
991 *used = true;
992 let expr = self.bind_expr(ast, replacement.expr, input)?;
993 (expr, ast.string(replacement.alias).to_string())
994 }
995 None => (
996 self.plan.add_expr(Expr::Column(column.binding), column.ty),
997 column.name,
998 ),
999 };
1000 exprs.push(self.over_aggregate(expr, input)?);
1001 names.push(name);
1002 }
1003 if let Some((replacement, _)) =
1007 replacements.iter().zip(&used).find(|(_, used)| !**used)
1008 {
1009 return Err(missing_replacement(ast.string(replacement.alias), input));
1010 }
1011 continue;
1012 }
1013 let expr = self.bind_expr(ast, target.expr, input)?;
1014 exprs.push(self.over_aggregate(expr, input)?);
1015 names.push(if target.alias == NONE {
1016 self.output_name(ast, target.expr, input)
1017 } else {
1018 ast.string(target.alias).to_string()
1019 });
1020 }
1021 Ok((exprs, names))
1022 }
1023
1024 fn output_name(&self, ast: &Ast, target: ast::ExprRef, input: &Scope) -> String {
1030 if let ast::Expr::Column { name } = ast.expr(target) {
1031 let parts: Vec<&str> = ast.name(name).collect();
1032 if let Ok(found) = input.resolve(&parts) {
1033 return found.name.clone();
1034 }
1035 }
1036 describe(ast, target, self.semantics)
1037 }
1038
1039 fn group_items(
1041 &self,
1042 ast: &Ast,
1043 select: &ast::Select,
1044 targets: &[ast::Target],
1045 ) -> Result<Vec<ast::ExprRef>> {
1046 if select.group_by_all {
1047 return Ok(targets
1050 .iter()
1051 .filter(|target| !has_aggregate(ast, target.expr))
1052 .map(|target| target.expr)
1053 .collect());
1054 }
1055 let mut items = Vec::new();
1056 for &item in ast.expr_list(select.group_by) {
1057 items.push(self.output_reference(ast, item, targets, "GROUP BY")?.unwrap_or(item));
1058 }
1059 Ok(items)
1060 }
1061
1062 fn output_reference(
1064 &self,
1065 ast: &Ast,
1066 item: ast::ExprRef,
1067 targets: &[ast::Target],
1068 clause: &str,
1069 ) -> Result<Option<ast::ExprRef>> {
1070 match ast.expr(item) {
1071 ast::Expr::Literal { kind: LiteralKind::Number, text } => {
1072 let written = ast.string(text);
1073 let position: usize = written.parse().map_err(|_| {
1074 Error::binder(format!("{clause} term {written} is not a column"))
1075 })?;
1076 if position == 0 || position > targets.len() {
1077 return Err(Error::binder(format!(
1078 "{clause} term out of range - should be between 1 and {}",
1079 targets.len()
1080 )));
1081 }
1082 Ok(Some(targets[position - 1].expr))
1083 }
1084 ast::Expr::Column { name } => {
1085 let parts: Vec<&str> = ast.name(name).collect();
1086 let [written] = parts.as_slice() else { return Ok(None) };
1087 let mut found = None;
1088 for target in targets {
1089 if target.alias != NONE && same_name(ast.string(target.alias), written) {
1090 if found.is_some() {
1091 return Ok(None);
1092 }
1093 found = Some(target.expr);
1094 }
1095 }
1096 Ok(found)
1097 }
1098 _ => Ok(None),
1099 }
1100 }
1101
1102 #[allow(clippy::too_many_arguments)]
1106 fn select_sort_keys(
1107 &mut self,
1108 ast: &Ast,
1109 query: &ast::Query,
1110 input: &Scope,
1111 output: &Scope,
1112 project: u32,
1113 exprs: &mut Vec<ExprRef>,
1114 names: &mut Vec<String>,
1115 extra: &mut Vec<usize>,
1116 ) -> Result<Vec<SortKey>> {
1117 if query.order_by_all {
1118 return Ok(self.every_column(output));
1119 }
1120 let items = ast.order_list(query.order_by).to_vec();
1121 let mut keys = Vec::with_capacity(items.len());
1122 for item in items {
1123 self.check_order_literal(ast, item.expr)?;
1124 let position = match self.output_position(ast, item.expr, output)? {
1125 Some(position) => position,
1126 None => {
1127 let bound = self.bind_expr(ast, item.expr, input)?;
1128 let bound = self.over_aggregate(bound, input)?;
1129 match exprs.iter().position(|&held| self.same_expr(held, bound)) {
1130 Some(position) => position,
1131 None => {
1132 exprs.push(bound);
1133 names.push(describe(ast, item.expr, self.semantics));
1134 extra.push(exprs.len() - 1);
1135 exprs.len() - 1
1136 }
1137 }
1138 }
1139 };
1140 let ty = self.plan.expr_type(exprs[position]).clone();
1141 let expr = self.column(project, position, ty);
1142 keys.push(self.sort_key(expr, item));
1143 }
1144 Ok(keys)
1145 }
1146
1147 fn sort_keys(
1149 &mut self,
1150 ast: &Ast,
1151 query: &ast::Query,
1152 output: &Scope,
1153 targets: &[ast::Target],
1154 ) -> Result<Vec<SortKey>> {
1155 if query.order_by_all {
1156 return Ok(self.every_column(output));
1157 }
1158 let items = ast.order_list(query.order_by).to_vec();
1159 let mut keys = Vec::with_capacity(items.len());
1160 for item in items {
1161 self.check_order_literal(ast, item.expr)?;
1162 let expr = match self.output_position(ast, item.expr, output)? {
1163 Some(position) => {
1164 let column = &output.columns[position];
1165 let (binding, ty) = (column.binding, column.ty.clone());
1166 self.plan.add_expr(Expr::Column(binding), ty)
1167 }
1168 None => {
1169 let _ = targets;
1170 self.bind_expr(ast, item.expr, output)?
1171 }
1172 };
1173 keys.push(self.sort_key(expr, item));
1174 }
1175 Ok(keys)
1176 }
1177
1178 fn every_column(&mut self, output: &Scope) -> Vec<SortKey> {
1179 let columns: Vec<(ColumnBinding, LogicalType)> =
1180 output.columns.iter().map(|column| (column.binding, column.ty.clone())).collect();
1181 columns
1182 .into_iter()
1183 .map(|(binding, ty)| {
1184 let expr = self.plan.add_expr(Expr::Column(binding), ty);
1185 let descending = self.semantics.default_descending();
1186 SortKey { expr, descending, nulls_first: self.semantics.nulls_first(descending) }
1187 })
1188 .collect()
1189 }
1190
1191 fn sort_key(&self, expr: ExprRef, item: ast::OrderItem) -> SortKey {
1193 let descending = match item.order {
1194 Order::Unstated => self.semantics.default_descending(),
1195 Order::Ascending => false,
1196 Order::Descending => true,
1197 };
1198 let nulls_first = match item.nulls {
1199 Nulls::First => true,
1200 Nulls::Last => false,
1201 Nulls::Unstated => self.semantics.nulls_first(descending),
1202 };
1203 SortKey { expr, descending, nulls_first }
1204 }
1205
1206 fn output_position(
1208 &self,
1209 ast: &Ast,
1210 item: ast::ExprRef,
1211 output: &Scope,
1212 ) -> Result<Option<usize>> {
1213 match ast.expr(item) {
1214 ast::Expr::Literal { kind: LiteralKind::Number, text } => {
1215 let written = ast.string(text);
1216 if written.contains(['.', 'e', 'E']) {
1217 return Ok(None);
1218 }
1219 let position: usize = written.parse().map_err(|_| {
1220 Error::binder(format!("ORDER BY term {written} is not a column"))
1221 })?;
1222 if position == 0 || position > output.len() {
1223 return Err(Error::binder(format!(
1224 "ORDER BY term out of range - should be between 1 and {}",
1225 output.len()
1226 )));
1227 }
1228 Ok(Some(position - 1))
1229 }
1230 ast::Expr::Column { name } => {
1231 let parts: Vec<&str> = ast.name(name).collect();
1232 let [written] = parts.as_slice() else { return Ok(None) };
1233 Ok(output.position_of(None, written))
1234 }
1235 _ => Ok(None),
1236 }
1237 }
1238
1239 fn check_order_literal(&self, ast: &Ast, item: ast::ExprRef) -> Result<()> {
1241 if !self.semantics.order_by_non_integer_literal()
1242 && matches!(
1243 ast.expr(item),
1244 ast::Expr::Literal { kind, text }
1245 if kind != LiteralKind::Number
1246 || ast.string(text).contains(['.', 'e', 'E'])
1247 )
1248 {
1249 return Err(Error::binder(
1250 "ORDER BY non-integer literal has no effect.\n* SET order_by_non_integer_literal=true to allow this behavior.",
1251 ));
1252 }
1253 Ok(())
1254 }
1255
1256 fn distinct_on(
1258 &mut self,
1259 ast: &Ast,
1260 distinct: Distinct,
1261 output: &Scope,
1262 ) -> Result<Vec<ExprRef>> {
1263 let Distinct::On(items) = distinct else {
1264 return Ok(Vec::new());
1265 };
1266 let items = ast.expr_list(items).to_vec();
1267 let mut on = Vec::with_capacity(items.len());
1268 for item in items {
1269 let Some(position) = self.output_position(ast, item, output)? else {
1270 return Err(Error::not_implemented(
1271 "DISTINCT ON an expression that is not in the select list",
1272 ));
1273 };
1274 let column = &output.columns[position];
1275 let (binding, ty) = (column.binding, column.ty.clone());
1276 on.push(self.plan.add_expr(Expr::Column(binding), ty));
1277 }
1278 Ok(on)
1279 }
1280
1281 fn apply_limit(&mut self, ast: &Ast, query: &ast::Query, input: NodeRef) -> Result<NodeRef> {
1282 if query.limit_percent {
1283 return Err(Error::not_implemented("LIMIT with a percentage"));
1284 }
1285 let count = self.constant_count(ast, query.limit, "LIMIT")?;
1286 let offset = self.constant_count(ast, query.offset, "OFFSET")?.unwrap_or(0);
1287 if count.is_none() && offset == 0 {
1288 return Ok(input);
1289 }
1290 Ok(self.add_node(Node::Limit { input, count, offset }))
1291 }
1292
1293 fn constant_count(
1295 &mut self,
1296 ast: &Ast,
1297 written: ast::ExprRef,
1298 clause: &str,
1299 ) -> Result<Option<u64>> {
1300 if written == NONE {
1301 return Ok(None);
1302 }
1303 self.clause = "LIMIT clause";
1304 let scope = Scope::empty();
1305 let bound = self.bind_expr(ast, written, &scope)?;
1306 let Expr::Constant(value) = *self.plan.expr(bound) else {
1307 return Err(Error::not_implemented(format!("a {clause} that is not a constant")));
1308 };
1309 let count = match self.plan.value(value) {
1310 Value::Null => return Ok(None),
1311 Value::TinyInt(count) => i128::from(*count),
1312 Value::SmallInt(count) => i128::from(*count),
1313 Value::Integer(count) => i128::from(*count),
1314 Value::BigInt(count) => i128::from(*count),
1315 Value::HugeInt(count) => *count,
1316 other => {
1317 return Err(Error::binder(format!(
1318 "{clause} takes a whole number of rows, not a value of type {}",
1319 other.logical_type()
1320 )));
1321 }
1322 };
1323 u64::try_from(count)
1324 .map(Some)
1325 .map_err(|_| Error::binder(format!("{clause} must not be negative")))
1326 }
1327
1328 fn bind_from(&mut self, ast: &Ast, from: ast::Slice) -> Result<(NodeRef, Scope)> {
1331 let sources = ast.source_list(from).to_vec();
1332 let Some((first, rest)) = sources.split_first() else {
1333 return Ok((self.add_node(Node::Dummy), Scope::empty()));
1336 };
1337 let (mut node, mut scope) = self.bind_source(ast, *first)?;
1338 for source in rest {
1339 let (right, right_scope, correlations) = self.bind_lateral(ast, *source, &scope)?;
1340 node = if correlations.is_empty() {
1341 self.add_node(Node::CrossProduct { left: node, right })
1342 } else {
1343 let conditions = self.plan.add_expr_list(&[]);
1344 self.add_node(Node::DependentJoin {
1345 left: node,
1346 right,
1347 kind: JoinKind::Inner,
1348 conditions,
1349 })
1350 };
1351 scope = scope.concat(right_scope);
1352 }
1353 Ok((node, scope))
1354 }
1355
1356 fn bind_lateral(
1368 &mut self,
1369 ast: &Ast,
1370 source: ast::SourceRef,
1371 left: &Scope,
1372 ) -> Result<(NodeRef, Scope, Vec<ColumnBinding>)> {
1373 self.lateral_scopes.push(self.outer_scopes.len());
1374 self.outer_scopes.push(left.clone());
1375 self.correlations.push(Vec::new());
1376 let bound = self.bind_source(ast, source);
1377 let read = self.correlations.pop().expect("correlation frame");
1378 self.outer_scopes.pop();
1379 self.lateral_scopes.pop();
1380 let (node, scope) = bound?;
1381
1382 let mut here = Vec::new();
1383 for binding in read {
1384 if left.columns.iter().any(|column| column.binding == binding) {
1385 here.push(binding);
1386 } else if let Some(enclosing) = self.correlations.last_mut() {
1387 if !enclosing.contains(&binding) {
1388 enclosing.push(binding);
1389 }
1390 }
1391 }
1392 Ok((node, scope, here))
1402 }
1403
1404 fn bind_source(&mut self, ast: &Ast, source: ast::SourceRef) -> Result<(NodeRef, Scope)> {
1405 match ast.source(source) {
1406 ast::Source::Table { name, alias, columns } => {
1407 self.bind_table(ast, name, alias, columns)
1408 }
1409 ast::Source::Function { name, args, alias, columns, pragma } => {
1410 self.bind_table_function(ast, name, args, alias, columns, pragma)
1411 }
1412 ast::Source::Subquery { query, alias, columns } => {
1413 let (node, mut scope) = self.bind_query(ast, query)?;
1414 let label = if alias == NONE {
1415 "unnamed_subquery".to_string()
1416 } else {
1417 ast.string(alias).to_string()
1418 };
1419 scope.relabel(&label);
1420 if !columns.is_empty() {
1421 let names: Vec<&str> = ast.name(columns).collect();
1422 scope.rename(&names, &label)?;
1423 }
1424 Ok((node, scope))
1425 }
1426 ast::Source::Values { rows, alias, columns } => {
1427 let bare = ast::Query::bare(ast::QueryBody::Values(rows));
1428 let (node, mut scope) = self.bind_values(ast, &bare, rows)?;
1429 let label =
1430 if alias == NONE { String::new() } else { ast.string(alias).to_string() };
1431 scope.relabel(&label);
1432 if !columns.is_empty() {
1433 let names: Vec<&str> = ast.name(columns).collect();
1434 scope.rename(&names, &label)?;
1435 }
1436 Ok((node, scope))
1437 }
1438 ast::Source::Cte { cte, alias, columns } => {
1439 self.bind_cte_scan(ast, cte, alias, columns)
1440 }
1441 ast::Source::Join { left, right, kind, natural, on, using } => {
1442 self.bind_join(ast, left, right, kind, natural, on, using)
1443 }
1444 }
1445 }
1446
1447 fn bind_cte_scan(
1454 &mut self,
1455 ast: &Ast,
1456 written: u32,
1457 alias: ast::StrRef,
1458 columns: ast::Slice,
1459 ) -> Result<(NodeRef, Scope)> {
1460 let Some(held) = self.materialized.iter().rev().find(|held| held.written == written) else {
1461 let name = ast.string(ast.cte(written).name);
1462 return Err(Error::binder(format!("Table with name {name} does not exist!")));
1463 };
1464 let cte = held.cte;
1465 let fields = held.fields.clone();
1466 let text = held.name.clone();
1467 let label = if alias == NONE { text.clone() } else { ast.string(alias).to_string() };
1468 let name = self.plan.intern(&text);
1469 let index = self.fresh_index();
1470 let mut scope = Scope::empty();
1471 for (at, field) in fields.iter().enumerate() {
1472 scope.push(Visible {
1473 table: label.clone(),
1474 name: field.name.clone(),
1475 binding: ColumnBinding::new(index, at as u32),
1476 ty: field.ty.clone(),
1477 not_null: field.not_null,
1478 });
1479 }
1480 if !columns.is_empty() {
1481 let names: Vec<&str> = ast.name(columns).collect();
1482 scope.rename(&names, &label)?;
1483 }
1484 let columns = self.plan.add_fields(&fields);
1485 let node = self.add_node(Node::CteScan { index, cte, name, columns });
1486 Ok((node, scope))
1487 }
1488
1489 fn bind_table(
1490 &mut self,
1491 ast: &Ast,
1492 name: ast::Slice,
1493 alias: ast::StrRef,
1494 columns: ast::Slice,
1495 ) -> Result<(NodeRef, Scope)> {
1496 let parts: Vec<&str> = ast.name(name).collect();
1497 let catalog = self.catalog;
1498 let resolved = match catalog.resolve(&parts) {
1501 Ok(resolved) => resolved,
1502 Err(missing) => {
1503 return self.bind_replacement_scan(ast, &parts, alias, columns, missing);
1504 }
1505 };
1506 if catalog.entry(&resolved)? == Entry::View {
1507 return self.bind_view(ast, &resolved, alias, columns);
1508 }
1509 let table = catalog.table(&resolved)?;
1510 let fields: Vec<Field> = table.columns().to_vec();
1511 let label =
1512 if alias == NONE { resolved.table.clone() } else { ast.string(alias).to_string() };
1513 let index = self.fresh_index();
1514 let mut scope = Scope::empty();
1515 for (at, field) in fields.iter().enumerate() {
1516 scope.push(Visible {
1517 table: label.clone(),
1518 name: field.name.clone(),
1519 binding: ColumnBinding::new(index, at as u32),
1520 ty: field.ty.clone(),
1521 not_null: field.not_null,
1522 });
1523 }
1524 if !columns.is_empty() {
1525 let names: Vec<&str> = ast.name(columns).collect();
1526 scope.rename(&names, &label)?;
1527 }
1528 let catalog_name = self.plan.intern(&resolved.catalog);
1529 let schema = self.plan.intern(&resolved.schema);
1530 let table_name = self.plan.intern(&resolved.table);
1531 let alias = self.plan.intern(&label);
1532 let columns = self.plan.add_fields(&fields);
1533 let node = self.add_node(Node::Get {
1534 catalog: catalog_name,
1535 schema,
1536 table: table_name,
1537 alias,
1538 index,
1539 columns,
1540 });
1541 Ok((node, scope))
1542 }
1543
1544 fn bind_view(
1556 &mut self,
1557 ast: &Ast,
1558 name: &QualifiedName,
1559 alias: ast::StrRef,
1560 columns: ast::Slice,
1561 ) -> Result<(NodeRef, Scope)> {
1562 let view = self.catalog.view(name)?;
1563 let full = name.to_string();
1564 if self.expanding.contains(&full) {
1565 return Err(Error::binder(format!(
1569 "infinite recursion detected: attempting to recursively bind view \"\"{}\"\"",
1570 name.table
1571 )));
1572 }
1573 let body = parse_ast_with_case(view.sql(), self.semantics.identifier_case())?;
1574 let query = match body.statements.as_slice() {
1575 [ast::Statement::Query(query)] => *query,
1576 _ => return Err(Error::binder(format!("view \"{}\" is not a query", name.table))),
1579 };
1580 self.expanding.push(full);
1581 let bound = self.bind_query(&body, query);
1582 self.expanding.pop();
1583 let (node, mut scope) = bound?;
1584
1585 let aliases: Vec<&str> = view.aliases().iter().map(String::as_str).collect();
1586 if !aliases.is_empty() {
1587 scope.rename(&aliases, "unnamed_subquery")?;
1588 }
1589 view.remember(scope.fields());
1596 let label = if alias == NONE { name.table.clone() } else { ast.string(alias).to_string() };
1597 scope.relabel(&label);
1598 if !columns.is_empty() {
1599 let names: Vec<&str> = ast.name(columns).collect();
1600 scope.rename(&names, &label)?;
1601 }
1602 Ok((node, scope))
1603 }
1604
1605 fn bind_table_function(
1613 &mut self,
1614 ast: &Ast,
1615 name: ast::Slice,
1616 args: ast::Slice,
1617 alias: ast::StrRef,
1618 columns: ast::Slice,
1619 pragma: bool,
1620 ) -> Result<(NodeRef, Scope)> {
1621 let parts: Vec<&str> = ast.name(name).collect();
1622 let function_name = *parts.last().unwrap_or(&"");
1626 if let Some(schema) = parts.iter().rev().nth(1) {
1627 if !schema.eq_ignore_ascii_case("main") && !schema.eq_ignore_ascii_case("system") {
1628 return Err(Error::catalog(format!(
1629 "Table Function with name {} does not exist!",
1630 parts.join(".")
1631 )));
1632 }
1633 }
1634 let Some(called) = TableFunction::lookup(function_name) else {
1638 if pragma {
1639 if args.is_empty() && self.catalog.resolve(&parts).is_ok() {
1645 return self.bind_table(ast, name, alias, columns);
1646 }
1647 let spelled = function_name.strip_prefix("pragma_").unwrap_or(function_name);
1648 return Err(Error::catalog(format!(
1649 "Pragma Function with name {spelled} does not exist!"
1650 )));
1651 }
1652 return Err(Error::catalog(format!(
1653 "Table Function with name {function_name} does not exist!"
1654 )));
1655 };
1656 let written = ast.target_list(args).to_vec();
1657 let empty = Scope::empty();
1658 let previous = std::mem::replace(&mut self.clause, "table function arguments");
1659 let mut bound = Vec::new();
1660 let mut written_options = Vec::new();
1661 for argument in written {
1662 let expr = self.bind_expr(ast, argument.expr, &empty)?;
1663 if argument.alias == NONE {
1664 bound.push(expr);
1665 } else {
1666 let name = ast.string(argument.alias).to_string();
1667 let (parameter, value) = self.named_argument(called, &name, expr)?;
1668 written_options.push((parameter, value, expr));
1669 }
1670 }
1671 self.clause = previous;
1672 let options = Options::of(&written_options)?;
1673
1674 let given: Vec<LogicalType> =
1677 bound.iter().map(|&expr| self.plan.expr_type(expr).clone()).collect();
1678 let resolved = if pragma {
1679 resolve_pragma(function_name, &given)?
1680 } else {
1681 resolve_table(function_name, &given)?
1682 };
1683 let mut cast: Vec<ExprRef> = bound
1684 .iter()
1685 .zip(&resolved.arguments)
1686 .map(|(&expr, ty)| self.checked_cast_to(expr, ty, false))
1687 .collect::<Result<_>>()?;
1688
1689 if resolved.function.takes_a_name() {
1690 let Columns::Fixed(fields) = resolved.columns else {
1691 return Err(Error::internal("a pragma that resolved to a file"));
1692 };
1693 let [argument] = cast[..] else {
1694 return Err(Error::internal("a pragma that resolved to more than one name"));
1695 };
1696 return self.bind_pragma(ast, resolved.function, &fields, argument, alias, columns);
1697 }
1698 let mut measured = Stat::Unknown;
1701 let mut counted: Vec<(String, u64)> = Vec::new();
1702 let fields = match resolved.columns {
1703 Columns::Fixed(fields) => fields,
1704 columns => {
1705 let paths = self.file_paths(cast[0], resolved.function.name())?;
1710 let mut fields = match columns {
1711 Columns::Csv => csv_fields(&paths, options.given)?,
1714 _ => {
1715 let footers = parquet_footers(&paths)?;
1716 measured = footers.rows;
1717 counted = footers.distincts;
1718 footers.fields
1719 }
1720 };
1721 if options.all_varchar {
1722 for field in &mut fields {
1727 field.ty = LogicalType::Varchar;
1728 }
1729 }
1730 if options.binary_as_string {
1731 for field in &mut fields {
1736 if field.ty == LogicalType::Blob {
1737 field.ty = LogicalType::Varchar;
1738 }
1739 }
1740 }
1741 if options.file_row_number {
1742 if fields.iter().any(|field| field.name == FILE_ROW_NUMBER) {
1748 return Err(Error::binder(format!(
1749 "Duplicate column name \"{FILE_ROW_NUMBER}\": the file already has a \
1750 column of that name, so file_row_number cannot add one"
1751 )));
1752 }
1753 fields.push(Field::required(FILE_ROW_NUMBER.to_string(), LogicalType::BigInt));
1754 }
1755 cast = paths.iter().map(|path| self.path_constant(path)).collect();
1756 fields
1757 }
1758 };
1759 let label = if alias == NONE {
1760 resolved.function.name().to_string()
1761 } else {
1762 ast.string(alias).to_string()
1763 };
1764 let names: Vec<&str> = ast.name(columns).collect();
1765 self.table_function_source(
1766 resolved.function,
1767 &cast,
1768 &written_options,
1769 Read { fields, rows: measured, distincts: counted },
1770 &label,
1771 &names,
1772 )
1773 }
1774
1775 fn bind_pragma(
1788 &mut self,
1789 ast: &Ast,
1790 function: TableFunction,
1791 fields: &[Field],
1792 argument: ExprRef,
1793 alias: ast::StrRef,
1794 columns: ast::Slice,
1795 ) -> Result<(NodeRef, Scope)> {
1796 let written = self.pragma_name(argument, function)?;
1797 let parts = identifier_parts(&written);
1798 let spelled: Vec<&str> = parts.iter().map(String::as_str).collect();
1799 let name = self.catalog.resolve(&spelled)?;
1800 let described = self.described(ast, &name)?;
1801 let mut rows = Vec::with_capacity(described.len());
1802 for (at, field) in described.iter().enumerate() {
1803 let items = if matches!(function, TableFunction::PragmaShow) {
1804 self.describing(field)
1805 } else {
1806 self.table_info(at, field)
1807 };
1808 rows.push(self.plan.add_expr_list(&items));
1809 }
1810 let rows = self.plan.add_rows(&rows);
1811 let held = self.plan.add_fields(fields);
1812 let index = self.fresh_index();
1813 let node = self.add_node(Node::Values { index, columns: held, rows });
1814 let label =
1815 if alias == NONE { function.name().to_string() } else { ast.string(alias).to_string() };
1816 let mut scope = Scope::empty();
1817 for (at, field) in fields.iter().enumerate() {
1818 scope.push(Visible {
1819 table: label.clone(),
1820 name: field.name.clone(),
1821 binding: ColumnBinding::new(index, at as u32),
1822 ty: field.ty.clone(),
1823 not_null: false,
1824 });
1825 }
1826 if !columns.is_empty() {
1827 let names: Vec<&str> = ast.name(columns).collect();
1828 scope.rename(&names, &label)?;
1829 }
1830 Ok((node, scope))
1831 }
1832
1833 fn pragma_name(&self, argument: ExprRef, function: TableFunction) -> Result<String> {
1843 let Expr::Constant(reference) = *self.plan.expr(argument) else {
1844 return Err(Error::not_implemented(format!(
1845 "{}() given a name that is not a constant",
1846 function.name()
1847 )));
1848 };
1849 match self.plan.value(reference) {
1850 Value::Varchar(name) => Ok(name.clone()),
1851 Value::Null => Ok("NULL".to_string()),
1852 other => {
1853 Err(Error::internal(format!("a pragma name bound as VARCHAR arrived as {other}")))
1854 }
1855 }
1856 }
1857
1858 fn described(&mut self, ast: &Ast, name: &QualifiedName) -> Result<Vec<Field>> {
1869 if self.catalog.entry(name)? == Entry::Table {
1870 return Ok(self.catalog.table(name)?.columns().to_vec());
1871 }
1872 let (_, scope) = self.bind_view(ast, name, NONE, ast::Slice::default())?;
1873 Ok(scope.fields())
1874 }
1875
1876 fn describing(&mut self, field: &Field) -> Vec<ExprRef> {
1878 let written = [
1879 field.name.clone(),
1880 field.ty.to_string(),
1881 if field.not_null { "NO" } else { "YES" }.to_owned(),
1882 ];
1883 let mut items: Vec<ExprRef> =
1884 written.into_iter().map(|text| self.plan.add_constant(Value::Varchar(text))).collect();
1885 for _ in 0..3 {
1886 let empty = self.plan.add_constant(Value::Null);
1887 items.push(self.cast_to(empty, &LogicalType::Varchar));
1888 }
1889 items
1890 }
1891
1892 fn table_info(&mut self, at: usize, field: &Field) -> Vec<ExprRef> {
1898 let cid = self.plan.add_constant(Value::Integer(i32::try_from(at).unwrap_or(i32::MAX)));
1899 let name = self.plan.add_constant(Value::Varchar(field.name.clone()));
1900 let ty = self.plan.add_constant(Value::Varchar(field.ty.to_string()));
1901 let not_null = self.plan.add_constant(Value::Boolean(field.not_null));
1902 let default = self.plan.add_constant(Value::Null);
1903 let default = self.cast_to(default, &LogicalType::Varchar);
1904 let key = self.plan.add_constant(Value::Boolean(false));
1905 vec![cid, name, ty, not_null, default, key]
1906 }
1907
1908 fn named_argument(
1922 &mut self,
1923 function: TableFunction,
1924 name: &str,
1925 expr: ExprRef,
1926 ) -> Result<(&'static str, Value)> {
1927 let known = function
1928 .parameters()
1929 .iter()
1930 .find(|(parameter, _)| parameter.eq_ignore_ascii_case(name));
1931 let Some((parameter, wanted)) = known else {
1932 let candidates: Vec<String> = function
1933 .parameters()
1934 .iter()
1935 .map(|(parameter, ty)| format!(" {parameter} {ty}"))
1936 .collect();
1937 return Err(Error::binder(format!(
1938 "Invalid named parameter \"{name}\" for function {}\nCandidates:\n{}\n",
1939 function.name(),
1940 candidates.join("\n")
1941 )));
1942 };
1943 let Expr::Constant(reference) = *self.plan.expr(expr) else {
1944 return Err(Error::not_implemented(format!(
1945 "the named parameter {parameter} with a value that is not a constant"
1946 )));
1947 };
1948 let value = self.plan.value(reference).clone();
1949 if value == Value::Null {
1950 return Err(Error::binder(null_parameter(function, parameter)));
1951 }
1952 let given = self.plan.expr_type(expr).clone();
1953 if given != *wanted {
1954 return Err(Error::not_implemented(format!(
1955 "the named parameter {parameter} given a {given} where a {wanted} was wanted"
1956 )));
1957 }
1958 Ok((parameter, value))
1959 }
1960
1961 fn bind_replacement_scan(
1972 &mut self,
1973 ast: &Ast,
1974 parts: &[&str],
1975 alias: ast::StrRef,
1976 columns: ast::Slice,
1977 missing: Error,
1978 ) -> Result<(NodeRef, Scope)> {
1979 let [path] = parts else { return Err(missing) };
1980 let path = *path;
1981 let extension = path.rsplit_once('.').map(|(_, after)| after).unwrap_or_default();
1982 let Some(function) = Self::reader_for(extension) else {
1983 if is_file(path) {
1984 return Err(Error::binder(format!(
1989 "No extension found that is capable of reading the file \"{path}\"\n* If this \
1990 file is a supported file format you can explicitly use the reader functions, \
1991 such as read_csv, read_json or read_parquet"
1992 )));
1993 }
1994 return Err(missing);
1995 };
1996 let paths = files(path)?;
2001 let read = match function {
2002 TableFunction::ReadParquet => {
2003 let footers = parquet_footers(&paths)?;
2004 Read { fields: footers.fields, rows: footers.rows, distincts: footers.distincts }
2005 }
2006 _ => Read::uncounted(csv_fields(&paths, Given::default())?),
2007 };
2008 let label = if alias == NONE {
2014 if is_pattern(path) {
2015 path.to_string()
2016 } else {
2017 let file = path.rsplit_once('/').map_or(path, |(_, file)| file);
2018 file.rsplit_once('.').map_or(file, |(stem, _)| stem).to_string()
2019 }
2020 } else {
2021 ast.string(alias).to_string()
2022 };
2023 let arguments: Vec<ExprRef> = paths.iter().map(|path| self.path_constant(path)).collect();
2024 let names: Vec<&str> = ast.name(columns).collect();
2025 self.table_function_source(function, &arguments, &[], read, &label, &names)
2026 }
2027
2028 fn path_constant(&mut self, path: &str) -> ExprRef {
2030 let value = self.plan.add_value(Value::Varchar(path.to_string()));
2031 self.plan.add_expr(Expr::Constant(value), LogicalType::Varchar)
2032 }
2033
2034 fn reader_for(extension: &str) -> Option<TableFunction> {
2041 if extension.eq_ignore_ascii_case("parquet") {
2042 return Some(TableFunction::ReadParquet);
2043 }
2044 if extension.eq_ignore_ascii_case("csv") || extension.eq_ignore_ascii_case("tsv") {
2045 return Some(TableFunction::ReadCsv);
2046 }
2047 None
2048 }
2049
2050 fn table_function_source(
2060 &mut self,
2061 function: TableFunction,
2062 args: &[ExprRef],
2063 written: &[(&'static str, Value, ExprRef)],
2064 read: Read,
2065 label: &str,
2066 names: &[&str],
2067 ) -> Result<(NodeRef, Scope)> {
2068 let Read { fields, rows, distincts } = read;
2069 let index = self.fresh_index();
2070 if rows.is_known() {
2075 self.plan.measure(index, rows);
2076 }
2077 for (column, distinct) in distincts {
2078 self.plan.measure_distinct(index, &column, distinct);
2079 }
2080 let mut scope = Scope::empty();
2081 for (at, field) in fields.iter().enumerate() {
2082 scope.push(Visible {
2083 table: label.to_string(),
2084 name: field.name.clone(),
2085 binding: ColumnBinding::new(index, at as u32),
2086 ty: field.ty.clone(),
2087 not_null: false,
2090 });
2091 }
2092 if !names.is_empty() {
2093 scope.rename(names, label)?;
2094 }
2095 let function = self.plan.intern(function.name());
2096 let args = self.plan.add_expr_list(args);
2097 let named: Vec<u32> =
2098 written.iter().map(|(parameter, _, _)| self.plan.intern(parameter)).collect();
2099 let settings: Vec<ExprRef> = written.iter().map(|(_, _, expr)| *expr).collect();
2100 let options = self.plan.add_name_list(&named);
2101 let settings = self.plan.add_expr_list(&settings);
2102 let columns = self.plan.add_fields(&fields);
2103 let node = self.add_node(Node::TableFunction {
2104 index,
2105 function,
2106 args,
2107 options,
2108 settings,
2109 columns,
2110 });
2111 Ok((node, scope))
2112 }
2113
2114 fn file_paths(&self, expr: ExprRef, name: &str) -> Result<Vec<String>> {
2121 let mut paths = Vec::new();
2122 for pattern in self.file_patterns(expr, name)? {
2123 paths.extend(files(&pattern)?);
2124 }
2125 Ok(paths)
2126 }
2127
2128 fn file_patterns(&self, expr: ExprRef, name: &str) -> Result<Vec<String>> {
2140 let Expr::Constant(reference) = *self.plan.expr(expr) else {
2141 return Err(Error::not_implemented(
2142 "a table function file name that is not a constant",
2143 ));
2144 };
2145 match self.plan.value(reference) {
2146 Value::Varchar(path) => Ok(vec![path.clone()]),
2147 Value::Null => Err(Error::parser(format!("{name} cannot take NULL list as parameter"))),
2149 Value::List { values, .. } => values
2150 .iter()
2151 .map(|value| match value {
2152 Value::Varchar(path) => Ok(path.clone()),
2153 _ => Err(Error::parser(format!(
2154 "{name} reader cannot take NULL input as parameter"
2155 ))),
2156 })
2157 .collect(),
2158 other => {
2159 Err(Error::internal(format!("a file name bound as VARCHAR arrived as {other}")))
2160 }
2161 }
2162 }
2163
2164 #[allow(clippy::too_many_arguments)]
2165 fn bind_join(
2166 &mut self,
2167 ast: &Ast,
2168 left: ast::SourceRef,
2169 right: ast::SourceRef,
2170 kind: ast::JoinKind,
2171 natural: bool,
2172 on: ast::ExprRef,
2173 using: ast::Slice,
2174 ) -> Result<(NodeRef, Scope)> {
2175 let (left_node, left_scope) = self.bind_source(ast, left)?;
2176 let (right_node, right_scope, correlated) = self.bind_lateral(ast, right, &left_scope)?;
2177 if !correlated.is_empty()
2181 && !matches!(kind, ast::JoinKind::Inner | ast::JoinKind::Cross | ast::JoinKind::Left)
2182 {
2183 return Err(Error::binder(
2184 "The combining JOIN type must be INNER or LEFT for a LATERAL reference",
2185 ));
2186 }
2187 let split = left_scope.len();
2188 let mut scope = left_scope.concat(right_scope);
2189
2190 let merged: Vec<String> = if natural {
2193 let mut names = Vec::new();
2194 for (at, column) in scope.columns.iter().enumerate().take(split) {
2195 if scope.columns[split..].iter().any(|right| same_name(&right.name, &column.name))
2196 && !names.iter().any(|held: &String| same_name(held, &column.name))
2197 {
2198 let _ = at;
2199 names.push(column.name.clone());
2200 }
2201 }
2202 names
2203 } else {
2204 let mut names: Vec<String> = Vec::new();
2210 for name in ast.name(using) {
2211 if !names.iter().any(|held| same_name(held, name)) {
2212 names.push(name.to_string());
2213 }
2214 }
2215 names
2216 };
2217
2218 let mut conditions = Vec::new();
2219 let mut dropped = Vec::new();
2220 for name in &merged {
2221 let left_at = scope.columns[..split]
2222 .iter()
2223 .position(|column| same_name(&column.name, name))
2224 .ok_or_else(|| {
2225 Error::binder(format!(
2226 "column \"{name}\" specified in USING clause does not exist in left table"
2227 ))
2228 })?;
2229 let right_at = scope.columns[split..]
2230 .iter()
2231 .position(|column| same_name(&column.name, name))
2232 .map(|at| at + split)
2233 .ok_or_else(|| {
2234 Error::binder(format!(
2235 "column \"{name}\" specified in USING clause does not exist in right table"
2236 ))
2237 })?;
2238 let left_column = &scope.columns[left_at];
2239 let (left_binding, left_type) = (left_column.binding, left_column.ty.clone());
2240 let right_column = &scope.columns[right_at];
2241 let (right_binding, right_type) = (right_column.binding, right_column.ty.clone());
2242 let left_expr = self.plan.add_expr(Expr::Column(left_binding), left_type);
2243 let right_expr = self.plan.add_expr(Expr::Column(right_binding), right_type);
2244 conditions.push(self.compare(rudb_plan::CompareOp::Equal, left_expr, right_expr)?);
2245 dropped.push(right_at);
2246 }
2247 dropped.sort_unstable();
2250 for at in dropped.into_iter().rev() {
2251 scope.remove(at);
2252 }
2253
2254 if on != NONE {
2255 if !merged.is_empty() {
2256 return Err(Error::binder("a join cannot have both ON and USING"));
2257 }
2258 self.clause = "JOIN condition";
2259 let predicate = self.bind_expr(ast, on, &scope)?;
2260 conditions.push(self.as_boolean(predicate, "JOIN")?);
2261 }
2262
2263 if kind == ast::JoinKind::Cross && !conditions.is_empty() {
2264 return Err(Error::binder("a CROSS JOIN cannot have a condition"));
2265 }
2266 if correlated.is_empty()
2270 && conditions.is_empty()
2271 && matches!(kind, ast::JoinKind::Cross | ast::JoinKind::Inner)
2272 {
2273 let node = self.add_node(Node::CrossProduct { left: left_node, right: right_node });
2274 return Ok((node, scope));
2275 }
2276 if matches!(kind, ast::JoinKind::Semi | ast::JoinKind::Anti) {
2285 scope.truncate(split);
2286 }
2287 let kind = match kind {
2288 ast::JoinKind::Inner | ast::JoinKind::Cross => JoinKind::Inner,
2289 ast::JoinKind::Left => JoinKind::Left,
2290 ast::JoinKind::Right => JoinKind::Right,
2291 ast::JoinKind::Full => JoinKind::Full,
2292 ast::JoinKind::Semi => JoinKind::Semi,
2293 ast::JoinKind::Anti => JoinKind::Anti,
2294 ast::JoinKind::Positional => JoinKind::Positional,
2295 };
2296 let conditions = self.plan.add_expr_list(&conditions);
2297 let node = if correlated.is_empty() {
2298 self.add_node(Node::Join {
2299 left: left_node,
2300 right: right_node,
2301 kind,
2302 conditions,
2303 build: BuildSide::default(),
2304 })
2305 } else {
2306 self.add_node(Node::DependentJoin {
2307 left: left_node,
2308 right: right_node,
2309 kind,
2310 conditions,
2311 })
2312 };
2313 Ok((node, scope))
2314 }
2315
2316 fn bind_filter(
2324 &mut self,
2325 ast: &Ast,
2326 filter: ast::ExprRef,
2327 scope: &Scope,
2328 ) -> Result<Option<ExprRef>> {
2329 if filter == NONE {
2330 return Ok(None);
2331 }
2332 let bound = self.bind_expr(ast, filter, scope)?;
2333 Ok(Some(self.checked_cast_to(bound, &LogicalType::Boolean, false)?))
2334 }
2335
2336 pub(crate) fn bind_aggregate(
2338 &mut self,
2339 ast: &Ast,
2340 name: &str,
2341 args: &[ast::ExprRef],
2342 distinct: bool,
2343 filter: ast::ExprRef,
2344 scope: &Scope,
2345 ) -> Result<ExprRef> {
2346 if self.in_filter {
2347 return Err(Error::binder("aggregate functions are not allowed in FILTER"));
2348 }
2349 if self.in_aggregate {
2350 return Err(Error::binder(format!(
2351 "aggregate function calls cannot be nested, and {name}() is inside one"
2352 )));
2353 }
2354 if self.aggregation.is_none() {
2355 return Err(Error::binder(format!(
2356 "aggregate function calls cannot be used in the {}",
2357 self.clause
2358 )));
2359 }
2360 self.in_aggregate = true;
2365 self.in_filter = true;
2366 let filter = self.bind_filter(ast, filter, scope);
2367 self.in_filter = false;
2368 self.in_aggregate = false;
2369 let filter = filter?;
2370
2371 self.in_aggregate = true;
2372 let mut bound = Vec::with_capacity(args.len());
2373 let mut failure = None;
2374 for &arg in args {
2375 match self.bind_expr(ast, arg, scope) {
2376 Ok(expr) => bound.push(expr),
2377 Err(error) => {
2378 failure = Some(error);
2379 break;
2380 }
2381 }
2382 }
2383 self.in_aggregate = false;
2384 if let Some(error) = failure {
2385 return Err(error);
2386 }
2387
2388 let types: Vec<LogicalType> =
2389 bound.iter().map(|&arg| self.plan.expr_type(arg).clone()).collect();
2390 let resolved = resolve(name, &types)?;
2391 let mut cast = Vec::with_capacity(bound.len());
2392 for (arg, wanted) in bound.iter().zip(&resolved.arguments) {
2393 cast.push(self.checked_cast_to(*arg, wanted, false)?);
2394 }
2395 let args = self.plan.add_expr_list(&cast);
2396 let name = self.plan.intern(resolved.name);
2397 let ty = resolved.returns;
2398 let call = self.plan.add_expr(Expr::Aggregate { name, args, distinct, filter }, ty.clone());
2399
2400 let existing = self.aggregation.as_ref().map(|held| held.aggregates.clone());
2403 let existing = existing.unwrap_or_default();
2404 let at = match existing.iter().position(|&held| self.same_expr(held, call)) {
2405 Some(at) => at,
2406 None => {
2407 let aggregation = self.aggregation.as_mut().expect("checked above");
2408 aggregation.aggregates.push(call);
2409 aggregation.aggregates.len() - 1
2410 }
2411 };
2412 let aggregation = self.aggregation.as_ref().expect("checked above");
2413 let (index, groups) = (aggregation.index, aggregation.groups.len());
2414 Ok(self.column(index, groups + at, ty))
2415 }
2416
2417 pub(crate) fn bind_window(
2425 &mut self,
2426 ast: &Ast,
2427 written: &WindowCall<'_>,
2428 scope: &Scope,
2429 ) -> Result<ExprRef> {
2430 let WindowCall { name, args, distinct, filter, ignore_nulls, spec } = *written;
2431 if self.in_aggregate {
2432 return Err(Error::binder(
2433 "aggregate function calls cannot contain window function calls",
2434 ));
2435 }
2436 if self.in_window {
2437 return Err(Error::binder("window function calls cannot be nested"));
2438 }
2439 let clause = if self.clause == "JOIN condition" { "WHERE clause" } else { self.clause };
2443 if clause != "SELECT clause" && clause != "ORDER BY clause" {
2444 return Err(Error::binder(format!("{clause} cannot contain window functions!")));
2445 }
2446
2447 let starred = args.iter().any(|&arg| {
2451 matches!(ast.expr(arg), ast::Expr::Star { qualifier, replacements }
2452 if qualifier.is_empty() && replacements.is_empty())
2453 });
2454 let (name, args): (&str, &[ast::ExprRef]) = if starred {
2455 if !same_name(name, "count") || args.len() != 1 {
2456 return Err(Error::binder(format!("* is not allowed in {name}()")));
2457 }
2458 ("count_star", &[])
2459 } else if same_name(name, "count") && args.is_empty() {
2460 ("count_star", &[])
2463 } else {
2464 (name, args)
2465 };
2466
2467 let held = ast.window(spec);
2468 self.in_window = true;
2469 let parts = self.window_parts(ast, args, held, scope);
2470 let filter = if parts.is_ok() { self.bind_filter(ast, filter, scope) } else { Ok(None) };
2475 self.in_window = false;
2476 let parts = parts?;
2477 let filter = filter?;
2478 let offsets = [parts.frame.start, parts.frame.end]
2481 .iter()
2482 .any(|end| matches!(end, WindowBound::Preceding(_) | WindowBound::Following(_)));
2483 if parts.frame.unit == WindowUnit::Range && offsets && parts.order.len() != 1 {
2484 return Err(Error::binder("RANGE frames must have only one ORDER BY expression"));
2485 }
2486
2487 let types: Vec<LogicalType> =
2488 parts.args.iter().map(|&arg| self.plan.expr_type(arg).clone()).collect();
2489 let resolved = window_signature(name, &types)?;
2490 if resolved.name == "fill" {
2493 let keys: Vec<LogicalType> =
2494 parts.order.iter().map(|key| self.plan.expr_type(key.expr).clone()).collect();
2495 refuse_fill(&types[0], &keys, distinct, ignore_nulls)?;
2496 }
2497 if distinct && kind_of(resolved.name) == Some(FunctionKind::Window) {
2501 return Err(Error::binder(format!(
2502 "DISTINCT is not implemented for the window function \"\"{name}\"\""
2503 )));
2504 }
2505 if filter.is_some() && kind_of(resolved.name) == Some(FunctionKind::Window) {
2508 return Err(Error::binder(format!(
2509 "FILTER is not implemented for the window function \"\"{name}\"\""
2510 )));
2511 }
2512 let mut cast = Vec::with_capacity(parts.args.len());
2513 for (arg, wanted) in parts.args.iter().zip(&resolved.arguments) {
2514 cast.push(self.checked_cast_to(*arg, wanted, false)?);
2515 }
2516 let args = self.plan.add_expr_list(&cast);
2517 let name = self.plan.intern(resolved.name);
2518 let ty = resolved.returns;
2519 let call = self
2520 .plan
2521 .add_expr(Expr::Window { name, args, distinct, filter, ignore_nulls }, ty.clone());
2522
2523 let at = self.window_run(parts.partition, parts.order, parts.frame, call);
2524 let index = self.windows.last().expect("the run was just filed").index;
2525 Ok(self.column(index, at, ty))
2526 }
2527
2528 fn window_run(
2535 &mut self,
2536 partition: Vec<ExprRef>,
2537 order: Vec<SortKey>,
2538 frame: WindowFrame,
2539 call: ExprRef,
2540 ) -> usize {
2541 let matches = self.windows.last().is_some_and(|run| {
2542 run.frame == frame
2543 && run.partition.len() == partition.len()
2544 && run.order.len() == order.len()
2545 && run.partition.iter().zip(&partition).all(|(&l, &r)| self.same_expr(l, r))
2546 && run.order.iter().zip(&order).all(|(l, r)| {
2547 l.descending == r.descending
2548 && l.nulls_first == r.nulls_first
2549 && self.same_expr(l.expr, r.expr)
2550 })
2551 });
2552 if !matches {
2553 let index = self.fresh_index();
2554 self.windows.push(WindowRun { index, partition, order, frame, calls: Vec::new() });
2555 }
2556 let calls = self.windows.last().expect("a run is open").calls.clone();
2559 if let Some(at) = calls.iter().position(|&held| self.same_expr(held, call)) {
2560 return at;
2561 }
2562 let run = self.windows.last_mut().expect("a run is open");
2563 run.calls.push(call);
2564 run.calls.len() - 1
2565 }
2566
2567 fn window_parts(
2573 &mut self,
2574 ast: &Ast,
2575 args: &[ast::ExprRef],
2576 held: ast::WindowSpec,
2577 scope: &Scope,
2578 ) -> Result<WindowParts> {
2579 let mut bound = Vec::with_capacity(args.len());
2580 for &arg in args {
2581 let expr = self.bind_expr(ast, arg, scope)?;
2582 bound.push(self.over_aggregate(expr, scope)?);
2583 }
2584 let mut partition = Vec::new();
2585 for &key in ast.expr_list(held.partition) {
2586 let expr = self.bind_expr(ast, key, scope)?;
2587 partition.push(self.over_aggregate(expr, scope)?);
2588 }
2589 let mut order = Vec::new();
2590 for item in ast.order_list(held.order).to_vec() {
2591 let expr = self.bind_expr(ast, item.expr, scope)?;
2592 let expr = self.over_aggregate(expr, scope)?;
2593 order.push(self.sort_key(expr, item));
2594 }
2595 let frame = WindowFrame {
2596 unit: match held.unit {
2597 ast::WindowUnit::Rows => WindowUnit::Rows,
2598 ast::WindowUnit::Range => WindowUnit::Range,
2599 ast::WindowUnit::Groups => WindowUnit::Groups,
2600 },
2601 start: self.window_bound(ast, held.start, scope)?,
2602 end: self.window_bound(ast, held.end, scope)?,
2603 exclude: match held.exclude {
2604 ast::WindowExclude::NoOthers => WindowExclude::NoOthers,
2605 ast::WindowExclude::CurrentRow => WindowExclude::CurrentRow,
2606 ast::WindowExclude::Group => WindowExclude::Group,
2607 ast::WindowExclude::Ties => WindowExclude::Ties,
2608 },
2609 };
2610 Ok(WindowParts { args: bound, partition, order, frame })
2611 }
2612
2613 fn window_bound(
2615 &mut self,
2616 ast: &Ast,
2617 bound: ast::WindowBound,
2618 scope: &Scope,
2619 ) -> Result<WindowBound> {
2620 let offset = |binder: &mut Self, written| {
2621 let expr = binder.bind_expr(ast, written, scope)?;
2622 binder.over_aggregate(expr, scope)
2623 };
2624 Ok(match bound {
2625 ast::WindowBound::UnboundedPreceding => WindowBound::UnboundedPreceding,
2626 ast::WindowBound::CurrentRow => WindowBound::CurrentRow,
2627 ast::WindowBound::UnboundedFollowing => WindowBound::UnboundedFollowing,
2628 ast::WindowBound::Preceding(written) => WindowBound::Preceding(offset(self, written)?),
2629 ast::WindowBound::Following(written) => WindowBound::Following(offset(self, written)?),
2630 })
2631 }
2632
2633 fn is_window_output(&self, binding: ColumnBinding) -> bool {
2635 self.windows.iter().any(|run| run.index == binding.table)
2636 }
2637
2638 pub(crate) fn over_aggregate(&mut self, expr: ExprRef, scope: &Scope) -> Result<ExprRef> {
2644 let Some(aggregation) = self.aggregation.as_ref() else {
2645 return Ok(expr);
2646 };
2647 let index = aggregation.index;
2648 let groups = aggregation.groups.clone();
2649 for (at, group) in groups.iter().enumerate() {
2650 if self.same_expr(expr, *group) {
2651 let ty = self.plan.expr_type(*group).clone();
2652 return Ok(self.column(index, at, ty));
2653 }
2654 }
2655 let ty = self.plan.expr_type(expr).clone();
2656 match self.plan.expr(expr).clone() {
2657 Expr::Column(binding) if binding.table == index => Ok(expr),
2658 Expr::Column(binding) if self.is_window_output(binding) => Ok(expr),
2663 Expr::Column(binding) if self.joined_above.contains(&binding.table) => Ok(expr),
2668 Expr::Column(binding) => {
2669 let name =
2670 scope.columns.iter().find(|column| column.binding == binding).map_or_else(
2671 || "a column".to_string(),
2672 |column| format!("\"{}\"", column.name),
2673 );
2674 Err(Error::binder(format!(
2675 "column {name} must appear in the GROUP BY clause or must be part of an aggregate function"
2676 )))
2677 }
2678 Expr::Constant(_) | Expr::Aggregate { .. } | Expr::Window { .. } => Ok(expr),
2679 Expr::Cast { input, try_cast } => {
2680 let input = self.over_aggregate(input, scope)?;
2681 Ok(self.plan.add_expr(Expr::Cast { input, try_cast }, ty))
2682 }
2683 Expr::Compare { op, left, right } => {
2684 let left = self.over_aggregate(left, scope)?;
2685 let right = self.over_aggregate(right, scope)?;
2686 Ok(self.plan.add_expr(Expr::Compare { op, left, right }, ty))
2687 }
2688 Expr::Conjunction { op, children } => {
2689 let written = self.plan.expr_list(children).to_vec();
2690 let mut rewritten = Vec::with_capacity(written.len());
2691 for child in written {
2692 rewritten.push(self.over_aggregate(child, scope)?);
2693 }
2694 let children = self.plan.add_expr_list(&rewritten);
2695 Ok(self.plan.add_expr(Expr::Conjunction { op, children }, ty))
2696 }
2697 Expr::Function { name, args } => {
2698 let written = self.plan.expr_list(args).to_vec();
2699 let mut rewritten = Vec::with_capacity(written.len());
2700 for arg in written {
2701 rewritten.push(self.over_aggregate(arg, scope)?);
2702 }
2703 let args = self.plan.add_expr_list(&rewritten);
2704 Ok(self.plan.add_expr(Expr::Function { name, args }, ty))
2705 }
2706 Expr::Case { arms, otherwise } => {
2707 let written = self.plan.arm_list(arms).to_vec();
2708 let mut rewritten = Vec::with_capacity(written.len());
2709 for arm in written {
2710 let when = self.over_aggregate(arm.when, scope)?;
2711 let then = self.over_aggregate(arm.then, scope)?;
2712 rewritten.push(rudb_plan::Arm { when, then });
2713 }
2714 let otherwise = match otherwise {
2715 Some(expr) => Some(self.over_aggregate(expr, scope)?),
2716 None => None,
2717 };
2718 let arms = self.plan.add_arms(&rewritten);
2719 Ok(self.plan.add_expr(Expr::Case { arms, otherwise }, ty))
2720 }
2721 }
2722 }
2723
2724 pub(crate) fn same_expr(&self, left: ExprRef, right: ExprRef) -> bool {
2726 same_expr(&self.plan, left, right)
2727 }
2728}
2729
2730#[derive(Debug, Default)]
2740struct Options {
2741 binary_as_string: bool,
2744 all_varchar: bool,
2746 file_row_number: bool,
2751 given: Given,
2753}
2754
2755impl Options {
2756 fn of(written: &[(&'static str, Value, ExprRef)]) -> Result<Self> {
2763 let mut options = Self::default();
2764 for (parameter, value, _) in written {
2765 match (*parameter, value) {
2766 ("binary_as_string", Value::Boolean(on)) => options.binary_as_string = *on,
2767 ("all_varchar", Value::Boolean(on)) => options.all_varchar = *on,
2768 ("file_row_number", Value::Boolean(on)) => options.file_row_number = *on,
2769 _ => {}
2770 }
2771 }
2772 let named: Vec<(&str, Value)> =
2773 written.iter().map(|(parameter, value, _)| (*parameter, value.clone())).collect();
2774 options.given = csv_given(&named)?;
2775 Ok(options)
2776 }
2777}
2778
2779fn null_parameter(function: TableFunction, parameter: &str) -> String {
2788 match parameter {
2789 "header" => format!("\"{parameter}\" expects a non-null boolean value (e.g. TRUE or 1)"),
2790 "all_varchar" => format!("{} \"{parameter}\" cannot be NULL", function.name()),
2791 _ => format!("Cannot use NULL as argument to \"{parameter}\""),
2792 }
2793}
2794
2795fn missing_replacement(name: &str, input: &Scope) -> Error {
2800 Error::binder(format!(
2801 "Column \"{name}\" in REPLACE list not found in FROM clause{}",
2802 input.candidates()
2803 ))
2804}
2805
2806fn subtractable(ty: &LogicalType, ordering: bool) -> bool {
2815 if ty.is_numeric() {
2816 return true;
2817 }
2818 match ty {
2819 LogicalType::Date
2820 | LogicalType::Time
2821 | LogicalType::Timestamp
2822 | LogicalType::TimestampS
2823 | LogicalType::TimestampMs
2824 | LogicalType::TimestampNs
2825 | LogicalType::TimestampTz => true,
2826 LogicalType::TimeTz => ordering,
2827 _ => false,
2828 }
2829}
2830
2831fn refuse_fill(
2840 argument: &LogicalType,
2841 order: &[LogicalType],
2842 distinct: bool,
2843 ignore_nulls: bool,
2844) -> Result<()> {
2845 if !subtractable(argument, false) {
2846 return Err(Error::binder("FILL argument must support subtraction"));
2847 }
2848 let [key] = order else {
2849 return Err(Error::binder("FILL functions must have only one ORDER BY expression"));
2850 };
2851 if !subtractable(key, true) {
2852 return Err(Error::binder("FILL ordering must support subtraction"));
2853 }
2854 if distinct {
2855 return Err(Error::binder(
2856 "DISTINCT is not implemented for the window function \"\"fill\"\"",
2857 ));
2858 }
2859 if ignore_nulls {
2860 return Err(Error::binder(
2861 "RESPECT/IGNORE NULLS is not supported for the window function \"fill\"",
2862 ));
2863 }
2864 Ok(())
2865}
2866
2867fn window_signature(name: &str, types: &[LogicalType]) -> Result<Resolved> {
2874 match kind_of(name) {
2875 Some(FunctionKind::Aggregate | FunctionKind::Window) => resolve(name, types),
2876 Some(FunctionKind::Scalar) => {
2877 Err(Error::catalog(format!("{name} is not an aggregate function")))
2878 }
2879 None => Err(Error::catalog(format!("Aggregate Function with name {name} does not exist!"))),
2880 }
2881}
2882
2883fn same_expr(plan: &Plan, left: ExprRef, right: ExprRef) -> bool {
2885 if left == right {
2886 return true;
2887 }
2888 if plan.expr_type(left) != plan.expr_type(right) {
2889 return false;
2890 }
2891 let lists = |left, right| {
2892 let left: &[ExprRef] = plan.expr_list(left);
2893 let right: &[ExprRef] = plan.expr_list(right);
2894 left.len() == right.len()
2895 && left.iter().zip(right).all(|(&left, &right)| same_expr(plan, left, right))
2896 };
2897 match (plan.expr(left), plan.expr(right)) {
2898 (Expr::Column(left), Expr::Column(right)) => left == right,
2899 (Expr::Constant(left), Expr::Constant(right)) => plan.value(*left) == plan.value(*right),
2900 (
2901 Expr::Cast { input: left, try_cast: left_try },
2902 Expr::Cast { input: right, try_cast: right_try },
2903 ) => left_try == right_try && same_expr(plan, *left, *right),
2904 (
2905 Expr::Compare { op: left_op, left: left_a, right: left_b },
2906 Expr::Compare { op: right_op, left: right_a, right: right_b },
2907 ) => {
2908 left_op == right_op
2909 && same_expr(plan, *left_a, *right_a)
2910 && same_expr(plan, *left_b, *right_b)
2911 }
2912 (
2913 Expr::Conjunction { op: left_op, children: left_children },
2914 Expr::Conjunction { op: right_op, children: right_children },
2915 ) => left_op == right_op && lists(*left_children, *right_children),
2916 (
2917 Expr::Function { name: left_name, args: left_args },
2918 Expr::Function { name: right_name, args: right_args },
2919 ) => plan.string(*left_name) == plan.string(*right_name) && lists(*left_args, *right_args),
2920 (
2921 Expr::Aggregate {
2922 name: left_name,
2923 args: left_args,
2924 distinct: left_distinct,
2925 filter: left_filter,
2926 },
2927 Expr::Aggregate {
2928 name: right_name,
2929 args: right_args,
2930 distinct: right_distinct,
2931 filter: right_filter,
2932 },
2933 ) => {
2934 plan.string(*left_name) == plan.string(*right_name)
2935 && left_distinct == right_distinct
2936 && match (left_filter, right_filter) {
2937 (None, None) => true,
2938 (Some(left), Some(right)) => same_expr(plan, *left, *right),
2939 _ => false,
2940 }
2941 && lists(*left_args, *right_args)
2942 }
2943 (
2947 Expr::Window {
2948 name: left_name,
2949 args: left_args,
2950 distinct: left_distinct,
2951 filter: left_filter,
2952 ignore_nulls: left_nulls,
2953 },
2954 Expr::Window {
2955 name: right_name,
2956 args: right_args,
2957 distinct: right_distinct,
2958 filter: right_filter,
2959 ignore_nulls: right_nulls,
2960 },
2961 ) => {
2962 plan.string(*left_name) == plan.string(*right_name)
2963 && left_distinct == right_distinct
2964 && left_nulls == right_nulls
2965 && match (left_filter, right_filter) {
2966 (None, None) => true,
2967 (Some(left), Some(right)) => same_expr(plan, *left, *right),
2968 _ => false,
2969 }
2970 && lists(*left_args, *right_args)
2971 }
2972 (
2973 Expr::Case { arms: left_arms, otherwise: left_otherwise },
2974 Expr::Case { arms: right_arms, otherwise: right_otherwise },
2975 ) => {
2976 let left_arms = plan.arm_list(*left_arms);
2977 let right_arms = plan.arm_list(*right_arms);
2978 left_arms.len() == right_arms.len()
2979 && left_arms.iter().zip(right_arms).all(|(left, right)| {
2980 same_expr(plan, left.when, right.when) && same_expr(plan, left.then, right.then)
2981 })
2982 && match (left_otherwise, right_otherwise) {
2983 (None, None) => true,
2984 (Some(left), Some(right)) => same_expr(plan, *left, *right),
2985 _ => false,
2986 }
2987 }
2988 _ => false,
2989 }
2990}