1use std::sync::Arc;
16
17use rudb_catalog::{Catalog, Entry, FileStamp, QualifiedName, same_name};
18use rudb_common::bounds::Zones;
19use rudb_common::{
20 Error, Field, LogicalType, Result, Semantics, Session, ShowBehavior, Span, Stat, Value,
21};
22use rudb_functions::{
23 Columns, FILE_ROW_NUMBER, Footers, FunctionKind, Given, Resolved, TableFunction, csv_fields,
24 csv_given, files, is_file, is_pattern, kind_of, parquet_footers, parquet_outline, resolve,
25 resolve_pragma, resolve_table,
26};
27use rudb_kernels::{percentage, row_count};
28use rudb_parse::ast::{self, Ast, Distinct, LiteralKind, Nulls, Order, Quantifier, SetOp};
29use rudb_parse::{NONE, identifier_parts, parse_ast_with_case};
30use rudb_plan::{
31 Bound, BuildSide, ColumnBinding, ConjunctionOp, Expr, ExprRef, JoinKind, Node, NodeRef, Plan,
32 SetOpKind, Share, SortKey, WindowBound, WindowExclude, WindowFrame, WindowUnit,
33};
34
35use crate::expr::{describe, has_aggregate};
36use crate::fold;
37use crate::parameters::Parameters;
38use crate::scope::{Scope, Visible};
39
40pub fn bind(ast: &Ast, catalog: &Catalog) -> Result<Plan> {
47 bind_with(ast, catalog, &Parameters::new(), &Session::new())
48}
49
50pub fn bind_with(
59 ast: &Ast,
60 catalog: &Catalog,
61 parameters: &Parameters,
62 session: &Session,
63) -> Result<Plan> {
64 let query = match ast.statements.as_slice() {
65 [ast::Statement::Query(query)] => *query,
66 [] => return Err(Error::binder("no statement to bind")),
67 [_] => return Err(Error::not_implemented("a statement that is not a query")),
70 _ => return Err(Error::not_implemented("a script of more than one statement")),
71 };
72 let mut binder = Binder::with(catalog, parameters, session);
73 let (root, _) = binder.bind_query(ast, query)?;
74 let mut plan = binder.into_plan();
75 plan.set_root(root);
76 plan.validate()?;
77 Ok(plan)
78}
79
80pub fn bind_sql(query: &str, catalog: &Catalog) -> Result<Plan> {
86 bind_sql_with(query, catalog, &Session::new())
87}
88
89pub fn bind_sql_with(query: &str, catalog: &Catalog, session: &Session) -> Result<Plan> {
95 let ast = parse_ast_with_case(query, session.semantics().identifier_case())?;
96 bind_with(&ast, catalog, &Parameters::new(), session)
97}
98
99#[derive(Debug)]
101pub(crate) struct Aggregation {
102 pub(crate) index: u32,
104 pub(crate) groups: Vec<ExprRef>,
106 pub(crate) aggregates: Vec<ExprRef>,
108}
109
110#[derive(Debug)]
118pub(crate) struct WindowRun {
119 index: u32,
121 partition: Vec<ExprRef>,
123 order: Vec<SortKey>,
125 frame: WindowFrame,
127 calls: Vec<ExprRef>,
129}
130
131pub(crate) struct WindowCall<'a> {
136 pub(crate) name: &'a str,
138 pub(crate) args: &'a [ast::ExprRef],
140 pub(crate) distinct: bool,
142 pub(crate) filter: ast::ExprRef,
144 pub(crate) ignore_nulls: bool,
146 pub(crate) order: ast::Slice,
149 pub(crate) spec: ast::WindowRef,
151}
152
153struct WindowParts {
155 args: Vec<ExprRef>,
157 partition: Vec<ExprRef>,
159 order: Vec<SortKey>,
161 inner: Vec<SortKey>,
164 frame: WindowFrame,
166}
167
168#[derive(Debug)]
175struct Read {
176 fields: Vec<Field>,
178 rows: Stat<u64>,
180 distincts: Vec<(String, Stat<u64>)>,
182 zones: Option<Arc<dyn Zones>>,
184}
185
186impl Read {
187 fn uncounted(fields: Vec<Field>) -> Self {
189 Self { fields, rows: Stat::Unknown, distincts: Vec::new(), zones: None }
190 }
191}
192
193#[derive(Debug)]
195struct Materialized {
196 written: u32,
198 cte: u32,
200 name: String,
202 fields: Vec<Field>,
204}
205
206#[derive(Debug)]
207pub(crate) struct PendingSubquery {
208 pub(crate) node: NodeRef,
209 pub(crate) kind: JoinKind,
210 pub(crate) conditions: Vec<ExprRef>,
211 pub(crate) dependent: bool,
212 pub(crate) reads: Vec<ColumnBinding>,
218 pub(crate) index: u32,
224 pub(crate) inside_aggregate: bool,
231}
232
233#[derive(Debug, Clone, Copy, PartialEq, Eq)]
235enum Side {
236 Left,
237 Right,
238}
239
240#[derive(Debug)]
242pub(crate) struct Binder<'a> {
243 catalog: &'a Catalog,
244 pub(crate) parameters: &'a Parameters,
246 pub(crate) session: &'a Session,
248 pub(crate) semantics: Semantics,
250 plan: Plan,
251 next_index: u32,
252 pub(crate) current_span: Span,
254 pub(crate) aggregation: Option<Aggregation>,
256 want_ascending: bool,
259 pub(crate) upsert: bool,
262 pub(crate) insert_defaults: Option<Vec<(LogicalType, Option<String>)>>,
265 pub(crate) default_as_null: bool,
268 pub(crate) in_aggregate: bool,
270 pub(crate) in_filter: bool,
272 pub(crate) windows: Vec<WindowRun>,
274 pub(crate) sequences: Vec<QualifiedName>,
276 pub(crate) in_window: bool,
278 pub(crate) scalar_subqueries: Vec<PendingSubquery>,
280 pub(crate) joined_above: Vec<u32>,
286 pub(crate) outer_scopes: Vec<Scope>,
287 pub(crate) lateral_scopes: Vec<usize>,
294 pub(crate) correlations: Vec<Vec<ColumnBinding>>,
295 pub(crate) lambda_frames: Vec<crate::lambda::Frame>,
297 pub(crate) clause: &'static str,
299 pub(crate) outlined: bool,
308 expanding: Vec<String>,
310 materialized: Vec<Materialized>,
316 next_cte: u32,
318 started: Option<i64>,
320}
321
322impl<'a> Binder<'a> {
323 pub(crate) fn with(
324 catalog: &'a Catalog,
325 parameters: &'a Parameters,
326 session: &'a Session,
327 ) -> Self {
328 Self {
329 catalog,
330 parameters,
331 session,
332 semantics: session.semantics(),
333 plan: Plan::new(),
334 next_index: 0,
335 current_span: Span::new(0, 0),
336 aggregation: None,
337 want_ascending: false,
338 upsert: false,
339 insert_defaults: None,
340 default_as_null: false,
341 in_aggregate: false,
342 in_filter: false,
343 windows: Vec::new(),
344 sequences: Vec::new(),
345 in_window: false,
346 scalar_subqueries: Vec::new(),
347 joined_above: Vec::new(),
348 outer_scopes: Vec::new(),
349 lateral_scopes: Vec::new(),
350 correlations: Vec::new(),
351 lambda_frames: Vec::new(),
352 clause: "SELECT clause",
353 outlined: false,
354 expanding: Vec::new(),
355 materialized: Vec::new(),
356 next_cte: 0,
357 started: None,
358 }
359 }
360
361 pub(crate) fn catalog(&self) -> &Catalog {
362 self.catalog
363 }
364
365 pub(crate) fn instant(&mut self) -> i64 {
372 *self.started.get_or_insert_with(crate::context::micros_now)
373 }
374
375 pub(crate) fn plan(&self) -> &Plan {
376 &self.plan
377 }
378
379 pub(crate) fn plan_mut(&mut self) -> &mut Plan {
380 &mut self.plan
381 }
382
383 pub(crate) fn add_expr(&mut self, expr: Expr, ty: LogicalType) -> ExprRef {
384 self.plan.add_expr_at(expr, ty, self.current_span)
385 }
386
387 pub(crate) fn add_constant(&mut self, value: Value) -> ExprRef {
388 let ty = value.logical_type();
389 let reference = self.plan.add_value(value);
390 self.plan.add_expr_at(Expr::Constant(reference), ty, self.current_span)
391 }
392
393 pub(crate) fn add_node(&mut self, node: Node) -> NodeRef {
394 self.plan.add_node_at(node, self.current_span)
395 }
396
397 pub(crate) fn into_plan(self) -> Plan {
398 self.plan
399 }
400
401 pub(crate) fn fresh_index(&mut self) -> u32 {
403 let index = self.next_index;
404 self.next_index += 1;
405 index
406 }
407
408 fn column(&mut self, index: u32, position: usize, ty: LogicalType) -> ExprRef {
410 let binding = ColumnBinding::new(index, position as u32);
411 self.plan.add_expr(Expr::Column(binding), ty)
412 }
413
414 fn attach_scalar_subqueries(&mut self, mut input: NodeRef) -> NodeRef {
416 let subqueries = std::mem::take(&mut self.scalar_subqueries);
417 for pending in subqueries {
418 input = self.attach_subquery(input, pending);
419 }
420 input
421 }
422
423 fn attach_subquery(&mut self, input: NodeRef, pending: PendingSubquery) -> NodeRef {
430 let PendingSubquery {
431 node: mut right,
432 kind,
433 conditions,
434 dependent,
435 reads: _,
436 index: _,
437 inside_aggregate: _,
438 } = pending;
439 if kind == JoinKind::Single && !self.semantics.scalar_subquery_error_on_multiple_rows() {
440 right = self.add_node(Node::Limit {
441 input: right,
442 count: Bound::Rows(1),
443 offset: Bound::Rows(0),
444 });
445 }
446 let conditions = self.plan.add_expr_list(&conditions);
447 if dependent {
448 self.add_node(Node::DependentJoin { left: input, right, kind, conditions })
449 } else {
450 self.add_node(Node::Join {
451 left: input,
452 right,
453 kind,
454 conditions,
455 build: BuildSide::default(),
456 })
457 }
458 }
459
460 pub(crate) fn bind_query(
463 &mut self,
464 ast: &Ast,
465 query: ast::QueryRef,
466 ) -> Result<(NodeRef, Scope)> {
467 let span = ast.query_span(query);
468 let outer = std::mem::replace(&mut self.current_span, span);
469 let result =
470 self.bind_query_inner(ast, query).map_err(|error| error.with_fallback_span(span));
471 self.current_span = outer;
472 result
473 }
474
475 fn bind_query_inner(&mut self, ast: &Ast, query: ast::QueryRef) -> Result<(NodeRef, Scope)> {
476 let written = ast.query(query);
477 if written.ctes.is_empty() {
478 return self.bind_body(ast, &written);
479 }
480 let depth = self.materialized.len();
484 let result = self.bind_materialized(ast, &written);
485 self.materialized.truncate(depth);
486 result
487 }
488
489 fn bind_materialized(&mut self, ast: &Ast, written: &ast::Query) -> Result<(NodeRef, Scope)> {
495 let depth = self.materialized.len();
496 let held = ast.cte_list(written.ctes).to_vec();
497 let mut definitions = Vec::with_capacity(held.len());
498 for &index in &held {
499 definitions.push(self.bind_definition(ast, index)?);
500 }
501 let (mut node, scope) = self.bind_body(ast, written)?;
502 for (at, definition) in definitions.into_iter().enumerate().rev() {
503 let entry = &self.materialized[depth + at];
504 let cte = entry.cte;
505 let name = entry.name.clone();
506 let fields = entry.fields.clone();
507 let name = self.plan.intern(&name);
508 let columns = self.plan.add_fields(&fields);
509 node =
510 self.add_node(Node::MaterializedCte { definition, body: node, name, cte, columns });
511 }
512 Ok((node, scope))
513 }
514
515 fn bind_definition(&mut self, ast: &Ast, index: u32) -> Result<NodeRef> {
525 let held = ast.cte(index);
526 let name = ast.string(held.name).to_string();
527 let (node, mut scope) = self.bind_query(ast, held.query)?;
528 if !held.columns.is_empty() {
529 let names: Vec<&str> = ast.name(held.columns).collect();
530 scope.rename_prefix(&names);
531 }
532 let table = self.fresh_index();
533 let mut exprs = Vec::with_capacity(scope.len());
534 let mut names = Vec::with_capacity(scope.len());
535 for column in &scope.columns {
536 exprs.push(self.plan.add_expr(Expr::Column(column.binding), column.ty.clone()));
537 names.push(self.plan.intern(&column.name));
538 }
539 let exprs = self.plan.add_expr_list(&exprs);
540 let names = self.plan.add_name_list(&names);
541 let node = self.add_node(Node::Project { input: node, index: table, exprs, names });
542 let cte = self.next_cte;
543 self.next_cte += 1;
544 self.materialized.push(Materialized { written: index, cte, name, fields: scope.fields() });
545 Ok(node)
546 }
547
548 fn bind_body(&mut self, ast: &Ast, written: &ast::Query) -> Result<(NodeRef, Scope)> {
549 match written.body {
550 ast::QueryBody::Select(select) => self.bind_select(ast, select, written),
551 ast::QueryBody::SetOp { op, quantifier, by_name, left, right } => {
552 let operator = Operator { op, quantifier, by_name };
553 self.bind_set_op(ast, written, operator, left, right)
554 }
555 ast::QueryBody::Values(rows) => self.bind_values(ast, written, rows),
556 ast::QueryBody::Describe(inner) => self.bind_describe(ast, written, inner),
557 ast::QueryBody::Show { name, relation } => self.bind_show(ast, written, name, relation),
558 }
559 }
560
561 fn bind_show(
563 &mut self,
564 ast: &Ast,
565 query: &ast::Query,
566 name: ast::Slice,
567 relation: ast::QueryRef,
568 ) -> Result<(NodeRef, Scope)> {
569 let text = ast.name_text(name);
570 let parts: Vec<&str> = ast.name(name).collect();
571 let table_exists = self.catalog.resolve(&parts).is_ok();
572 let as_table = match self.semantics.show_behavior() {
573 ShowBehavior::Auto => table_exists,
574 ShowBehavior::Setting => false,
575 ShowBehavior::Table => true,
576 };
577 if as_table {
578 return self.bind_describe(ast, query, relation);
579 }
580 let shown = match self.session.iter().find(|(name, _)| name.eq_ignore_ascii_case(&text)) {
585 Some((_, value)) => value.to_string(),
586 None => match self.beyond(&text)? {
587 Some(Value::Varchar(declared)) => declared,
588 Some(other) => other.to_string(),
589 None => {
590 return Err(Error::catalog(format!(
591 "Setting with name \"{text}\" does not exist"
592 )));
593 }
594 },
595 };
596 let field = Field::new(text, LogicalType::Varchar);
597 let expr = self.plan.add_constant(Value::Varchar(shown));
598 let row = self.plan.add_expr_list(&[expr]);
599 let rows = self.plan.add_rows(&[row]);
600 let columns = self.plan.add_fields(std::slice::from_ref(&field));
601 let index = self.fresh_index();
602 let node = self.add_node(Node::Values { index, columns, rows });
603 let mut scope = Scope::empty();
604 scope.push(Visible {
605 table: String::new(),
606 name: field.name,
607 binding: ColumnBinding::new(index, 0),
608 ty: LogicalType::Varchar,
609 not_null: false,
610 key: None,
611 default: None,
612 qualified: false,
613 also: None,
614 });
615 Ok((node, scope))
616 }
617
618 fn bind_describe(
633 &mut self,
634 ast: &Ast,
635 query: &ast::Query,
636 inner: ast::QueryRef,
637 ) -> Result<(NodeRef, Scope)> {
638 let (_, described) = self.bind_query(ast, inner)?;
639 let fields: Vec<Field> = ["column_name", "column_type", "null", "key", "default", "extra"]
640 .iter()
641 .map(|name| Field::new(*name, LogicalType::Varchar))
642 .collect();
643 let mut slices = Vec::with_capacity(described.columns.len());
644 for column in described.columns.clone() {
645 let written = [
648 column.name.clone(),
649 column.ty.to_string(),
650 if column.not_null { "NO" } else { "YES" }.to_owned(),
651 ];
652 let mut items: Vec<ExprRef> = written
653 .into_iter()
654 .map(|text| self.plan.add_constant(Value::Varchar(text)))
655 .collect();
656 let mark = match column.key {
657 Some(mark) => self.plan.add_constant(Value::Varchar(mark.to_owned())),
658 None => {
659 let empty = self.plan.add_constant(Value::Null);
660 self.cast_to(empty, &LogicalType::Varchar)
661 }
662 };
663 items.push(mark);
664 if let Some(default) = &column.default {
665 items.push(self.plan.add_constant(Value::Varchar(default.clone())));
666 }
667 while items.len() < 6 {
668 let empty = self.plan.add_constant(Value::Null);
669 items.push(self.cast_to(empty, &LogicalType::Varchar));
670 }
671 slices.push(self.plan.add_expr_list(&items));
672 }
673 let rows = self.plan.add_rows(&slices);
674 let columns = self.plan.add_fields(&fields);
675 let index = self.fresh_index();
676 let mut node = self.add_node(Node::Values { index, columns, rows });
677 let mut scope = Scope::empty();
678 for (at, field) in fields.iter().enumerate() {
679 scope.push(Visible {
680 table: String::new(),
681 name: field.name.clone(),
682 binding: ColumnBinding::new(index, at as u32),
683 ty: field.ty.clone(),
684 not_null: false,
685 key: None,
686 default: None,
687 qualified: false,
688 also: None,
689 });
690 }
691 let keys = self.sort_keys(ast, query, &scope, &[])?;
692 if !keys.is_empty() {
693 let keys = self.plan.add_sort_keys(&keys);
694 node = self.add_node(Node::Sort { input: node, keys });
695 }
696 node = self.apply_limit(ast, query, node, &mut scope)?;
697 Ok((node, scope))
698 }
699
700 pub(crate) fn bind_default(&mut self, text: Option<&str>, ty: &LogicalType) -> Result<ExprRef> {
704 let Some(text) = text else {
705 let value = self.plan.add_value(Value::Null);
706 return Ok(self.plan.add_expr(Expr::Constant(value), ty.clone()));
707 };
708 let ast = rudb_parse::parse_ast(&format!("SELECT {text}"))?;
709 let found = match ast.statements.first() {
710 Some(&ast::Statement::Query(query)) => match ast.query(query).body {
711 ast::QueryBody::Select(select) => {
712 ast.target_list(ast.select(select).targets).first().map(|target| target.expr)
713 }
714 _ => None,
715 },
716 _ => None,
717 };
718 let Some(expr) = found else {
719 return Err(Error::internal(format!("a default that is not an expression: {text}")));
720 };
721 let expr = self.bind_expr(&ast, expr, &Scope::empty())?;
722 self.checked_cast_to(expr, ty, false)
723 }
724
725 fn passes_through(&self, expr: ExprRef, input: &Scope) -> bool {
731 let Expr::Column(binding) = *self.plan.expr(expr) else { return false };
732 input.columns.iter().any(|column| column.binding == binding && column.not_null)
733 }
734
735 fn key_through(&self, expr: ExprRef, input: &Scope) -> Option<&'static str> {
739 self.through(expr, input).and_then(|column| column.key)
740 }
741
742 fn through<'s>(&self, expr: ExprRef, input: &'s Scope) -> Option<&'s Visible> {
744 let Expr::Column(binding) = *self.plan.expr(expr) else { return None };
745 input.columns.iter().find(|column| column.binding == binding)
746 }
747
748 fn bind_values(
755 &mut self,
756 ast: &Ast,
757 query: &ast::Query,
758 rows: ast::Slice,
759 ) -> Result<(NodeRef, Scope)> {
760 let written = ast.rows(rows).to_vec();
761 let Some(first) = written.first() else {
762 return Err(Error::binder("VALUES needs at least one row"));
763 };
764 let width = first.len as usize;
765 for (at, row) in written.iter().enumerate() {
766 if row.len as usize != width {
767 return Err(Error::binder(format!(
768 "VALUES lists must all be the same length, expected {width} columns but row {} has {}",
769 at + 1,
770 row.len
771 )));
772 }
773 }
774 let empty = Scope::empty();
776 let defaults = self.insert_defaults.take();
777 let previous = std::mem::replace(&mut self.clause, "VALUES clause");
778 let mut bound: Vec<Vec<ExprRef>> = Vec::with_capacity(written.len());
779 for row in &written {
780 let mut items = Vec::with_capacity(width);
781 for (at, &expr) in ast.expr_list(*row).iter().enumerate() {
782 let column = defaults.as_ref().and_then(|defaults| defaults.get(at));
783 items.push(match (ast.expr(expr), column) {
784 (ast::Expr::Default, Some((ty, default))) => {
785 self.bind_default(default.as_deref(), ty)?
786 }
787 _ => self.bind_expr(ast, expr, &empty)?,
788 });
789 }
790 bound.push(items);
791 }
792 self.clause = previous;
793 let mut types = Vec::with_capacity(width);
794 for at in 0..width {
795 let mut ty = self.plan.expr_type(bound[0][at]).clone();
796 for row in &bound[1..] {
797 let other = self.plan.expr_type(row[at]).clone();
798 ty = ty.promote(&other).ok_or_else(|| {
799 Error::binder(format!(
800 "Cannot combine a value of type {ty} with a value of type {other} in column {} of a VALUES",
801 at + 1
802 ))
803 })?;
804 }
805 types.push(ty);
806 }
807 let mut slices = Vec::with_capacity(bound.len());
808 for row in &bound {
809 let items: Vec<ExprRef> = row
810 .iter()
811 .zip(&types)
812 .map(|(&expr, ty)| self.checked_cast_to(expr, ty, false))
813 .collect::<Result<_>>()?;
814 slices.push(self.plan.add_expr_list(&items));
815 }
816 let rows = self.plan.add_rows(&slices);
817 let fields: Vec<Field> = types
818 .iter()
819 .enumerate()
820 .map(|(at, ty)| Field::new(format!("col{at}"), ty.clone()))
821 .collect();
822 let columns = self.plan.add_fields(&fields);
823 let index = self.fresh_index();
824 let mut node = self.add_node(Node::Values { index, columns, rows });
825 let mut scope = Scope::empty();
826 for (at, field) in fields.iter().enumerate() {
827 scope.push(Visible {
828 table: String::new(),
829 name: field.name.clone(),
830 binding: ColumnBinding::new(index, at as u32),
831 ty: field.ty.clone(),
832 not_null: false,
833 key: None,
834 default: None,
835 qualified: false,
836 also: None,
837 });
838 }
839 let keys = self.sort_keys(ast, query, &scope, &[])?;
840 if !keys.is_empty() {
841 let keys = self.plan.add_sort_keys(&keys);
842 node = self.add_node(Node::Sort { input: node, keys });
843 }
844 node = self.apply_limit(ast, query, node, &mut scope)?;
845 Ok((node, scope))
846 }
847
848 fn bind_set_op(
849 &mut self,
850 ast: &Ast,
851 query: &ast::Query,
852 operator: Operator,
853 left: ast::QueryRef,
854 right: ast::QueryRef,
855 ) -> Result<(NodeRef, Scope)> {
856 let (left_node, left_scope) = self.bind_query(ast, left)?;
857 let (right_node, right_scope) = self.bind_query(ast, right)?;
858 let merged = if operator.by_name {
859 match_by_name(&left_scope, &right_scope)?
860 } else {
861 match_by_position(&left_scope, &right_scope)?
862 };
863 let left_node = self.conform(left_node, &left_scope, &merged, |column| column.left)?;
864 let right_node = self.conform(right_node, &right_scope, &merged, |column| column.right)?;
865 let index = self.fresh_index();
866 let kind = match operator.op {
867 SetOp::Union => SetOpKind::Union,
868 SetOp::Except => SetOpKind::Except,
869 SetOp::Intersect => SetOpKind::Intersect,
870 };
871 let all = operator.quantifier == Quantifier::All;
874 let mut node =
875 self.add_node(Node::SetOp { left: left_node, right: right_node, kind, all, index });
876 let mut scope = Scope::empty();
877 for (at, column) in merged.iter().enumerate() {
878 scope.push(Visible {
879 table: String::new(),
880 name: column.name.clone(),
881 binding: ColumnBinding::new(index, at as u32),
882 ty: column.ty.clone(),
883 not_null: false,
886 key: None,
887 default: None,
888 qualified: false,
889 also: None,
890 });
891 }
892 let keys = self.sort_keys(ast, query, &scope, &[])?;
896 if !keys.is_empty() {
897 let keys = self.plan.add_sort_keys(&keys);
898 node = self.add_node(Node::Sort { input: node, keys });
899 }
900 node = self.apply_limit(ast, query, node, &mut scope)?;
901 Ok((node, scope))
902 }
903
904 fn conform(
910 &mut self,
911 node: NodeRef,
912 scope: &Scope,
913 merged: &[Merged],
914 pick: impl Fn(&Merged) -> Option<usize>,
915 ) -> Result<NodeRef> {
916 let unchanged = merged.len() == scope.len()
917 && merged
918 .iter()
919 .enumerate()
920 .all(|(at, column)| pick(column) == Some(at) && column.ty == scope.columns[at].ty);
921 if unchanged {
922 return Ok(node);
923 }
924 let index = self.fresh_index();
925 let mut exprs = Vec::with_capacity(merged.len());
926 let mut names = Vec::with_capacity(merged.len());
927 for column in merged {
928 let expr = match pick(column) {
929 Some(at) => {
930 let held = &scope.columns[at];
931 self.plan.add_expr(Expr::Column(held.binding), held.ty.clone())
932 }
933 None => self.plan.add_constant(Value::Null),
934 };
935 exprs.push(self.checked_cast_to(expr, &column.ty, false)?);
936 names.push(self.plan.intern(&column.name));
937 }
938 let exprs = self.plan.add_expr_list(&exprs);
939 let names = self.plan.add_name_list(&names);
940 Ok(self.add_node(Node::Project { input: node, index, exprs, names }))
941 }
942
943 fn bind_select(
946 &mut self,
947 ast: &Ast,
948 select: ast::SelectRef,
949 query: &ast::Query,
950 ) -> Result<(NodeRef, Scope)> {
951 let written = ast.select(select);
952 self.want_ascending |= !written.group_by.is_empty() || written.group_by_all;
953 let outer_windows = std::mem::take(&mut self.windows);
957 let outer_joined_above = std::mem::take(&mut self.joined_above);
962 let (mut node, input) = self.bind_from(ast, written.from)?;
963 node = self.attach_scalar_subqueries(node);
964
965 if written.filter != NONE {
966 self.clause = "WHERE clause";
967 let predicate = self.bind_expr(ast, written.filter, &input)?;
968 let predicate = self.as_boolean(predicate, "WHERE")?;
969 node = self.attach_scalar_subqueries(node);
970 node = self.add_node(Node::Filter { input: node, predicate });
971 }
972
973 let targets = ast.target_list(written.targets).to_vec();
974 if targets.is_empty() {
975 return Err(Error::binder("a SELECT needs at least one expression to select"));
976 }
977
978 let group_items = self.group_items(ast, &written, &targets)?;
979 let aggregating = !group_items.is_empty()
980 || written.having != NONE
981 || targets.iter().any(|target| has_aggregate(ast, target.expr));
982 if aggregating {
983 self.clause = "GROUP BY clause";
984 let mut groups = Vec::with_capacity(group_items.len());
985 for item in &group_items {
986 groups.push(self.bind_expr(ast, *item, &input)?);
987 }
988 let index = self.fresh_index();
989 self.aggregation = Some(Aggregation { index, groups, aggregates: Vec::new() });
990 }
991
992 let mut above = Vec::new();
999
1000 self.clause = "SELECT clause";
1001 let (mut exprs, mut names) = self.bind_targets(ast, &targets, &input, &mut above)?;
1002 let visible = exprs.len();
1003
1004 let mut having = None;
1005 if written.having != NONE {
1006 self.clause = "HAVING clause";
1007 let before = self.scalar_subqueries.len();
1008 let predicate = self.bind_expr(ast, written.having, &input)?;
1009 self.lift_over_aggregate(before, &mut above, &input)?;
1010 let predicate = self.over_aggregate(predicate, &input)?;
1011 having = Some(self.as_boolean(predicate, "HAVING")?);
1012 }
1013
1014 let project = self.fresh_index();
1017 let mut output = Scope::empty();
1018 for (at, (expr, name)) in exprs.iter().zip(&names).enumerate() {
1019 output.push(Visible {
1020 table: String::new(),
1021 name: name.clone(),
1022 binding: ColumnBinding::new(project, at as u32),
1023 ty: self.plan.expr_type(*expr).clone(),
1024 not_null: self.passes_through(*expr, &input),
1025 key: self.key_through(*expr, &input),
1026 default: self.through(*expr, &input).and_then(|column| column.default.clone()),
1027 qualified: false,
1028 also: None,
1029 });
1030 }
1031
1032 self.clause = "ORDER BY clause";
1033 let mut extra = Vec::new();
1034 let keys = self.select_sort_keys(
1035 ast, query, &input, &output, project, &mut exprs, &mut names, &mut extra, &mut above,
1036 )?;
1037 self.joined_above = outer_joined_above;
1038 if !extra.is_empty() && written.distinct != Distinct::No {
1039 return Err(Error::binder(
1040 "For SELECT DISTINCT, ORDER BY expressions must appear in the select list",
1041 ));
1042 }
1043 let on = self.distinct_on(ast, written.distinct, &output)?;
1044
1045 node = self.attach_scalar_subqueries(node);
1046
1047 if let Some(aggregation) = self.aggregation.take() {
1048 let index = aggregation.index;
1049 let groups = self.plan.add_expr_list(&aggregation.groups);
1050 let aggregates = self.plan.add_expr_list(&aggregation.aggregates);
1051 node = self.add_node(Node::Aggregate { input: node, index, groups, aggregates });
1052 }
1053 if !above.is_empty() {
1054 debug_assert!(self.scalar_subqueries.is_empty(), "a query is waiting to be joined");
1055 self.scalar_subqueries = above;
1056 node = self.attach_scalar_subqueries(node);
1057 }
1058 if let Some(predicate) = having {
1059 node = self.add_node(Node::Filter { input: node, predicate });
1060 }
1061
1062 for run in std::mem::replace(&mut self.windows, outer_windows) {
1066 let partition = self.plan.add_expr_list(&run.partition);
1067 let order = self.plan.add_sort_keys(&run.order);
1068 let expressions = self.plan.add_expr_list(&run.calls);
1069 node = self.add_node(Node::Window {
1070 input: node,
1071 index: run.index,
1072 partition,
1073 order,
1074 frame: run.frame,
1075 expressions,
1076 });
1077 }
1078
1079 let interned: Vec<u32> = names.iter().map(|name| self.plan.intern(name)).collect();
1080 let exprs_slice = self.plan.add_expr_list(&exprs);
1081 let names_slice = self.plan.add_name_list(&interned);
1082 node = self.add_node(Node::Project {
1083 input: node,
1084 index: project,
1085 exprs: exprs_slice,
1086 names: names_slice,
1087 });
1088
1089 if written.distinct != Distinct::No {
1090 let on = self.plan.add_expr_list(&on);
1091 node = self.add_node(Node::Distinct { input: node, on });
1092 }
1093 if !keys.is_empty() {
1094 let keys = self.plan.add_sort_keys(&keys);
1095 node = self.add_node(Node::Sort { input: node, keys });
1096 }
1097 node = self.apply_limit(ast, query, node, &mut output)?;
1098
1099 if extra.is_empty() {
1100 output.columns.truncate(visible);
1101 return Ok((node, output));
1102 }
1103 let index = self.fresh_index();
1106 let mut kept = Vec::with_capacity(visible);
1107 let mut kept_names = Vec::with_capacity(visible);
1108 let mut scope = Scope::empty();
1109 for (at, name) in names.iter().enumerate().take(visible) {
1110 let ty = output.columns[at].ty.clone();
1111 let binding = output.columns[at].binding;
1115 kept.push(self.plan.add_expr(Expr::Column(binding), ty.clone()));
1116 kept_names.push(self.plan.intern(name));
1117 scope.push(Visible {
1118 table: String::new(),
1119 name: name.clone(),
1120 binding: ColumnBinding::new(index, at as u32),
1121 ty,
1122 not_null: output.columns[at].not_null,
1123 key: output.columns[at].key,
1124 default: output.columns[at].default.clone(),
1125 qualified: false,
1126 also: None,
1127 });
1128 }
1129 let exprs = self.plan.add_expr_list(&kept);
1130 let names = self.plan.add_name_list(&kept_names);
1131 node = self.add_node(Node::Project { input: node, index, exprs, names });
1132 Ok((node, scope))
1133 }
1134
1135 fn lift_over_aggregate(
1154 &mut self,
1155 before: usize,
1156 above: &mut Vec<PendingSubquery>,
1157 scope: &Scope,
1158 ) -> Result<()> {
1159 if self.aggregation.is_none() {
1160 return Ok(());
1161 }
1162 let mut lifted = Vec::new();
1163 for mut pending in self.scalar_subqueries.split_off(before) {
1164 let stays = pending.inside_aggregate
1165 || (pending.dependent && !self.lift_correlated(&mut pending));
1166 if stays {
1167 self.scalar_subqueries.push(pending);
1168 } else {
1169 self.joined_above.push(pending.index);
1170 lifted.push(pending);
1171 }
1172 }
1173 for pending in &mut lifted {
1178 let conditions = std::mem::take(&mut pending.conditions);
1179 let mut over = Vec::with_capacity(conditions.len());
1180 for condition in conditions {
1181 over.push(self.over_aggregate(condition, scope)?);
1182 }
1183 pending.conditions = over;
1184 }
1185 above.append(&mut lifted);
1186 Ok(())
1187 }
1188
1189 fn lift_correlated(&mut self, pending: &mut PendingSubquery) -> bool {
1206 let Some(index) = self.aggregation.as_ref().map(|aggregation| aggregation.index) else {
1207 return false;
1208 };
1209 let mut moved = Vec::with_capacity(pending.reads.len());
1210 for read in &pending.reads {
1211 let Some(at) = self.group_of(*read) else {
1212 return false;
1213 };
1214 moved.push((*read, ColumnBinding::new(index, at as u32)));
1215 }
1216 let mut rewrites = Vec::new();
1217 self.plan.subtree_columns(pending.node, &mut |reference, binding| {
1218 if let Some(&(_, to)) = moved.iter().find(|(from, _)| *from == binding) {
1219 rewrites.push((reference, to));
1220 }
1221 });
1222 for (reference, to) in rewrites {
1223 self.plan.rebind(reference, to);
1224 }
1225 pending.reads = moved.into_iter().map(|(_, to)| to).collect();
1226 true
1227 }
1228
1229 fn bind_targets(
1230 &mut self,
1231 ast: &Ast,
1232 targets: &[ast::Target],
1233 input: &Scope,
1234 above: &mut Vec<PendingSubquery>,
1235 ) -> Result<(Vec<ExprRef>, Vec<String>)> {
1236 let mut exprs = Vec::with_capacity(targets.len());
1237 let mut names = Vec::with_capacity(targets.len());
1238 for target in targets {
1239 if let ast::Expr::Star { qualifier, replacements } = ast.expr(target.expr) {
1240 let table = ast.name(qualifier).last().map(str::to_string);
1241 let expanded: Vec<Visible> =
1242 input.star(table.as_deref())?.into_iter().cloned().collect();
1243 let replacements = ast.target_list(replacements).to_vec();
1244 let mut used = vec![false; replacements.len()];
1245 for column in expanded {
1246 let found = replacements.iter().zip(&mut used).find(|(replacement, _)| {
1247 same_name(ast.string(replacement.alias), &column.name)
1248 });
1249 let before = self.scalar_subqueries.len();
1254 let (expr, name) = match found {
1255 Some((replacement, used)) => {
1256 *used = true;
1257 let expr = self.bind_expr(ast, replacement.expr, input)?;
1258 (expr, ast.string(replacement.alias).to_string())
1259 }
1260 None => (
1261 self.plan.add_expr(Expr::Column(column.binding), column.ty),
1262 column.name,
1263 ),
1264 };
1265 self.lift_over_aggregate(before, above, input)?;
1266 exprs.push(self.over_aggregate(expr, input)?);
1267 names.push(name);
1268 }
1269 if let Some((replacement, _)) =
1273 replacements.iter().zip(&used).find(|(_, used)| !**used)
1274 {
1275 return Err(missing_replacement(ast.string(replacement.alias), input));
1276 }
1277 continue;
1278 }
1279 let before = self.scalar_subqueries.len();
1280 let expr = self.bind_expr(ast, target.expr, input)?;
1281 self.lift_over_aggregate(before, above, input)?;
1282 exprs.push(self.over_aggregate(expr, input)?);
1283 names.push(if target.alias == NONE {
1284 self.output_name(ast, target.expr, input)
1285 } else {
1286 ast.string(target.alias).to_string()
1287 });
1288 }
1289 Ok((exprs, names))
1290 }
1291
1292 fn output_name(&self, ast: &Ast, target: ast::ExprRef, input: &Scope) -> String {
1298 if let ast::Expr::Column { name } = ast.expr(target) {
1299 let parts: Vec<&str> = ast.name(name).collect();
1300 if let Ok(found) = input.resolve(&parts) {
1301 let written = parts.last().copied().unwrap_or_default();
1304 if let Some(also) = &found.also {
1305 if !same_name(&found.name, written) && same_name(also, written) {
1306 return also.clone();
1307 }
1308 }
1309 return found.name.clone();
1310 }
1311 }
1312 describe(ast, target, self.semantics)
1313 }
1314
1315 fn group_items(
1317 &self,
1318 ast: &Ast,
1319 select: &ast::Select,
1320 targets: &[ast::Target],
1321 ) -> Result<Vec<ast::ExprRef>> {
1322 if select.group_by_all {
1323 return Ok(targets
1326 .iter()
1327 .filter(|target| !has_aggregate(ast, target.expr))
1328 .map(|target| target.expr)
1329 .collect());
1330 }
1331 let mut items = Vec::new();
1332 for &item in ast.expr_list(select.group_by) {
1333 items.push(self.output_reference(ast, item, targets, "GROUP BY")?.unwrap_or(item));
1334 }
1335 Ok(items)
1336 }
1337
1338 fn output_reference(
1340 &self,
1341 ast: &Ast,
1342 item: ast::ExprRef,
1343 targets: &[ast::Target],
1344 clause: &str,
1345 ) -> Result<Option<ast::ExprRef>> {
1346 match ast.expr(item) {
1347 ast::Expr::Literal { kind: LiteralKind::Number, text } => {
1348 let written = ast.string(text);
1349 let position: usize = written.parse().map_err(|_| {
1350 Error::binder(format!("{clause} term {written} is not a column"))
1351 })?;
1352 if position == 0 || position > targets.len() {
1353 return Err(Error::binder(format!(
1354 "{clause} term out of range - should be between 1 and {}",
1355 targets.len()
1356 )));
1357 }
1358 Ok(Some(targets[position - 1].expr))
1359 }
1360 ast::Expr::Column { name } => {
1361 let parts: Vec<&str> = ast.name(name).collect();
1362 let [written] = parts.as_slice() else { return Ok(None) };
1363 let mut found = None;
1364 for target in targets {
1365 if target.alias != NONE && same_name(ast.string(target.alias), written) {
1366 if found.is_some() {
1367 return Ok(None);
1368 }
1369 found = Some(target.expr);
1370 }
1371 }
1372 Ok(found)
1373 }
1374 _ => Ok(None),
1375 }
1376 }
1377
1378 #[allow(clippy::too_many_arguments)]
1382 fn select_sort_keys(
1383 &mut self,
1384 ast: &Ast,
1385 query: &ast::Query,
1386 input: &Scope,
1387 output: &Scope,
1388 project: u32,
1389 exprs: &mut Vec<ExprRef>,
1390 names: &mut Vec<String>,
1391 extra: &mut Vec<usize>,
1392 above: &mut Vec<PendingSubquery>,
1393 ) -> Result<Vec<SortKey>> {
1394 if query.order_by_all {
1395 return Ok(self.every_column(output));
1396 }
1397 let items = ast.order_list(query.order_by).to_vec();
1398 let mut keys = Vec::with_capacity(items.len());
1399 for item in items {
1400 self.check_order_literal(ast, item.expr)?;
1401 let position = match self.output_position(ast, item.expr, output)? {
1402 Some(position) => position,
1403 None => {
1404 let before = self.scalar_subqueries.len();
1405 let bound = self.bind_expr(ast, item.expr, input)?;
1406 self.lift_over_aggregate(before, above, input)?;
1407 let bound = self.over_aggregate(bound, input)?;
1408 match exprs.iter().position(|&held| self.same_expr(held, bound)) {
1409 Some(position) => position,
1410 None => {
1411 exprs.push(bound);
1412 names.push(describe(ast, item.expr, self.semantics));
1413 extra.push(exprs.len() - 1);
1414 exprs.len() - 1
1415 }
1416 }
1417 }
1418 };
1419 let ty = self.plan.expr_type(exprs[position]).clone();
1420 let expr = self.column(project, position, ty);
1421 keys.push(self.sort_key(expr, item));
1422 }
1423 Ok(keys)
1424 }
1425
1426 fn sort_keys(
1428 &mut self,
1429 ast: &Ast,
1430 query: &ast::Query,
1431 output: &Scope,
1432 targets: &[ast::Target],
1433 ) -> Result<Vec<SortKey>> {
1434 if query.order_by_all {
1435 return Ok(self.every_column(output));
1436 }
1437 let items = ast.order_list(query.order_by).to_vec();
1438 let mut keys = Vec::with_capacity(items.len());
1439 for item in items {
1440 self.check_order_literal(ast, item.expr)?;
1441 let expr = match self.output_position(ast, item.expr, output)? {
1442 Some(position) => {
1443 let column = &output.columns[position];
1444 let (binding, ty) = (column.binding, column.ty.clone());
1445 self.plan.add_expr(Expr::Column(binding), ty)
1446 }
1447 None => {
1448 let _ = targets;
1449 self.bind_expr(ast, item.expr, output)?
1450 }
1451 };
1452 keys.push(self.sort_key(expr, item));
1453 }
1454 Ok(keys)
1455 }
1456
1457 fn every_column(&mut self, output: &Scope) -> Vec<SortKey> {
1458 let columns: Vec<(ColumnBinding, LogicalType)> =
1459 output.columns.iter().map(|column| (column.binding, column.ty.clone())).collect();
1460 columns
1461 .into_iter()
1462 .map(|(binding, ty)| {
1463 let expr = self.plan.add_expr(Expr::Column(binding), ty);
1464 let descending = self.semantics.default_descending();
1465 SortKey { expr, descending, nulls_first: self.semantics.nulls_first(descending) }
1466 })
1467 .collect()
1468 }
1469
1470 fn sort_key(&self, expr: ExprRef, item: ast::OrderItem) -> SortKey {
1472 let descending = match item.order {
1473 Order::Unstated => self.semantics.default_descending(),
1474 Order::Ascending => false,
1475 Order::Descending => true,
1476 };
1477 let nulls_first = match item.nulls {
1478 Nulls::First => true,
1479 Nulls::Last => false,
1480 Nulls::Unstated => self.semantics.nulls_first(descending),
1481 };
1482 SortKey { expr, descending, nulls_first }
1483 }
1484
1485 fn output_position(
1487 &self,
1488 ast: &Ast,
1489 item: ast::ExprRef,
1490 output: &Scope,
1491 ) -> Result<Option<usize>> {
1492 match ast.expr(item) {
1493 ast::Expr::Literal { kind: LiteralKind::Number, text } => {
1494 let written = ast.string(text);
1495 if written.contains(['.', 'e', 'E']) {
1496 return Ok(None);
1497 }
1498 let position: usize = written.parse().map_err(|_| {
1499 Error::binder(format!("ORDER BY term {written} is not a column"))
1500 })?;
1501 if position == 0 || position > output.len() {
1502 return Err(Error::binder(format!(
1503 "ORDER BY term out of range - should be between 1 and {}",
1504 output.len()
1505 )));
1506 }
1507 Ok(Some(position - 1))
1508 }
1509 ast::Expr::Column { name } => {
1510 let parts: Vec<&str> = ast.name(name).collect();
1511 let [written] = parts.as_slice() else { return Ok(None) };
1512 Ok(output.position_of(None, written))
1513 }
1514 _ => Ok(None),
1515 }
1516 }
1517
1518 fn check_order_literal(&self, ast: &Ast, item: ast::ExprRef) -> Result<()> {
1520 if !self.semantics.order_by_non_integer_literal()
1521 && matches!(
1522 ast.expr(item),
1523 ast::Expr::Literal { kind, text }
1524 if kind != LiteralKind::Number
1525 || ast.string(text).contains(['.', 'e', 'E'])
1526 )
1527 {
1528 return Err(Error::binder(
1529 "ORDER BY non-integer literal has no effect.\n* SET order_by_non_integer_literal=true to allow this behavior.",
1530 ));
1531 }
1532 Ok(())
1533 }
1534
1535 fn distinct_on(
1537 &mut self,
1538 ast: &Ast,
1539 distinct: Distinct,
1540 output: &Scope,
1541 ) -> Result<Vec<ExprRef>> {
1542 let Distinct::On(items) = distinct else {
1543 return Ok(Vec::new());
1544 };
1545 let items = ast.expr_list(items).to_vec();
1546 let mut on = Vec::with_capacity(items.len());
1547 for item in items {
1548 let Some(position) = self.output_position(ast, item, output)? else {
1549 return Err(Error::not_implemented(
1550 "DISTINCT ON an expression that is not in the select list",
1551 ));
1552 };
1553 let column = &output.columns[position];
1554 let (binding, ty) = (column.binding, column.ty.clone());
1555 on.push(self.plan.add_expr(Expr::Column(binding), ty));
1556 }
1557 Ok(on)
1558 }
1559
1560 fn apply_limit(
1567 &mut self,
1568 ast: &Ast,
1569 query: &ast::Query,
1570 input: NodeRef,
1571 scope: &mut Scope,
1572 ) -> Result<NodeRef> {
1573 let waiting = self.scalar_subqueries.len();
1574 if query.limit_percent {
1575 let percent = self.share(ast, query.limit)?;
1576 let offset = self.skipped(ast, query.offset)?;
1577 let node = |binder: &mut Self, input| match percent {
1578 Some(percent) => binder.add_node(Node::LimitPercent { input, percent, offset }),
1579 None => binder.limited(input, Bound::All, offset),
1582 };
1583 return self.over_subqueries(waiting, input, scope, node);
1584 }
1585 let count = self.count_bound(ast, query.limit, "LIMIT")?;
1586 let offset = self.skipped(ast, query.offset)?;
1587 let node = |binder: &mut Self, input| binder.limited(input, count, offset);
1588 self.over_subqueries(waiting, input, scope, node)
1589 }
1590
1591 fn skipped(&mut self, ast: &Ast, written: ast::ExprRef) -> Result<Bound> {
1596 Ok(match self.count_bound(ast, written, "OFFSET")? {
1597 Bound::All => Bound::Rows(0),
1598 named => named,
1599 })
1600 }
1601
1602 fn over_subqueries(
1609 &mut self,
1610 waiting: usize,
1611 input: NodeRef,
1612 scope: &mut Scope,
1613 node: impl FnOnce(&mut Self, NodeRef) -> NodeRef,
1614 ) -> Result<NodeRef> {
1615 let joined = self.scalar_subqueries.split_off(waiting);
1616 if joined.is_empty() {
1617 return Ok(node(self, input));
1618 }
1619 let mut input = input;
1620 for pending in joined {
1621 input = self.attach_subquery(input, pending);
1622 }
1623 let limit = node(self, input);
1624 Ok(self.reproject(limit, scope))
1625 }
1626
1627 fn limited(&mut self, input: NodeRef, count: Bound, offset: Bound) -> NodeRef {
1630 if count == Bound::All && offset == Bound::Rows(0) {
1631 return input;
1632 }
1633 self.add_node(Node::Limit { input, count, offset })
1634 }
1635
1636 fn reproject(&mut self, node: NodeRef, scope: &mut Scope) -> NodeRef {
1642 let index = self.fresh_index();
1643 let mut exprs = Vec::with_capacity(scope.columns.len());
1644 let mut names = Vec::with_capacity(scope.columns.len());
1645 for column in &scope.columns {
1646 exprs.push(self.plan.add_expr(Expr::Column(column.binding), column.ty.clone()));
1647 names.push(self.plan.intern(&column.name));
1648 }
1649 for (at, column) in scope.columns.iter_mut().enumerate() {
1650 column.binding = ColumnBinding::new(index, at as u32);
1651 }
1652 let exprs = self.plan.add_expr_list(&exprs);
1653 let names = self.plan.add_name_list(&names);
1654 self.add_node(Node::Project { input: node, index, exprs, names })
1655 }
1656
1657 fn share(&mut self, ast: &Ast, written: ast::ExprRef) -> Result<Option<Share>> {
1674 if written == NONE {
1675 return Ok(None);
1676 }
1677 self.clause = "LIMIT clause";
1678 let scope = Scope::empty();
1679 let bound = self.bind_expr(ast, written, &scope)?;
1680 let Some(value) = fold::value_of(&self.plan, bound)? else {
1681 return Ok(Some(Share::Read(bound)));
1682 };
1683 if value.is_null() {
1684 return Ok(None);
1685 }
1686 let percent = percentage(&value)?;
1687 if !(0.0..=100.0).contains(&percent) {
1688 return Err(Error::out_of_range(
1689 "Limit percent out of range, should be between 0% and 100%",
1690 ));
1691 }
1692 Ok(Some(Share::Percent(percent)))
1693 }
1694
1695 fn count_bound(&mut self, ast: &Ast, written: ast::ExprRef, clause: &str) -> Result<Bound> {
1714 if written == NONE {
1715 return Ok(Bound::All);
1716 }
1717 self.clause = "LIMIT clause";
1718 let scope = Scope::empty();
1719 let bound = self.bind_expr(ast, written, &scope)?;
1720 let Some(value) = fold::value_of(&self.plan, bound)? else {
1721 return Ok(Bound::Read(bound));
1722 };
1723 if value.is_null() {
1726 return Ok(Bound::All);
1727 }
1728 row_count(&value, clause).map(Bound::Rows)
1729 }
1730
1731 fn bind_from(&mut self, ast: &Ast, from: ast::Slice) -> Result<(NodeRef, Scope)> {
1734 let sources = ast.source_list(from).to_vec();
1735 let Some((first, rest)) = sources.split_first() else {
1736 return Ok((self.add_node(Node::Dummy), Scope::empty()));
1739 };
1740 let (mut node, mut scope) = self.bind_source(ast, *first)?;
1741 for source in rest {
1742 let (right, right_scope, correlations) = self.bind_lateral(ast, *source, &scope)?;
1743 node = if correlations.is_empty() {
1744 self.add_node(Node::CrossProduct { left: node, right })
1745 } else {
1746 let conditions = self.plan.add_expr_list(&[]);
1747 self.add_node(Node::DependentJoin {
1748 left: node,
1749 right,
1750 kind: JoinKind::Inner,
1751 conditions,
1752 })
1753 };
1754 scope = scope.concat(right_scope);
1755 }
1756 Ok((node, scope))
1757 }
1758
1759 fn bind_lateral(
1771 &mut self,
1772 ast: &Ast,
1773 source: ast::SourceRef,
1774 left: &Scope,
1775 ) -> Result<(NodeRef, Scope, Vec<ColumnBinding>)> {
1776 self.lateral_scopes.push(self.outer_scopes.len());
1777 self.outer_scopes.push(left.clone());
1778 self.correlations.push(Vec::new());
1779 let bound = self.bind_source(ast, source);
1780 let read = self.correlations.pop().expect("correlation frame");
1781 self.outer_scopes.pop();
1782 self.lateral_scopes.pop();
1783 let (node, scope) = bound?;
1784
1785 let mut here = Vec::new();
1786 for binding in read {
1787 if left.columns.iter().any(|column| column.binding == binding) {
1788 here.push(binding);
1789 } else if let Some(enclosing) = self.correlations.last_mut() {
1790 if !enclosing.contains(&binding) {
1791 enclosing.push(binding);
1792 }
1793 }
1794 }
1795 Ok((node, scope, here))
1805 }
1806
1807 fn bind_source(&mut self, ast: &Ast, source: ast::SourceRef) -> Result<(NodeRef, Scope)> {
1808 match ast.source(source) {
1809 ast::Source::Table { name, alias, columns } => {
1810 self.bind_table(ast, name, alias, columns)
1811 }
1812 ast::Source::Function { name, args, alias, columns, pragma } => {
1813 self.bind_table_function(ast, name, args, alias, columns, pragma)
1814 }
1815 ast::Source::Subquery { query, alias, columns } => {
1816 let (node, mut scope) = self.bind_query(ast, query)?;
1817 let label = if alias == NONE {
1818 "unnamed_subquery".to_string()
1819 } else {
1820 ast.string(alias).to_string()
1821 };
1822 scope.relabel(&label);
1823 if !columns.is_empty() {
1824 let names: Vec<&str> = ast.name(columns).collect();
1825 scope.rename(&names, &label)?;
1826 }
1827 Ok((node, scope))
1828 }
1829 ast::Source::Values { rows, alias, columns } => {
1830 let bare = ast::Query::bare(ast::QueryBody::Values(rows));
1831 let (node, mut scope) = self.bind_values(ast, &bare, rows)?;
1832 let label =
1833 if alias == NONE { String::new() } else { ast.string(alias).to_string() };
1834 scope.relabel(&label);
1835 if !columns.is_empty() {
1836 let names: Vec<&str> = ast.name(columns).collect();
1837 scope.rename(&names, &label)?;
1838 }
1839 Ok((node, scope))
1840 }
1841 ast::Source::Cte { cte, alias, columns } => {
1842 self.bind_cte_scan(ast, cte, alias, columns)
1843 }
1844 ast::Source::Join { left, right, kind, natural, on, using } => {
1845 self.bind_join(ast, left, right, kind, natural, on, using)
1846 }
1847 }
1848 }
1849
1850 fn bind_cte_scan(
1857 &mut self,
1858 ast: &Ast,
1859 written: u32,
1860 alias: ast::StrRef,
1861 columns: ast::Slice,
1862 ) -> Result<(NodeRef, Scope)> {
1863 let Some(held) = self.materialized.iter().rev().find(|held| held.written == written) else {
1864 let name = ast.string(ast.cte(written).name);
1865 return Err(Error::binder(format!("Table with name {name} does not exist!")));
1866 };
1867 let cte = held.cte;
1868 let fields = held.fields.clone();
1869 let text = held.name.clone();
1870 let label = if alias == NONE { text.clone() } else { ast.string(alias).to_string() };
1871 let name = self.plan.intern(&text);
1872 let index = self.fresh_index();
1873 let mut scope = Scope::empty();
1874 for (at, field) in fields.iter().enumerate() {
1875 scope.push(Visible {
1876 table: label.clone(),
1877 name: field.name.clone(),
1878 binding: ColumnBinding::new(index, at as u32),
1879 ty: field.ty.clone(),
1880 not_null: field.not_null,
1881 key: None,
1882 default: None,
1883 qualified: false,
1884 also: None,
1885 });
1886 }
1887 if !columns.is_empty() {
1888 let names: Vec<&str> = ast.name(columns).collect();
1889 scope.rename(&names, &label)?;
1890 }
1891 let columns = self.plan.add_fields(&fields);
1892 let node = self.add_node(Node::CteScan { index, cte, name, columns });
1893 Ok((node, scope))
1894 }
1895
1896 fn bind_table(
1897 &mut self,
1898 ast: &Ast,
1899 name: ast::Slice,
1900 alias: ast::StrRef,
1901 columns: ast::Slice,
1902 ) -> Result<(NodeRef, Scope)> {
1903 let parts: Vec<&str> = ast.name(name).collect();
1904 let catalog = self.catalog;
1905 let resolved = match catalog.resolve(&parts) {
1908 Ok(resolved) => resolved,
1909 Err(missing) => {
1910 return self.bind_replacement_scan(ast, &parts, alias, columns, missing);
1911 }
1912 };
1913 if catalog.entry(&resolved)? == Entry::View {
1914 return self.bind_view(ast, &resolved, alias, columns);
1915 }
1916 let label =
1917 if alias == NONE { resolved.table.clone() } else { ast.string(alias).to_string() };
1918 self.bind_catalog_table(ast, &resolved, label, columns)
1919 }
1920
1921 pub(crate) fn bind_catalog_table(
1924 &mut self,
1925 ast: &Ast,
1926 resolved: &QualifiedName,
1927 label: String,
1928 columns: ast::Slice,
1929 ) -> Result<(NodeRef, Scope)> {
1930 let table = self.catalog.table(resolved)?;
1931 let fields: Vec<Field> = table.columns().to_vec();
1932 let excluded = self.upsert && same_name(&label, "excluded");
1935 let mut marks = vec![None; fields.len()];
1938 for key in table.keys() {
1939 for &column in &key.columns {
1940 if key.primary || marks[column].is_none() {
1941 marks[column] = Some(if key.primary { "PRI" } else { "UNI" });
1942 }
1943 }
1944 }
1945 let index = self.fresh_index();
1946 let mut scope = Scope::empty();
1947 for (at, field) in fields.iter().enumerate() {
1948 scope.push(Visible {
1949 table: label.clone(),
1950 name: field.name.clone(),
1951 binding: ColumnBinding::new(index, at as u32),
1952 ty: field.ty.clone(),
1953 not_null: field.not_null,
1954 key: marks[at],
1955 default: table.default(at).map(str::to_owned),
1956 qualified: excluded,
1957 also: None,
1958 });
1959 }
1960 if !columns.is_empty() {
1961 let names: Vec<&str> = ast.name(columns).collect();
1962 scope.rename(&names, &label)?;
1963 }
1964 let resolved = if excluded { &QualifiedName::excluded() } else { resolved };
1965 let catalog_name = self.plan.intern(&resolved.catalog);
1966 let schema = self.plan.intern(&resolved.schema);
1967 let table_name = self.plan.intern(&resolved.table);
1968 let alias = self.plan.intern(&label);
1969 let columns = self.plan.add_fields(&fields);
1970 if let Some(zones) = table.rows().zones().filter(|_| !excluded) {
1975 self.plan.set_zones(index, zones);
1976 }
1977 if let Some(frequencies) = table.frequencies().filter(|_| !excluded) {
1978 self.plan.set_frequencies(index, frequencies);
1979 }
1980 for (column, distinct) in table.distincts() {
1981 if !excluded {
1982 self.plan.measure_distinct(index, &column, distinct);
1983 }
1984 }
1985 if self.want_ascending && !excluded {
1986 for column in table.ascending() {
1987 self.plan.mark_ascending(index, &column);
1988 }
1989 }
1990 let node = self.add_node(Node::Get {
1991 catalog: catalog_name,
1992 schema,
1993 table: table_name,
1994 alias,
1995 index,
1996 columns,
1997 });
1998 Ok((node, scope))
1999 }
2000
2001 fn bind_view(
2013 &mut self,
2014 ast: &Ast,
2015 name: &QualifiedName,
2016 alias: ast::StrRef,
2017 columns: ast::Slice,
2018 ) -> Result<(NodeRef, Scope)> {
2019 let view = self.catalog.view(name)?;
2020 let full = name.to_string();
2021 if self.expanding.contains(&full) {
2022 return Err(Error::binder(format!(
2026 "infinite recursion detected: attempting to recursively bind view \"\"{}\"\"",
2027 name.table
2028 )));
2029 }
2030 let body = parse_ast_with_case(view.sql(), self.semantics.identifier_case())?;
2031 let query = match body.statements.as_slice() {
2032 [ast::Statement::Query(query)] => *query,
2033 _ => return Err(Error::binder(format!("view \"{}\" is not a query", name.table))),
2036 };
2037 self.expanding.push(full);
2038 let bound = self.bind_query(&body, query);
2039 self.expanding.pop();
2040 let (node, mut scope) = bound?;
2041
2042 let aliases: Vec<&str> = view.aliases().iter().map(String::as_str).collect();
2043 if !aliases.is_empty() {
2044 scope.rename(&aliases, "unnamed_subquery")?;
2045 }
2046 view.remember(scope.fields());
2053 let label = if alias == NONE { name.table.clone() } else { ast.string(alias).to_string() };
2054 scope.relabel(&label);
2055 if !columns.is_empty() {
2056 let names: Vec<&str> = ast.name(columns).collect();
2057 scope.rename(&names, &label)?;
2058 }
2059 Ok((node, scope))
2060 }
2061
2062 fn bind_table_function(
2070 &mut self,
2071 ast: &Ast,
2072 name: ast::Slice,
2073 args: ast::Slice,
2074 alias: ast::StrRef,
2075 columns: ast::Slice,
2076 pragma: bool,
2077 ) -> Result<(NodeRef, Scope)> {
2078 let renamed = columns;
2081 let parts: Vec<&str> = ast.name(name).collect();
2082 let function_name = *parts.last().unwrap_or(&"");
2086 if let Some(schema) = parts.iter().rev().nth(1) {
2087 if !schema.eq_ignore_ascii_case("main") && !schema.eq_ignore_ascii_case("system") {
2088 return Err(Error::catalog(format!(
2089 "Table Function with name {} does not exist!",
2090 parts.join(".")
2091 )));
2092 }
2093 }
2094 let Some(called) = TableFunction::lookup(function_name) else {
2098 if pragma {
2099 if args.is_empty() && self.catalog.resolve(&parts).is_ok() {
2105 return self.bind_table(ast, name, alias, columns);
2106 }
2107 let spelled = function_name.strip_prefix("pragma_").unwrap_or(function_name);
2108 return Err(Error::catalog(format!(
2109 "Pragma Function with name {spelled} does not exist!"
2110 )));
2111 }
2112 return Err(Error::catalog(format!(
2113 "Table Function with name {function_name} does not exist!"
2114 )));
2115 };
2116 let written = ast.target_list(args).to_vec();
2117 let empty = Scope::empty();
2118 let previous = std::mem::replace(&mut self.clause, "table function arguments");
2119 let mut bound = Vec::new();
2120 let mut written_options = Vec::new();
2121 for argument in written {
2122 let expr = self.bind_expr(ast, argument.expr, &empty)?;
2123 if argument.alias == NONE {
2124 bound.push(expr);
2125 } else {
2126 let name = ast.string(argument.alias).to_string();
2127 let (parameter, value) = self.named_argument(called, &name, expr)?;
2128 written_options.push((parameter, value, expr));
2129 }
2130 }
2131 self.clause = previous;
2132 let options = Options::of(&written_options)?;
2133
2134 let given: Vec<LogicalType> =
2137 bound.iter().map(|&expr| self.plan.expr_type(expr).clone()).collect();
2138 let resolved = if pragma {
2139 resolve_pragma(function_name, &given)?
2140 } else {
2141 resolve_table(function_name, &given)?
2142 };
2143 let mut cast: Vec<ExprRef> = bound
2144 .iter()
2145 .zip(&resolved.arguments)
2146 .map(|(&expr, ty)| self.checked_cast_to(expr, ty, false))
2147 .collect::<Result<_>>()?;
2148
2149 if resolved.function.answered_when_bound() {
2150 let Columns::Fixed(fields) = resolved.columns else {
2151 return Err(Error::internal("a pragma that resolved to a file"));
2152 };
2153 let [argument] = cast[..] else {
2154 return Err(Error::internal("a pragma that resolved to more than one name"));
2155 };
2156 return self.bind_pragma(ast, resolved.function, &fields, argument, alias, columns);
2157 }
2158 let mut measured = Stat::Unknown;
2161 let mut counted: Vec<(String, Stat<u64>)> = Vec::new();
2162 let mut bounded: Option<Arc<dyn Zones>> = None;
2163 let fields = match resolved.columns {
2164 Columns::Fixed(fields) => fields,
2165 columns => {
2166 let paths = self.file_paths(cast[0], resolved.function.name())?;
2171 let mut mirrorable = None;
2172 if resolved.function == TableFunction::ReadParquet && !options.file_row_number {
2173 if let Some((path, stamp)) = mirror_target(&paths) {
2174 if let Some(name) =
2175 self.catalog.mirror(&path, options.binary_as_string, stamp)
2176 {
2177 let name = name.clone();
2178 let label = if alias == NONE {
2179 resolved.function.name().to_string()
2180 } else {
2181 ast.string(alias).to_string()
2182 };
2183 return self.bind_catalog_table(ast, &name, label, renamed);
2184 }
2185 mirrorable = Some(path);
2186 }
2187 }
2188 let mut fields = match columns {
2189 Columns::Csv => csv_fields(&paths, options.given)?,
2192 _ => {
2193 let footers = self.footers(&paths, mirrorable.as_deref())?;
2194 if let Some(path) = mirrorable.as_deref() {
2195 self.want_mirror(path, options.binary_as_string, &footers.rows);
2196 }
2197 measured = footers.rows;
2198 counted = footers.distincts;
2199 bounded = footers.zones;
2200 footers.fields
2201 }
2202 };
2203 if options.all_varchar {
2204 for field in &mut fields {
2209 field.ty = LogicalType::Varchar;
2210 }
2211 }
2212 if options.binary_as_string {
2213 for field in &mut fields {
2218 if field.ty == LogicalType::Blob {
2219 field.ty = LogicalType::Varchar;
2220 }
2221 }
2222 }
2223 if options.file_row_number {
2224 if fields.iter().any(|field| field.name == FILE_ROW_NUMBER) {
2230 return Err(Error::binder(format!(
2231 "Duplicate column name \"{FILE_ROW_NUMBER}\": the file already has a \
2232 column of that name, so file_row_number cannot add one"
2233 )));
2234 }
2235 fields.push(Field::required(FILE_ROW_NUMBER.to_string(), LogicalType::BigInt));
2236 }
2237 cast = paths.iter().map(|path| self.path_constant(path)).collect();
2238 fields
2239 }
2240 };
2241 let label = if alias == NONE {
2242 resolved.function.name().to_string()
2243 } else {
2244 ast.string(alias).to_string()
2245 };
2246 let names: Vec<&str> = ast.name(columns).collect();
2247 self.table_function_source(
2248 resolved.function,
2249 &cast,
2250 &written_options,
2251 Read { fields, rows: measured, distincts: counted, zones: bounded },
2252 &label,
2253 &names,
2254 )
2255 }
2256
2257 fn bind_pragma(
2270 &mut self,
2271 ast: &Ast,
2272 function: TableFunction,
2273 fields: &[Field],
2274 argument: ExprRef,
2275 alias: ast::StrRef,
2276 columns: ast::Slice,
2277 ) -> Result<(NodeRef, Scope)> {
2278 let written = self.pragma_name(argument, function)?;
2279 let parts = identifier_parts(&written);
2280 let spelled: Vec<&str> = parts.iter().map(String::as_str).collect();
2281 let name = self.catalog.resolve(&spelled)?;
2282 let described = self.described(ast, &name)?;
2283 let mut rows = Vec::with_capacity(described.len());
2284 for (at, field) in described.iter().enumerate() {
2285 let items = if matches!(function, TableFunction::PragmaShow) {
2286 self.describing(field)
2287 } else {
2288 self.table_info(at, field)
2289 };
2290 rows.push(self.plan.add_expr_list(&items));
2291 }
2292 let rows = self.plan.add_rows(&rows);
2293 let held = self.plan.add_fields(fields);
2294 let index = self.fresh_index();
2295 let node = self.add_node(Node::Values { index, columns: held, rows });
2296 let label =
2297 if alias == NONE { function.name().to_string() } else { ast.string(alias).to_string() };
2298 let mut scope = Scope::empty();
2299 for (at, field) in fields.iter().enumerate() {
2300 scope.push(Visible {
2301 table: label.clone(),
2302 name: field.name.clone(),
2303 binding: ColumnBinding::new(index, at as u32),
2304 ty: field.ty.clone(),
2305 not_null: false,
2306 key: None,
2307 default: None,
2308 qualified: false,
2309 also: None,
2310 });
2311 }
2312 if !columns.is_empty() {
2313 let names: Vec<&str> = ast.name(columns).collect();
2314 scope.rename(&names, &label)?;
2315 }
2316 Ok((node, scope))
2317 }
2318
2319 fn pragma_name(&self, argument: ExprRef, function: TableFunction) -> Result<String> {
2329 let Expr::Constant(reference) = *self.plan.expr(argument) else {
2330 return Err(Error::not_implemented(format!(
2331 "{}() given a name that is not a constant",
2332 function.name()
2333 )));
2334 };
2335 match self.plan.value(reference) {
2336 Value::Varchar(name) => Ok(name.clone()),
2337 Value::Null => Ok("NULL".to_string()),
2338 other => {
2339 Err(Error::internal(format!("a pragma name bound as VARCHAR arrived as {other}")))
2340 }
2341 }
2342 }
2343
2344 fn described(&mut self, ast: &Ast, name: &QualifiedName) -> Result<Vec<Field>> {
2355 if self.catalog.entry(name)? == Entry::Table {
2356 return Ok(self.catalog.table(name)?.columns().to_vec());
2357 }
2358 let (_, scope) = self.bind_view(ast, name, NONE, ast::Slice::default())?;
2359 Ok(scope.fields())
2360 }
2361
2362 fn describing(&mut self, field: &Field) -> Vec<ExprRef> {
2364 let written = [
2365 field.name.clone(),
2366 field.ty.to_string(),
2367 if field.not_null { "NO" } else { "YES" }.to_owned(),
2368 ];
2369 let mut items: Vec<ExprRef> =
2370 written.into_iter().map(|text| self.plan.add_constant(Value::Varchar(text))).collect();
2371 for _ in 0..3 {
2372 let empty = self.plan.add_constant(Value::Null);
2373 items.push(self.cast_to(empty, &LogicalType::Varchar));
2374 }
2375 items
2376 }
2377
2378 fn table_info(&mut self, at: usize, field: &Field) -> Vec<ExprRef> {
2384 let cid = self.plan.add_constant(Value::Integer(i32::try_from(at).unwrap_or(i32::MAX)));
2385 let name = self.plan.add_constant(Value::Varchar(field.name.clone()));
2386 let ty = self.plan.add_constant(Value::Varchar(field.ty.to_string()));
2387 let not_null = self.plan.add_constant(Value::Boolean(field.not_null));
2388 let default = self.plan.add_constant(Value::Null);
2389 let default = self.cast_to(default, &LogicalType::Varchar);
2390 let key = self.plan.add_constant(Value::Boolean(false));
2391 vec![cid, name, ty, not_null, default, key]
2392 }
2393
2394 fn named_argument(
2408 &mut self,
2409 function: TableFunction,
2410 name: &str,
2411 expr: ExprRef,
2412 ) -> Result<(&'static str, Value)> {
2413 let known = function
2414 .parameters()
2415 .iter()
2416 .find(|(parameter, _)| parameter.eq_ignore_ascii_case(name));
2417 let Some((parameter, wanted)) = known else {
2418 let candidates: Vec<String> = function
2419 .parameters()
2420 .iter()
2421 .map(|(parameter, ty)| format!(" {parameter} {ty}"))
2422 .collect();
2423 return Err(Error::binder(format!(
2424 "Invalid named parameter \"{name}\" for function {}\nCandidates:\n{}\n",
2425 function.name(),
2426 candidates.join("\n")
2427 )));
2428 };
2429 let Expr::Constant(reference) = *self.plan.expr(expr) else {
2430 return Err(Error::not_implemented(format!(
2431 "the named parameter {parameter} with a value that is not a constant"
2432 )));
2433 };
2434 let value = self.plan.value(reference).clone();
2435 if value == Value::Null {
2436 return Err(Error::binder(null_parameter(function, parameter)));
2437 }
2438 let given = self.plan.expr_type(expr).clone();
2439 if given != *wanted {
2440 return Err(Error::not_implemented(format!(
2441 "the named parameter {parameter} given a {given} where a {wanted} was wanted"
2442 )));
2443 }
2444 Ok((parameter, value))
2445 }
2446
2447 fn bind_replacement_scan(
2458 &mut self,
2459 ast: &Ast,
2460 parts: &[&str],
2461 alias: ast::StrRef,
2462 columns: ast::Slice,
2463 missing: Error,
2464 ) -> Result<(NodeRef, Scope)> {
2465 let [path] = parts else { return Err(missing) };
2466 let path = *path;
2467 let extension = path.rsplit_once('.').map(|(_, after)| after).unwrap_or_default();
2468 let Some(function) = Self::reader_for(extension) else {
2469 if is_file(path) {
2470 return Err(Error::binder(format!(
2475 "No extension found that is capable of reading the file \"{path}\"\n* If this \
2476 file is a supported file format you can explicitly use the reader functions, \
2477 such as read_csv, read_json or read_parquet"
2478 )));
2479 }
2480 return Err(missing);
2481 };
2482 let paths = files(path)?;
2487 let label = if alias == NONE {
2493 if is_pattern(path) {
2494 path.to_string()
2495 } else {
2496 let file = path.rsplit_once('/').map_or(path, |(_, file)| file);
2497 file.rsplit_once('.').map_or(file, |(stem, _)| stem).to_string()
2498 }
2499 } else {
2500 ast.string(alias).to_string()
2501 };
2502 let mut mirrorable = None;
2503 if function == TableFunction::ReadParquet {
2504 if let Some((canonical, stamp)) = mirror_target(&paths) {
2505 if let Some(name) = self.catalog.mirror(&canonical, false, stamp) {
2506 let name = name.clone();
2507 return self.bind_catalog_table(ast, &name, label, columns);
2508 }
2509 mirrorable = Some(canonical);
2510 }
2511 }
2512 let read = match function {
2513 TableFunction::ReadParquet => {
2514 let footers = self.footers(&paths, mirrorable.as_deref())?;
2515 if let Some(canonical) = mirrorable.as_deref() {
2516 self.want_mirror(canonical, false, &footers.rows);
2517 }
2518 Read {
2519 fields: footers.fields,
2520 rows: footers.rows,
2521 distincts: footers.distincts,
2522 zones: footers.zones,
2523 }
2524 }
2525 _ => Read::uncounted(csv_fields(&paths, Given::default())?),
2526 };
2527 let arguments: Vec<ExprRef> = paths.iter().map(|path| self.path_constant(path)).collect();
2528 let names: Vec<&str> = ast.name(columns).collect();
2529 self.table_function_source(function, &arguments, &[], read, &label, &names)
2530 }
2531
2532 fn footers(&self, paths: &[String], mirrorable: Option<&str>) -> Result<Footers> {
2538 if let Some(path) = mirrorable.filter(|_| self.outlined) {
2539 let outline = parquet_outline(path)?;
2540 if outline.rows.value().is_some() {
2541 return Ok(outline);
2542 }
2543 }
2544 parquet_footers(paths)
2545 }
2546
2547 fn want_mirror(&mut self, path: &str, binary_as_string: bool, rows: &Stat<u64>) {
2551 if let Some(&rows) = rows.value() {
2552 self.plan.want_mirror(path, binary_as_string, rows);
2553 }
2554 }
2555
2556 fn path_constant(&mut self, path: &str) -> ExprRef {
2558 let value = self.plan.add_value(Value::Varchar(path.to_string()));
2559 self.plan.add_expr(Expr::Constant(value), LogicalType::Varchar)
2560 }
2561
2562 fn reader_for(extension: &str) -> Option<TableFunction> {
2569 if extension.eq_ignore_ascii_case("parquet") {
2570 return Some(TableFunction::ReadParquet);
2571 }
2572 if extension.eq_ignore_ascii_case("csv") || extension.eq_ignore_ascii_case("tsv") {
2573 return Some(TableFunction::ReadCsv);
2574 }
2575 None
2576 }
2577
2578 fn table_function_source(
2588 &mut self,
2589 function: TableFunction,
2590 args: &[ExprRef],
2591 written: &[(&'static str, Value, ExprRef)],
2592 read: Read,
2593 label: &str,
2594 names: &[&str],
2595 ) -> Result<(NodeRef, Scope)> {
2596 let Read { fields, rows, distincts, zones } = read;
2597 let index = self.fresh_index();
2598 if rows.is_known() {
2603 self.plan.measure(index, rows);
2604 }
2605 for (column, distinct) in distincts {
2606 self.plan.measure_distinct(index, &column, distinct);
2607 }
2608 if let Some(zones) = zones {
2609 self.plan.set_zones(index, zones);
2610 }
2611 let mut scope = Scope::empty();
2612 for (at, field) in fields.iter().enumerate() {
2613 scope.push(Visible {
2614 table: label.to_string(),
2615 name: field.name.clone(),
2616 binding: ColumnBinding::new(index, at as u32),
2617 ty: field.ty.clone(),
2618 not_null: false,
2621 key: None,
2622 default: None,
2623 qualified: false,
2624 also: None,
2625 });
2626 }
2627 if !names.is_empty() {
2628 scope.rename(names, label)?;
2629 } else if matches!(function, TableFunction::Range | TableFunction::GenerateSeries) {
2630 for column in &mut scope.columns {
2633 column.also = Some(std::mem::replace(&mut column.name, label.to_string()));
2634 }
2635 }
2636 let function = self.plan.intern(function.name());
2637 let args = self.plan.add_expr_list(args);
2638 let named: Vec<u32> =
2639 written.iter().map(|(parameter, _, _)| self.plan.intern(parameter)).collect();
2640 let settings: Vec<ExprRef> = written.iter().map(|(_, _, expr)| *expr).collect();
2641 let options = self.plan.add_name_list(&named);
2642 let settings = self.plan.add_expr_list(&settings);
2643 let columns = self.plan.add_fields(&fields);
2644 let node = self.add_node(Node::TableFunction {
2645 index,
2646 function,
2647 args,
2648 options,
2649 settings,
2650 columns,
2651 });
2652 Ok((node, scope))
2653 }
2654
2655 fn file_paths(&self, expr: ExprRef, name: &str) -> Result<Vec<String>> {
2662 let mut paths = Vec::new();
2663 for pattern in self.file_patterns(expr, name)? {
2664 paths.extend(files(&pattern)?);
2665 }
2666 Ok(paths)
2667 }
2668
2669 fn file_patterns(&self, expr: ExprRef, name: &str) -> Result<Vec<String>> {
2688 let Some(value) = fold::value_of(&self.plan, expr)? else {
2689 return Err(Error::not_implemented(
2690 "a table function file name that is not a constant",
2691 ));
2692 };
2693 match value {
2694 Value::Varchar(path) => Ok(vec![path]),
2695 Value::Null => Err(Error::parser(format!("{name} cannot take NULL list as parameter"))),
2697 Value::List { values, .. } if values.is_empty() => {
2702 Err(Error::io(format!("\"{name}\" needs at least one file to read")))
2703 }
2704 Value::List { values, .. } => values
2705 .iter()
2706 .map(|value| match value {
2707 Value::Varchar(path) => Ok(path.clone()),
2708 _ => Err(Error::parser(format!(
2709 "{name} reader cannot take NULL input as parameter"
2710 ))),
2711 })
2712 .collect(),
2713 other => {
2714 Err(Error::internal(format!("a file name bound as VARCHAR arrived as {other}")))
2715 }
2716 }
2717 }
2718
2719 fn side_of(
2737 &self,
2738 pending: &PendingSubquery,
2739 left_tables: &[u32],
2740 right_tables: &[u32],
2741 ) -> Option<Side> {
2742 let mut needs_left = false;
2743 let mut needs_right = false;
2744 let mut note = |binding: ColumnBinding| {
2745 needs_left |= left_tables.contains(&binding.table);
2746 needs_right |= right_tables.contains(&binding.table);
2747 };
2748 for &binding in &pending.reads {
2749 note(binding);
2750 }
2751 for &condition in &pending.conditions {
2756 self.plan.read_columns(condition, &mut |_, binding| note(binding));
2757 }
2758 match (needs_left, needs_right) {
2759 (true, true) => None,
2760 (_, true) => Some(Side::Right),
2761 _ => Some(Side::Left),
2762 }
2763 }
2764
2765 #[allow(clippy::too_many_arguments)]
2786 fn bind_pair_dependent_join(
2787 &mut self,
2788 kind: ast::JoinKind,
2789 independent: bool,
2790 left: NodeRef,
2791 right: NodeRef,
2792 pair: Vec<PendingSubquery>,
2793 conditions: Vec<ExprRef>,
2794 scope: Scope,
2795 ) -> Result<(NodeRef, Scope)> {
2796 if kind != ast::JoinKind::Inner {
2797 return Err(Error::not_implemented(
2798 "a subquery that reads both sides of that join, written in the condition of a join \
2799 that is not an inner join"
2800 .to_string(),
2801 ));
2802 }
2803 if !independent {
2806 return Err(Error::not_implemented(
2807 "a subquery that reads both sides of that join, written in the condition of a join \
2808 whose right side is lateral"
2809 .to_string(),
2810 ));
2811 }
2812 let mut node = self.add_node(Node::CrossProduct { left, right });
2813 for pending in pair {
2814 node = self.attach_subquery(node, pending);
2815 }
2816 let mut conditions = conditions.into_iter();
2820 let mut predicate = conditions.next().expect("a join condition was bound");
2821 for next in conditions {
2822 let children = self.plan.add_expr_list(&[predicate, next]);
2823 let conjunction = Expr::Conjunction { op: ConjunctionOp::And, children };
2824 predicate = self.plan.add_expr(conjunction, LogicalType::Boolean);
2825 }
2826 let node = self.add_node(Node::Filter { input: node, predicate });
2827 Ok((node, scope))
2828 }
2829
2830 #[allow(clippy::too_many_arguments)]
2831 fn bind_join(
2832 &mut self,
2833 ast: &Ast,
2834 left: ast::SourceRef,
2835 right: ast::SourceRef,
2836 kind: ast::JoinKind,
2837 natural: bool,
2838 on: ast::ExprRef,
2839 using: ast::Slice,
2840 ) -> Result<(NodeRef, Scope)> {
2841 let (left_node, left_scope) = self.bind_source(ast, left)?;
2842 let (right_node, right_scope, correlated) = self.bind_lateral(ast, right, &left_scope)?;
2843 if !correlated.is_empty()
2847 && !matches!(kind, ast::JoinKind::Inner | ast::JoinKind::Cross | ast::JoinKind::Left)
2848 {
2849 return Err(Error::binder(
2850 "The combining JOIN type must be INNER or LEFT for a LATERAL reference",
2851 ));
2852 }
2853 let split = left_scope.len();
2854 let left_tables: Vec<u32> =
2860 left_scope.columns.iter().map(|column| column.binding.table).collect();
2861 let right_tables: Vec<u32> =
2862 right_scope.columns.iter().map(|column| column.binding.table).collect();
2863 let mut scope = left_scope.concat(right_scope);
2864
2865 let merged: Vec<String> = if natural {
2868 let mut names = Vec::new();
2869 for (at, column) in scope.columns.iter().enumerate().take(split) {
2870 if scope.columns[split..].iter().any(|right| same_name(&right.name, &column.name))
2871 && !names.iter().any(|held: &String| same_name(held, &column.name))
2872 {
2873 let _ = at;
2874 names.push(column.name.clone());
2875 }
2876 }
2877 names
2878 } else {
2879 let mut names: Vec<String> = Vec::new();
2885 for name in ast.name(using) {
2886 if !names.iter().any(|held| same_name(held, name)) {
2887 names.push(name.to_string());
2888 }
2889 }
2890 names
2891 };
2892
2893 let mut conditions = Vec::new();
2894 let mut dropped = Vec::new();
2895 for name in &merged {
2896 let left_at = scope.columns[..split]
2897 .iter()
2898 .position(|column| same_name(&column.name, name))
2899 .ok_or_else(|| {
2900 Error::binder(format!(
2901 "column \"{name}\" specified in USING clause does not exist in left table"
2902 ))
2903 })?;
2904 let right_at = scope.columns[split..]
2905 .iter()
2906 .position(|column| same_name(&column.name, name))
2907 .map(|at| at + split)
2908 .ok_or_else(|| {
2909 Error::binder(format!(
2910 "column \"{name}\" specified in USING clause does not exist in right table"
2911 ))
2912 })?;
2913 let left_column = &scope.columns[left_at];
2914 let (left_binding, left_type) = (left_column.binding, left_column.ty.clone());
2915 let right_column = &scope.columns[right_at];
2916 let (right_binding, right_type) = (right_column.binding, right_column.ty.clone());
2917 let left_expr = self.plan.add_expr(Expr::Column(left_binding), left_type);
2918 let right_expr = self.plan.add_expr(Expr::Column(right_binding), right_type);
2919 conditions.push(self.compare(rudb_plan::CompareOp::Equal, left_expr, right_expr)?);
2920 dropped.push(right_at);
2921 }
2922 dropped.sort_unstable();
2925 for at in dropped.into_iter().rev() {
2926 scope.remove(at);
2927 }
2928
2929 let mut left_node = left_node;
2930 let mut right_node = right_node;
2931 let mut pair = Vec::new();
2932 if on != NONE {
2933 if !merged.is_empty() {
2934 return Err(Error::binder("a join cannot have both ON and USING"));
2935 }
2936 self.clause = "JOIN condition";
2937 let waiting = self.scalar_subqueries.len();
2938 let predicate = self.bind_expr(ast, on, &scope)?;
2939 conditions.push(self.as_boolean(predicate, "JOIN")?);
2940 for pending in self.scalar_subqueries.split_off(waiting) {
2941 match self.side_of(&pending, &left_tables, &right_tables) {
2942 Some(Side::Right) => right_node = self.attach_subquery(right_node, pending),
2943 Some(Side::Left) => left_node = self.attach_subquery(left_node, pending),
2944 None => pair.push(pending),
2945 }
2946 }
2947 }
2948
2949 if kind == ast::JoinKind::Cross && !conditions.is_empty() {
2950 return Err(Error::binder("a CROSS JOIN cannot have a condition"));
2951 }
2952 if !pair.is_empty() {
2953 return self.bind_pair_dependent_join(
2954 kind,
2955 correlated.is_empty(),
2956 left_node,
2957 right_node,
2958 pair,
2959 conditions,
2960 scope,
2961 );
2962 }
2963 if correlated.is_empty()
2967 && conditions.is_empty()
2968 && matches!(kind, ast::JoinKind::Cross | ast::JoinKind::Inner)
2969 {
2970 let node = self.add_node(Node::CrossProduct { left: left_node, right: right_node });
2971 return Ok((node, scope));
2972 }
2973 if matches!(kind, ast::JoinKind::Semi | ast::JoinKind::Anti) {
2982 scope.truncate(split);
2983 }
2984 let kind = match kind {
2985 ast::JoinKind::Inner | ast::JoinKind::Cross => JoinKind::Inner,
2986 ast::JoinKind::Left => JoinKind::Left,
2987 ast::JoinKind::Right => JoinKind::Right,
2988 ast::JoinKind::Full => JoinKind::Full,
2989 ast::JoinKind::Semi => JoinKind::Semi,
2990 ast::JoinKind::Anti => JoinKind::Anti,
2991 ast::JoinKind::Positional => JoinKind::Positional,
2992 };
2993 let conditions = self.plan.add_expr_list(&conditions);
2994 let node = if correlated.is_empty() {
2995 self.add_node(Node::Join {
2996 left: left_node,
2997 right: right_node,
2998 kind,
2999 conditions,
3000 build: BuildSide::default(),
3001 })
3002 } else {
3003 self.add_node(Node::DependentJoin {
3004 left: left_node,
3005 right: right_node,
3006 kind,
3007 conditions,
3008 })
3009 };
3010 Ok((node, scope))
3011 }
3012
3013 fn bind_filter(
3021 &mut self,
3022 ast: &Ast,
3023 filter: ast::ExprRef,
3024 scope: &Scope,
3025 ) -> Result<Option<ExprRef>> {
3026 if filter == NONE {
3027 return Ok(None);
3028 }
3029 let bound = self.bind_expr(ast, filter, scope)?;
3030 Ok(Some(self.checked_cast_to(bound, &LogicalType::Boolean, false)?))
3031 }
3032
3033 pub(crate) fn bind_aggregate(
3038 &mut self,
3039 ast: &Ast,
3040 name: &str,
3041 args: &[ast::ExprRef],
3042 distinct: bool,
3043 filter: ast::ExprRef,
3044 scope: &Scope,
3045 ) -> Result<ExprRef> {
3046 let frames = std::mem::take(&mut self.lambda_frames);
3047 let bound = self.bind_aggregate_over_rows(ast, name, args, distinct, filter, scope);
3048 self.lambda_frames = frames;
3049 bound
3050 }
3051
3052 fn bind_aggregate_over_rows(
3053 &mut self,
3054 ast: &Ast,
3055 name: &str,
3056 args: &[ast::ExprRef],
3057 distinct: bool,
3058 filter: ast::ExprRef,
3059 scope: &Scope,
3060 ) -> Result<ExprRef> {
3061 if self.in_filter {
3062 return Err(Error::binder("aggregate functions are not allowed in FILTER"));
3063 }
3064 if self.in_aggregate {
3065 return Err(Error::binder(format!(
3066 "aggregate function calls cannot be nested, and {name}() is inside one"
3067 )));
3068 }
3069 if self.aggregation.is_none() {
3070 return Err(Error::binder(format!(
3071 "aggregate function calls cannot be used in the {}",
3072 self.clause
3073 )));
3074 }
3075 self.in_aggregate = true;
3080 self.in_filter = true;
3081 let filter = self.bind_filter(ast, filter, scope);
3082 self.in_filter = false;
3083 self.in_aggregate = false;
3084 let filter = filter?;
3085
3086 self.in_aggregate = true;
3087 let mut bound = Vec::with_capacity(args.len());
3088 let mut failure = None;
3089 for &arg in args {
3090 match self.bind_expr(ast, arg, scope) {
3091 Ok(expr) => bound.push(expr),
3092 Err(error) => {
3093 failure = Some(error);
3094 break;
3095 }
3096 }
3097 }
3098 self.in_aggregate = false;
3099 if let Some(error) = failure {
3100 return Err(error);
3101 }
3102
3103 let types: Vec<LogicalType> =
3104 bound.iter().map(|&arg| self.plan.expr_type(arg).clone()).collect();
3105 let resolved = resolve(name, &types)?;
3106 if resolved.name == "string_agg"
3109 && bound.len() == 2
3110 && !matches!(fold::value_of(&self.plan, bound[1]), Ok(Some(_)))
3111 {
3112 return Err(Error::binder(
3113 "The \"separator\" argument in function \"string_agg\" must be a constant expression",
3114 ));
3115 }
3116 let mut cast = Vec::with_capacity(bound.len());
3117 for (arg, wanted) in bound.iter().zip(&resolved.arguments) {
3118 cast.push(self.checked_cast_to(*arg, wanted, false)?);
3119 }
3120 let args = self.plan.add_expr_list(&cast);
3121 let name = self.plan.intern(resolved.name);
3122 let ty = resolved.returns;
3123 let call = self.plan.add_expr(Expr::Aggregate { name, args, distinct, filter }, ty.clone());
3124
3125 let existing = self.aggregation.as_ref().map(|held| held.aggregates.clone());
3128 let existing = existing.unwrap_or_default();
3129 let at = match existing.iter().position(|&held| self.same_expr(held, call)) {
3130 Some(at) => at,
3131 None => {
3132 let aggregation = self.aggregation.as_mut().expect("checked above");
3133 aggregation.aggregates.push(call);
3134 aggregation.aggregates.len() - 1
3135 }
3136 };
3137 let aggregation = self.aggregation.as_ref().expect("checked above");
3138 let (index, groups) = (aggregation.index, aggregation.groups.len());
3139 Ok(self.column(index, groups + at, ty))
3140 }
3141
3142 pub(crate) fn bind_window(
3153 &mut self,
3154 ast: &Ast,
3155 written: &WindowCall<'_>,
3156 scope: &Scope,
3157 ) -> Result<ExprRef> {
3158 let frames = std::mem::take(&mut self.lambda_frames);
3159 let bound = self.bind_window_over_rows(ast, written, scope);
3160 self.lambda_frames = frames;
3161 bound
3162 }
3163
3164 fn bind_window_over_rows(
3165 &mut self,
3166 ast: &Ast,
3167 written: &WindowCall<'_>,
3168 scope: &Scope,
3169 ) -> Result<ExprRef> {
3170 let WindowCall { name, args, distinct, filter, ignore_nulls, spec, .. } = *written;
3171 if self.in_aggregate {
3172 return Err(Error::binder(
3173 "aggregate function calls cannot contain window function calls",
3174 ));
3175 }
3176 if self.in_window {
3177 return Err(Error::binder("window function calls cannot be nested"));
3178 }
3179 let clause = if self.clause == "JOIN condition" { "WHERE clause" } else { self.clause };
3183 if clause != "SELECT clause" && clause != "ORDER BY clause" {
3184 return Err(Error::binder(format!("{clause} cannot contain window functions!")));
3185 }
3186
3187 let starred = args.iter().any(|&arg| {
3191 matches!(ast.expr(arg), ast::Expr::Star { qualifier, replacements }
3192 if qualifier.is_empty() && replacements.is_empty())
3193 });
3194 let (name, args): (&str, &[ast::ExprRef]) = if starred {
3195 if !same_name(name, "count") || args.len() != 1 {
3196 return Err(Error::binder(format!("* is not allowed in {name}()")));
3197 }
3198 ("count_star", &[])
3199 } else if same_name(name, "count") && args.is_empty() {
3200 ("count_star", &[])
3203 } else {
3204 (name, args)
3205 };
3206
3207 let held = ast.window(spec);
3208 self.in_window = true;
3209 let parts = self.window_parts(ast, written, args, held, scope);
3210 let filter = if parts.is_ok() { self.bind_filter(ast, filter, scope) } else { Ok(None) };
3215 self.in_window = false;
3216 let parts = parts?;
3217 let filter = filter?;
3218 let offsets = [parts.frame.start, parts.frame.end]
3221 .iter()
3222 .any(|end| matches!(end, WindowBound::Preceding(_) | WindowBound::Following(_)));
3223 if parts.frame.unit == WindowUnit::Range && offsets && parts.order.len() != 1 {
3224 return Err(Error::binder("RANGE frames must have only one ORDER BY expression"));
3225 }
3226
3227 let types: Vec<LogicalType> =
3228 parts.args.iter().map(|&arg| self.plan.expr_type(arg).clone()).collect();
3229 let resolved = window_signature(name, &types)?;
3230 if resolved.name == "fill" {
3233 let keys: Vec<LogicalType> =
3234 parts.order.iter().map(|key| self.plan.expr_type(key.expr).clone()).collect();
3235 refuse_fill(&types[0], &keys, distinct, ignore_nulls)?;
3236 }
3237 if distinct && kind_of(resolved.name) == Some(FunctionKind::Window) {
3241 return Err(Error::binder(format!(
3242 "DISTINCT is not implemented for the window function \"\"{name}\"\""
3243 )));
3244 }
3245 if filter.is_some() && kind_of(resolved.name) == Some(FunctionKind::Window) {
3248 return Err(Error::binder(format!(
3249 "FILTER is not implemented for the window function \"\"{name}\"\""
3250 )));
3251 }
3252 if !parts.inner.is_empty() && kind_of(resolved.name) == Some(FunctionKind::Window) {
3260 let counts = matches!(resolved.name, "first_value" | "last_value" | "nth_value");
3261 if !counts {
3262 if parts.frame.exclude != WindowExclude::NoOthers {
3263 return Err(Error::binder(format!(
3264 "EXCLUDE is not supported for the window function \"\"{}\"\"",
3265 resolved.name
3266 )));
3267 }
3268 return Err(Error::not_implemented(format!(
3269 "ORDER BY inside the arguments of the window function \"{}\"",
3270 resolved.name
3271 )));
3272 }
3273 }
3274 let mut cast = Vec::with_capacity(parts.args.len());
3275 for (arg, wanted) in parts.args.iter().zip(&resolved.arguments) {
3276 cast.push(self.checked_cast_to(*arg, wanted, false)?);
3277 }
3278 let args = self.plan.add_expr_list(&cast);
3279 let order = self.plan.add_sort_keys(&parts.inner);
3280 let name = self.plan.intern(resolved.name);
3281 let ty = resolved.returns;
3282 let call = self.plan.add_expr(
3283 Expr::Window { name, args, distinct, filter, ignore_nulls, order },
3284 ty.clone(),
3285 );
3286
3287 let at = self.window_run(parts.partition, parts.order, parts.frame, call);
3288 let index = self.windows.last().expect("the run was just filed").index;
3289 Ok(self.column(index, at, ty))
3290 }
3291
3292 fn window_run(
3299 &mut self,
3300 partition: Vec<ExprRef>,
3301 order: Vec<SortKey>,
3302 frame: WindowFrame,
3303 call: ExprRef,
3304 ) -> usize {
3305 let matches = self.windows.last().is_some_and(|run| {
3306 run.frame == frame
3307 && run.partition.len() == partition.len()
3308 && run.order.len() == order.len()
3309 && run.partition.iter().zip(&partition).all(|(&l, &r)| self.same_expr(l, r))
3310 && run.order.iter().zip(&order).all(|(l, r)| {
3311 l.descending == r.descending
3312 && l.nulls_first == r.nulls_first
3313 && self.same_expr(l.expr, r.expr)
3314 })
3315 });
3316 if !matches {
3317 let index = self.fresh_index();
3318 self.windows.push(WindowRun { index, partition, order, frame, calls: Vec::new() });
3319 }
3320 let calls = self.windows.last().expect("a run is open").calls.clone();
3323 if let Some(at) = calls.iter().position(|&held| self.same_expr(held, call)) {
3324 return at;
3325 }
3326 let run = self.windows.last_mut().expect("a run is open");
3327 run.calls.push(call);
3328 run.calls.len() - 1
3329 }
3330
3331 fn window_parts(
3337 &mut self,
3338 ast: &Ast,
3339 written: &WindowCall<'_>,
3340 args: &[ast::ExprRef],
3341 held: ast::WindowSpec,
3342 scope: &Scope,
3343 ) -> Result<WindowParts> {
3344 let mut bound = Vec::with_capacity(args.len());
3345 for &arg in args {
3346 let expr = self.bind_expr(ast, arg, scope)?;
3347 bound.push(self.over_aggregate(expr, scope)?);
3348 }
3349 let mut inner = Vec::new();
3353 for item in ast.order_list(written.order).to_vec() {
3354 let expr = self.bind_expr(ast, item.expr, scope)?;
3355 let expr = self.over_aggregate(expr, scope)?;
3356 inner.push(self.sort_key(expr, item));
3357 }
3358 let mut partition = Vec::new();
3359 for &key in ast.expr_list(held.partition) {
3360 let expr = self.bind_expr(ast, key, scope)?;
3361 partition.push(self.over_aggregate(expr, scope)?);
3362 }
3363 let mut order = Vec::new();
3364 for item in ast.order_list(held.order).to_vec() {
3365 let expr = self.bind_expr(ast, item.expr, scope)?;
3366 let expr = self.over_aggregate(expr, scope)?;
3367 order.push(self.sort_key(expr, item));
3368 }
3369 let frame = WindowFrame {
3370 unit: match held.unit {
3371 ast::WindowUnit::Rows => WindowUnit::Rows,
3372 ast::WindowUnit::Range => WindowUnit::Range,
3373 ast::WindowUnit::Groups => WindowUnit::Groups,
3374 },
3375 start: self.window_bound(ast, held.start, scope)?,
3376 end: self.window_bound(ast, held.end, scope)?,
3377 exclude: match held.exclude {
3378 ast::WindowExclude::NoOthers => WindowExclude::NoOthers,
3379 ast::WindowExclude::CurrentRow => WindowExclude::CurrentRow,
3380 ast::WindowExclude::Group => WindowExclude::Group,
3381 ast::WindowExclude::Ties => WindowExclude::Ties,
3382 },
3383 };
3384 Ok(WindowParts { args: bound, partition, order, inner, frame })
3385 }
3386
3387 fn window_bound(
3389 &mut self,
3390 ast: &Ast,
3391 bound: ast::WindowBound,
3392 scope: &Scope,
3393 ) -> Result<WindowBound> {
3394 let offset = |binder: &mut Self, written| {
3395 let expr = binder.bind_expr(ast, written, scope)?;
3396 binder.over_aggregate(expr, scope)
3397 };
3398 Ok(match bound {
3399 ast::WindowBound::UnboundedPreceding => WindowBound::UnboundedPreceding,
3400 ast::WindowBound::CurrentRow => WindowBound::CurrentRow,
3401 ast::WindowBound::UnboundedFollowing => WindowBound::UnboundedFollowing,
3402 ast::WindowBound::Preceding(written) => WindowBound::Preceding(offset(self, written)?),
3403 ast::WindowBound::Following(written) => WindowBound::Following(offset(self, written)?),
3404 })
3405 }
3406
3407 fn group_of(&self, read: ColumnBinding) -> Option<usize> {
3413 self.aggregation.as_ref()?.groups.iter().position(
3414 |group| matches!(*self.plan.expr(*group), Expr::Column(binding) if binding == read),
3415 )
3416 }
3417
3418 fn ungrouped_correlation(&self, binding: ColumnBinding) -> Option<ColumnBinding> {
3424 let pending =
3425 self.scalar_subqueries.iter().find(|pending| pending.index == binding.table)?;
3426 pending.reads.iter().copied().find(|read| self.group_of(*read).is_none())
3427 }
3428
3429 fn is_window_output(&self, binding: ColumnBinding) -> bool {
3431 self.windows.iter().any(|run| run.index == binding.table)
3432 }
3433
3434 fn is_correlation(&self, binding: ColumnBinding) -> bool {
3440 self.correlations.last().is_some_and(|frame| frame.contains(&binding))
3441 }
3442
3443 fn name_of(&self, binding: ColumnBinding, scope: &Scope) -> String {
3449 std::iter::once(scope)
3450 .chain(self.outer_scopes.iter().rev())
3451 .flat_map(|visible| visible.columns.iter())
3452 .find(|column| column.binding == binding)
3453 .map_or_else(|| "a column".to_string(), |column| format!("\"{}\"", column.name))
3454 }
3455
3456 pub(crate) fn over_aggregate(&mut self, expr: ExprRef, scope: &Scope) -> Result<ExprRef> {
3462 let Some(aggregation) = self.aggregation.as_ref() else {
3463 return Ok(expr);
3464 };
3465 let index = aggregation.index;
3466 let groups = aggregation.groups.clone();
3467 for (at, group) in groups.iter().enumerate() {
3468 if self.same_expr(expr, *group) {
3469 let ty = self.plan.expr_type(*group).clone();
3470 return Ok(self.column(index, at, ty));
3471 }
3472 }
3473 let ty = self.plan.expr_type(expr).clone();
3474 match self.plan.expr(expr).clone() {
3475 Expr::Column(binding) if binding.table == index => Ok(expr),
3476 Expr::Column(binding) if self.is_window_output(binding) => Ok(expr),
3481 Expr::Column(binding) if self.joined_above.contains(&binding.table) => Ok(expr),
3486 Expr::Column(binding) if self.is_correlation(binding) => Ok(expr),
3492 Expr::Column(binding) => {
3501 let read = self.ungrouped_correlation(binding).unwrap_or(binding);
3502 let name = self.name_of(read, scope);
3503 Err(Error::binder(format!(
3504 "column {name} must appear in the GROUP BY clause or must be part of an aggregate function"
3505 )))
3506 }
3507 Expr::Constant(_)
3508 | Expr::Aggregate { .. }
3509 | Expr::Window { .. }
3510 | Expr::LambdaParam(_) => Ok(expr),
3511 Expr::Lambda { table, params, body } => {
3515 let body = self.over_aggregate(body, scope)?;
3516 Ok(self.plan.add_expr(Expr::Lambda { table, params, body }, ty))
3517 }
3518 Expr::Cast { input, try_cast } => {
3519 let input = self.over_aggregate(input, scope)?;
3520 Ok(self.plan.add_expr(Expr::Cast { input, try_cast }, ty))
3521 }
3522 Expr::Compare { op, left, right } => {
3523 let left = self.over_aggregate(left, scope)?;
3524 let right = self.over_aggregate(right, scope)?;
3525 Ok(self.plan.add_expr(Expr::Compare { op, left, right }, ty))
3526 }
3527 Expr::Conjunction { op, children } => {
3528 let written = self.plan.expr_list(children).to_vec();
3529 let mut rewritten = Vec::with_capacity(written.len());
3530 for child in written {
3531 rewritten.push(self.over_aggregate(child, scope)?);
3532 }
3533 let children = self.plan.add_expr_list(&rewritten);
3534 Ok(self.plan.add_expr(Expr::Conjunction { op, children }, ty))
3535 }
3536 Expr::Function { name, args } => {
3537 let written = self.plan.expr_list(args).to_vec();
3538 let mut rewritten = Vec::with_capacity(written.len());
3539 for arg in written {
3540 rewritten.push(self.over_aggregate(arg, scope)?);
3541 }
3542 let args = self.plan.add_expr_list(&rewritten);
3543 Ok(self.plan.add_expr(Expr::Function { name, args }, ty))
3544 }
3545 Expr::Case { arms, otherwise } => {
3546 let written = self.plan.arm_list(arms).to_vec();
3547 let mut rewritten = Vec::with_capacity(written.len());
3548 for arm in written {
3549 let when = self.over_aggregate(arm.when, scope)?;
3550 let then = self.over_aggregate(arm.then, scope)?;
3551 rewritten.push(rudb_plan::Arm { when, then });
3552 }
3553 let otherwise = match otherwise {
3554 Some(expr) => Some(self.over_aggregate(expr, scope)?),
3555 None => None,
3556 };
3557 let arms = self.plan.add_arms(&rewritten);
3558 Ok(self.plan.add_expr(Expr::Case { arms, otherwise }, ty))
3559 }
3560 }
3561 }
3562
3563 pub(crate) fn same_expr(&self, left: ExprRef, right: ExprRef) -> bool {
3565 same_expr(&self.plan, left, right)
3566 }
3567}
3568
3569#[derive(Debug, Default)]
3579struct Options {
3580 binary_as_string: bool,
3583 all_varchar: bool,
3585 file_row_number: bool,
3590 given: Given,
3592}
3593
3594impl Options {
3595 fn of(written: &[(&'static str, Value, ExprRef)]) -> Result<Self> {
3602 let mut options = Self::default();
3603 for (parameter, value, _) in written {
3604 match (*parameter, value) {
3605 ("binary_as_string", Value::Boolean(on)) => options.binary_as_string = *on,
3606 ("all_varchar", Value::Boolean(on)) => options.all_varchar = *on,
3607 ("file_row_number", Value::Boolean(on)) => options.file_row_number = *on,
3608 _ => {}
3609 }
3610 }
3611 let named: Vec<(&str, Value)> =
3612 written.iter().map(|(parameter, value, _)| (*parameter, value.clone())).collect();
3613 options.given = csv_given(&named)?;
3614 Ok(options)
3615 }
3616}
3617
3618fn mirror_target(paths: &[String]) -> Option<(String, FileStamp)> {
3621 let [path] = paths else { return None };
3622 let canonical = std::fs::canonicalize(path).ok()?;
3623 let stamp = FileStamp::of(&canonical)?;
3624 Some((canonical.to_str()?.to_string(), stamp))
3625}
3626
3627#[derive(Clone, Copy)]
3629struct Operator {
3630 op: SetOp,
3632 quantifier: Quantifier,
3634 by_name: bool,
3636}
3637
3638struct Merged {
3640 name: String,
3642 ty: LogicalType,
3644 left: Option<usize>,
3646 right: Option<usize>,
3648}
3649
3650fn match_by_position(left: &Scope, right: &Scope) -> Result<Vec<Merged>> {
3654 if left.len() != right.len() {
3655 return Err(Error::binder(format!(
3656 "Set operations can only apply to expressions with the same number of result columns, but left side has {} and right side has {}",
3657 left.len(),
3658 right.len()
3659 )));
3660 }
3661 let mut merged = Vec::with_capacity(left.len());
3662 for (at, (held, other)) in left.columns.iter().zip(&right.columns).enumerate() {
3663 merged.push(Merged {
3664 name: held.name.clone(),
3665 ty: meet(&held.ty, &other.ty)?,
3666 left: Some(at),
3667 right: Some(at),
3668 });
3669 }
3670 Ok(merged)
3671}
3672
3673fn match_by_name(left: &Scope, right: &Scope) -> Result<Vec<Merged>> {
3681 named_once(left)?;
3682 named_once(right)?;
3683 let mut merged = Vec::with_capacity(left.len() + right.len());
3684 for (at, held) in left.columns.iter().enumerate() {
3685 let other = right.columns.iter().position(|column| same_name(&column.name, &held.name));
3686 let ty = match other {
3687 Some(other) => meet(&held.ty, &right.columns[other].ty)?,
3688 None => held.ty.clone(),
3689 };
3690 merged.push(Merged { name: held.name.clone(), ty, left: Some(at), right: other });
3691 }
3692 for (at, held) in right.columns.iter().enumerate() {
3693 if left.columns.iter().any(|column| same_name(&column.name, &held.name)) {
3694 continue;
3695 }
3696 merged.push(Merged {
3697 name: held.name.clone(),
3698 ty: held.ty.clone(),
3699 left: None,
3700 right: Some(at),
3701 });
3702 }
3703 Ok(merged)
3704}
3705
3706fn named_once(scope: &Scope) -> Result<()> {
3712 for (at, held) in scope.columns.iter().enumerate() {
3713 if scope.columns[..at].iter().any(|column| same_name(&column.name, &held.name)) {
3714 return Err(Error::binder(format!(
3715 "UNION (ALL) BY NAME operation doesn't support duplicate names in the SELECT list - the name \"\"{}\"\" occurs multiple times",
3716 held.name
3717 )));
3718 }
3719 }
3720 Ok(())
3721}
3722
3723fn meet(left: &LogicalType, right: &LogicalType) -> Result<LogicalType> {
3725 left.promote(right).ok_or_else(|| {
3726 Error::binder(format!(
3727 "Cannot combine a column of type {left} with a column of type {right} in a set operation"
3728 ))
3729 })
3730}
3731
3732fn null_parameter(function: TableFunction, parameter: &str) -> String {
3741 match parameter {
3742 "header" => format!("\"{parameter}\" expects a non-null boolean value (e.g. TRUE or 1)"),
3743 "all_varchar" => format!("{} \"{parameter}\" cannot be NULL", function.name()),
3744 _ => format!("Cannot use NULL as argument to \"{parameter}\""),
3745 }
3746}
3747
3748fn missing_replacement(name: &str, input: &Scope) -> Error {
3753 Error::binder(format!(
3754 "Column \"{name}\" in REPLACE list not found in FROM clause{}",
3755 input.candidates()
3756 ))
3757}
3758
3759fn subtractable(ty: &LogicalType, ordering: bool) -> bool {
3768 if ty.is_numeric() {
3769 return true;
3770 }
3771 match ty {
3772 LogicalType::Date
3773 | LogicalType::Time
3774 | LogicalType::Timestamp
3775 | LogicalType::TimestampS
3776 | LogicalType::TimestampMs
3777 | LogicalType::TimestampNs
3778 | LogicalType::TimestampTz => true,
3779 LogicalType::TimeTz => ordering,
3780 _ => false,
3781 }
3782}
3783
3784fn refuse_fill(
3793 argument: &LogicalType,
3794 order: &[LogicalType],
3795 distinct: bool,
3796 ignore_nulls: bool,
3797) -> Result<()> {
3798 if !subtractable(argument, false) {
3799 return Err(Error::binder("FILL argument must support subtraction"));
3800 }
3801 let [key] = order else {
3802 return Err(Error::binder("FILL functions must have only one ORDER BY expression"));
3803 };
3804 if !subtractable(key, true) {
3805 return Err(Error::binder("FILL ordering must support subtraction"));
3806 }
3807 if distinct {
3808 return Err(Error::binder(
3809 "DISTINCT is not implemented for the window function \"\"fill\"\"",
3810 ));
3811 }
3812 if ignore_nulls {
3813 return Err(Error::binder(
3814 "RESPECT/IGNORE NULLS is not supported for the window function \"fill\"",
3815 ));
3816 }
3817 Ok(())
3818}
3819
3820fn window_signature(name: &str, types: &[LogicalType]) -> Result<Resolved> {
3827 match kind_of(name) {
3828 Some(FunctionKind::Aggregate | FunctionKind::Window) => resolve(name, types),
3829 Some(FunctionKind::Scalar) => {
3830 Err(Error::catalog(format!("{name} is not an aggregate function")))
3831 }
3832 None => Err(Error::catalog(format!("Aggregate Function with name {name} does not exist!"))),
3833 }
3834}
3835
3836fn same_expr(plan: &Plan, left: ExprRef, right: ExprRef) -> bool {
3838 if left == right {
3839 return true;
3840 }
3841 if plan.expr_type(left) != plan.expr_type(right) {
3842 return false;
3843 }
3844 let lists = |left, right| {
3845 let left: &[ExprRef] = plan.expr_list(left);
3846 let right: &[ExprRef] = plan.expr_list(right);
3847 left.len() == right.len()
3848 && left.iter().zip(right).all(|(&left, &right)| same_expr(plan, left, right))
3849 };
3850 match (plan.expr(left), plan.expr(right)) {
3851 (Expr::Column(left), Expr::Column(right)) => left == right,
3852 (Expr::Constant(left), Expr::Constant(right)) => plan.value(*left) == plan.value(*right),
3853 (
3854 Expr::Cast { input: left, try_cast: left_try },
3855 Expr::Cast { input: right, try_cast: right_try },
3856 ) => left_try == right_try && same_expr(plan, *left, *right),
3857 (
3858 Expr::Compare { op: left_op, left: left_a, right: left_b },
3859 Expr::Compare { op: right_op, left: right_a, right: right_b },
3860 ) => {
3861 left_op == right_op
3862 && same_expr(plan, *left_a, *right_a)
3863 && same_expr(plan, *left_b, *right_b)
3864 }
3865 (
3866 Expr::Conjunction { op: left_op, children: left_children },
3867 Expr::Conjunction { op: right_op, children: right_children },
3868 ) => left_op == right_op && lists(*left_children, *right_children),
3869 (
3870 Expr::Function { name: left_name, args: left_args },
3871 Expr::Function { name: right_name, args: right_args },
3872 ) => plan.string(*left_name) == plan.string(*right_name) && lists(*left_args, *right_args),
3873 (
3874 Expr::Aggregate {
3875 name: left_name,
3876 args: left_args,
3877 distinct: left_distinct,
3878 filter: left_filter,
3879 },
3880 Expr::Aggregate {
3881 name: right_name,
3882 args: right_args,
3883 distinct: right_distinct,
3884 filter: right_filter,
3885 },
3886 ) => {
3887 plan.string(*left_name) == plan.string(*right_name)
3888 && left_distinct == right_distinct
3889 && match (left_filter, right_filter) {
3890 (None, None) => true,
3891 (Some(left), Some(right)) => same_expr(plan, *left, *right),
3892 _ => false,
3893 }
3894 && lists(*left_args, *right_args)
3895 }
3896 (
3900 Expr::Window {
3901 name: left_name,
3902 args: left_args,
3903 distinct: left_distinct,
3904 filter: left_filter,
3905 ignore_nulls: left_nulls,
3906 order: left_order,
3907 },
3908 Expr::Window {
3909 name: right_name,
3910 args: right_args,
3911 distinct: right_distinct,
3912 filter: right_filter,
3913 ignore_nulls: right_nulls,
3914 order: right_order,
3915 },
3916 ) => {
3917 let left_keys = plan.sort_key_list(*left_order);
3920 let right_keys = plan.sort_key_list(*right_order);
3921 plan.string(*left_name) == plan.string(*right_name)
3922 && left_distinct == right_distinct
3923 && left_nulls == right_nulls
3924 && left_keys.len() == right_keys.len()
3925 && left_keys.iter().zip(right_keys).all(|(left, right)| {
3926 left.descending == right.descending
3927 && left.nulls_first == right.nulls_first
3928 && same_expr(plan, left.expr, right.expr)
3929 })
3930 && match (left_filter, right_filter) {
3931 (None, None) => true,
3932 (Some(left), Some(right)) => same_expr(plan, *left, *right),
3933 _ => false,
3934 }
3935 && lists(*left_args, *right_args)
3936 }
3937 (
3938 Expr::Case { arms: left_arms, otherwise: left_otherwise },
3939 Expr::Case { arms: right_arms, otherwise: right_otherwise },
3940 ) => {
3941 let left_arms = plan.arm_list(*left_arms);
3942 let right_arms = plan.arm_list(*right_arms);
3943 left_arms.len() == right_arms.len()
3944 && left_arms.iter().zip(right_arms).all(|(left, right)| {
3945 same_expr(plan, left.when, right.when) && same_expr(plan, left.then, right.then)
3946 })
3947 && match (left_otherwise, right_otherwise) {
3948 (None, None) => true,
3949 (Some(left), Some(right)) => same_expr(plan, *left, *right),
3950 _ => false,
3951 }
3952 }
3953 _ => false,
3954 }
3955}