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 if !excluded {
1991 for (column, bytes) in table.widths() {
1992 self.plan.measure_width(index, &column, bytes);
1993 }
1994 }
1995 let node = self.add_node(Node::Get {
1996 catalog: catalog_name,
1997 schema,
1998 table: table_name,
1999 alias,
2000 index,
2001 columns,
2002 });
2003 Ok((node, scope))
2004 }
2005
2006 fn bind_view(
2018 &mut self,
2019 ast: &Ast,
2020 name: &QualifiedName,
2021 alias: ast::StrRef,
2022 columns: ast::Slice,
2023 ) -> Result<(NodeRef, Scope)> {
2024 let view = self.catalog.view(name)?;
2025 let full = name.to_string();
2026 if self.expanding.contains(&full) {
2027 return Err(Error::binder(format!(
2031 "infinite recursion detected: attempting to recursively bind view \"\"{}\"\"",
2032 name.table
2033 )));
2034 }
2035 let body = parse_ast_with_case(view.sql(), self.semantics.identifier_case())?;
2036 let query = match body.statements.as_slice() {
2037 [ast::Statement::Query(query)] => *query,
2038 _ => return Err(Error::binder(format!("view \"{}\" is not a query", name.table))),
2041 };
2042 self.expanding.push(full);
2043 let bound = self.bind_query(&body, query);
2044 self.expanding.pop();
2045 let (node, mut scope) = bound?;
2046
2047 let aliases: Vec<&str> = view.aliases().iter().map(String::as_str).collect();
2048 if !aliases.is_empty() {
2049 scope.rename(&aliases, "unnamed_subquery")?;
2050 }
2051 view.remember(scope.fields());
2058 let label = if alias == NONE { name.table.clone() } else { ast.string(alias).to_string() };
2059 scope.relabel(&label);
2060 if !columns.is_empty() {
2061 let names: Vec<&str> = ast.name(columns).collect();
2062 scope.rename(&names, &label)?;
2063 }
2064 Ok((node, scope))
2065 }
2066
2067 fn bind_table_function(
2075 &mut self,
2076 ast: &Ast,
2077 name: ast::Slice,
2078 args: ast::Slice,
2079 alias: ast::StrRef,
2080 columns: ast::Slice,
2081 pragma: bool,
2082 ) -> Result<(NodeRef, Scope)> {
2083 let renamed = columns;
2086 let parts: Vec<&str> = ast.name(name).collect();
2087 let function_name = *parts.last().unwrap_or(&"");
2091 if let Some(schema) = parts.iter().rev().nth(1) {
2092 if !schema.eq_ignore_ascii_case("main") && !schema.eq_ignore_ascii_case("system") {
2093 return Err(Error::catalog(format!(
2094 "Table Function with name {} does not exist!",
2095 parts.join(".")
2096 )));
2097 }
2098 }
2099 let Some(called) = TableFunction::lookup(function_name) else {
2103 if pragma {
2104 if args.is_empty() && self.catalog.resolve(&parts).is_ok() {
2110 return self.bind_table(ast, name, alias, columns);
2111 }
2112 let spelled = function_name.strip_prefix("pragma_").unwrap_or(function_name);
2113 return Err(Error::catalog(format!(
2114 "Pragma Function with name {spelled} does not exist!"
2115 )));
2116 }
2117 return Err(Error::catalog(format!(
2118 "Table Function with name {function_name} does not exist!"
2119 )));
2120 };
2121 let written = ast.target_list(args).to_vec();
2122 let empty = Scope::empty();
2123 let previous = std::mem::replace(&mut self.clause, "table function arguments");
2124 let mut bound = Vec::new();
2125 let mut written_options = Vec::new();
2126 for argument in written {
2127 let expr = self.bind_expr(ast, argument.expr, &empty)?;
2128 if argument.alias == NONE {
2129 bound.push(expr);
2130 } else {
2131 let name = ast.string(argument.alias).to_string();
2132 let (parameter, value) = self.named_argument(called, &name, expr)?;
2133 written_options.push((parameter, value, expr));
2134 }
2135 }
2136 self.clause = previous;
2137 let options = Options::of(&written_options)?;
2138
2139 let given: Vec<LogicalType> =
2142 bound.iter().map(|&expr| self.plan.expr_type(expr).clone()).collect();
2143 let resolved = if pragma {
2144 resolve_pragma(function_name, &given)?
2145 } else {
2146 resolve_table(function_name, &given)?
2147 };
2148 let mut cast: Vec<ExprRef> = bound
2149 .iter()
2150 .zip(&resolved.arguments)
2151 .map(|(&expr, ty)| self.checked_cast_to(expr, ty, false))
2152 .collect::<Result<_>>()?;
2153
2154 if resolved.function.answered_when_bound() {
2155 let Columns::Fixed(fields) = resolved.columns else {
2156 return Err(Error::internal("a pragma that resolved to a file"));
2157 };
2158 let [argument] = cast[..] else {
2159 return Err(Error::internal("a pragma that resolved to more than one name"));
2160 };
2161 return self.bind_pragma(ast, resolved.function, &fields, argument, alias, columns);
2162 }
2163 let mut measured = Stat::Unknown;
2166 let mut counted: Vec<(String, Stat<u64>)> = Vec::new();
2167 let mut bounded: Option<Arc<dyn Zones>> = None;
2168 let fields = match resolved.columns {
2169 Columns::Fixed(fields) => fields,
2170 columns => {
2171 let paths = self.file_paths(cast[0], resolved.function.name())?;
2176 let mut mirrorable = None;
2177 if resolved.function == TableFunction::ReadParquet && !options.file_row_number {
2178 if let Some((path, stamp)) = mirror_target(&paths) {
2179 if let Some(name) =
2180 self.catalog.mirror(&path, options.binary_as_string, stamp)
2181 {
2182 let name = name.clone();
2183 let label = if alias == NONE {
2184 resolved.function.name().to_string()
2185 } else {
2186 ast.string(alias).to_string()
2187 };
2188 return self.bind_catalog_table(ast, &name, label, renamed);
2189 }
2190 mirrorable = Some(path);
2191 }
2192 }
2193 let mut fields = match columns {
2194 Columns::Csv => csv_fields(&paths, options.given)?,
2197 _ => {
2198 let footers = self.footers(&paths, mirrorable.as_deref())?;
2199 if let Some(path) = mirrorable.as_deref() {
2200 self.want_mirror(path, options.binary_as_string, &footers.rows);
2201 }
2202 measured = footers.rows;
2203 counted = footers.distincts;
2204 bounded = footers.zones;
2205 footers.fields
2206 }
2207 };
2208 if options.all_varchar {
2209 for field in &mut fields {
2214 field.ty = LogicalType::Varchar;
2215 }
2216 }
2217 if options.binary_as_string {
2218 for field in &mut fields {
2223 if field.ty == LogicalType::Blob {
2224 field.ty = LogicalType::Varchar;
2225 }
2226 }
2227 }
2228 if options.file_row_number {
2229 if fields.iter().any(|field| field.name == FILE_ROW_NUMBER) {
2235 return Err(Error::binder(format!(
2236 "Duplicate column name \"{FILE_ROW_NUMBER}\": the file already has a \
2237 column of that name, so file_row_number cannot add one"
2238 )));
2239 }
2240 fields.push(Field::required(FILE_ROW_NUMBER.to_string(), LogicalType::BigInt));
2241 }
2242 cast = paths.iter().map(|path| self.path_constant(path)).collect();
2243 fields
2244 }
2245 };
2246 let label = if alias == NONE {
2247 resolved.function.name().to_string()
2248 } else {
2249 ast.string(alias).to_string()
2250 };
2251 let names: Vec<&str> = ast.name(columns).collect();
2252 self.table_function_source(
2253 resolved.function,
2254 &cast,
2255 &written_options,
2256 Read { fields, rows: measured, distincts: counted, zones: bounded },
2257 &label,
2258 &names,
2259 )
2260 }
2261
2262 fn bind_pragma(
2275 &mut self,
2276 ast: &Ast,
2277 function: TableFunction,
2278 fields: &[Field],
2279 argument: ExprRef,
2280 alias: ast::StrRef,
2281 columns: ast::Slice,
2282 ) -> Result<(NodeRef, Scope)> {
2283 let written = self.pragma_name(argument, function)?;
2284 let parts = identifier_parts(&written);
2285 let spelled: Vec<&str> = parts.iter().map(String::as_str).collect();
2286 let name = self.catalog.resolve(&spelled)?;
2287 let described = self.described(ast, &name)?;
2288 let mut rows = Vec::with_capacity(described.len());
2289 for (at, field) in described.iter().enumerate() {
2290 let items = if matches!(function, TableFunction::PragmaShow) {
2291 self.describing(field)
2292 } else {
2293 self.table_info(at, field)
2294 };
2295 rows.push(self.plan.add_expr_list(&items));
2296 }
2297 let rows = self.plan.add_rows(&rows);
2298 let held = self.plan.add_fields(fields);
2299 let index = self.fresh_index();
2300 let node = self.add_node(Node::Values { index, columns: held, rows });
2301 let label =
2302 if alias == NONE { function.name().to_string() } else { ast.string(alias).to_string() };
2303 let mut scope = Scope::empty();
2304 for (at, field) in fields.iter().enumerate() {
2305 scope.push(Visible {
2306 table: label.clone(),
2307 name: field.name.clone(),
2308 binding: ColumnBinding::new(index, at as u32),
2309 ty: field.ty.clone(),
2310 not_null: false,
2311 key: None,
2312 default: None,
2313 qualified: false,
2314 also: None,
2315 });
2316 }
2317 if !columns.is_empty() {
2318 let names: Vec<&str> = ast.name(columns).collect();
2319 scope.rename(&names, &label)?;
2320 }
2321 Ok((node, scope))
2322 }
2323
2324 fn pragma_name(&self, argument: ExprRef, function: TableFunction) -> Result<String> {
2334 let Expr::Constant(reference) = *self.plan.expr(argument) else {
2335 return Err(Error::not_implemented(format!(
2336 "{}() given a name that is not a constant",
2337 function.name()
2338 )));
2339 };
2340 match self.plan.value(reference) {
2341 Value::Varchar(name) => Ok(name.clone()),
2342 Value::Null => Ok("NULL".to_string()),
2343 other => {
2344 Err(Error::internal(format!("a pragma name bound as VARCHAR arrived as {other}")))
2345 }
2346 }
2347 }
2348
2349 fn described(&mut self, ast: &Ast, name: &QualifiedName) -> Result<Vec<Field>> {
2360 if self.catalog.entry(name)? == Entry::Table {
2361 return Ok(self.catalog.table(name)?.columns().to_vec());
2362 }
2363 let (_, scope) = self.bind_view(ast, name, NONE, ast::Slice::default())?;
2364 Ok(scope.fields())
2365 }
2366
2367 fn describing(&mut self, field: &Field) -> Vec<ExprRef> {
2369 let written = [
2370 field.name.clone(),
2371 field.ty.to_string(),
2372 if field.not_null { "NO" } else { "YES" }.to_owned(),
2373 ];
2374 let mut items: Vec<ExprRef> =
2375 written.into_iter().map(|text| self.plan.add_constant(Value::Varchar(text))).collect();
2376 for _ in 0..3 {
2377 let empty = self.plan.add_constant(Value::Null);
2378 items.push(self.cast_to(empty, &LogicalType::Varchar));
2379 }
2380 items
2381 }
2382
2383 fn table_info(&mut self, at: usize, field: &Field) -> Vec<ExprRef> {
2389 let cid = self.plan.add_constant(Value::Integer(i32::try_from(at).unwrap_or(i32::MAX)));
2390 let name = self.plan.add_constant(Value::Varchar(field.name.clone()));
2391 let ty = self.plan.add_constant(Value::Varchar(field.ty.to_string()));
2392 let not_null = self.plan.add_constant(Value::Boolean(field.not_null));
2393 let default = self.plan.add_constant(Value::Null);
2394 let default = self.cast_to(default, &LogicalType::Varchar);
2395 let key = self.plan.add_constant(Value::Boolean(false));
2396 vec![cid, name, ty, not_null, default, key]
2397 }
2398
2399 fn named_argument(
2413 &mut self,
2414 function: TableFunction,
2415 name: &str,
2416 expr: ExprRef,
2417 ) -> Result<(&'static str, Value)> {
2418 let known = function
2419 .parameters()
2420 .iter()
2421 .find(|(parameter, _)| parameter.eq_ignore_ascii_case(name));
2422 let Some((parameter, wanted)) = known else {
2423 let candidates: Vec<String> = function
2424 .parameters()
2425 .iter()
2426 .map(|(parameter, ty)| format!(" {parameter} {ty}"))
2427 .collect();
2428 return Err(Error::binder(format!(
2429 "Invalid named parameter \"{name}\" for function {}\nCandidates:\n{}\n",
2430 function.name(),
2431 candidates.join("\n")
2432 )));
2433 };
2434 let Expr::Constant(reference) = *self.plan.expr(expr) else {
2435 return Err(Error::not_implemented(format!(
2436 "the named parameter {parameter} with a value that is not a constant"
2437 )));
2438 };
2439 let value = self.plan.value(reference).clone();
2440 if value == Value::Null {
2441 return Err(Error::binder(null_parameter(function, parameter)));
2442 }
2443 let given = self.plan.expr_type(expr).clone();
2444 if given != *wanted {
2445 return Err(Error::not_implemented(format!(
2446 "the named parameter {parameter} given a {given} where a {wanted} was wanted"
2447 )));
2448 }
2449 Ok((parameter, value))
2450 }
2451
2452 fn bind_replacement_scan(
2463 &mut self,
2464 ast: &Ast,
2465 parts: &[&str],
2466 alias: ast::StrRef,
2467 columns: ast::Slice,
2468 missing: Error,
2469 ) -> Result<(NodeRef, Scope)> {
2470 let [path] = parts else { return Err(missing) };
2471 let path = *path;
2472 let extension = path.rsplit_once('.').map(|(_, after)| after).unwrap_or_default();
2473 let Some(function) = Self::reader_for(extension) else {
2474 if is_file(path) {
2475 return Err(Error::binder(format!(
2480 "No extension found that is capable of reading the file \"{path}\"\n* If this \
2481 file is a supported file format you can explicitly use the reader functions, \
2482 such as read_csv, read_json or read_parquet"
2483 )));
2484 }
2485 return Err(missing);
2486 };
2487 let paths = files(path)?;
2492 let label = if alias == NONE {
2498 if is_pattern(path) {
2499 path.to_string()
2500 } else {
2501 let file = path.rsplit_once('/').map_or(path, |(_, file)| file);
2502 file.rsplit_once('.').map_or(file, |(stem, _)| stem).to_string()
2503 }
2504 } else {
2505 ast.string(alias).to_string()
2506 };
2507 let mut mirrorable = None;
2508 if function == TableFunction::ReadParquet {
2509 if let Some((canonical, stamp)) = mirror_target(&paths) {
2510 if let Some(name) = self.catalog.mirror(&canonical, false, stamp) {
2511 let name = name.clone();
2512 return self.bind_catalog_table(ast, &name, label, columns);
2513 }
2514 mirrorable = Some(canonical);
2515 }
2516 }
2517 let read = match function {
2518 TableFunction::ReadParquet => {
2519 let footers = self.footers(&paths, mirrorable.as_deref())?;
2520 if let Some(canonical) = mirrorable.as_deref() {
2521 self.want_mirror(canonical, false, &footers.rows);
2522 }
2523 Read {
2524 fields: footers.fields,
2525 rows: footers.rows,
2526 distincts: footers.distincts,
2527 zones: footers.zones,
2528 }
2529 }
2530 _ => Read::uncounted(csv_fields(&paths, Given::default())?),
2531 };
2532 let arguments: Vec<ExprRef> = paths.iter().map(|path| self.path_constant(path)).collect();
2533 let names: Vec<&str> = ast.name(columns).collect();
2534 self.table_function_source(function, &arguments, &[], read, &label, &names)
2535 }
2536
2537 fn footers(&self, paths: &[String], mirrorable: Option<&str>) -> Result<Footers> {
2543 if let Some(path) = mirrorable.filter(|_| self.outlined) {
2544 let outline = parquet_outline(path)?;
2545 if outline.rows.value().is_some() {
2546 return Ok(outline);
2547 }
2548 }
2549 parquet_footers(paths)
2550 }
2551
2552 fn want_mirror(&mut self, path: &str, binary_as_string: bool, rows: &Stat<u64>) {
2556 if let Some(&rows) = rows.value() {
2557 self.plan.want_mirror(path, binary_as_string, rows);
2558 }
2559 }
2560
2561 fn path_constant(&mut self, path: &str) -> ExprRef {
2563 let value = self.plan.add_value(Value::Varchar(path.to_string()));
2564 self.plan.add_expr(Expr::Constant(value), LogicalType::Varchar)
2565 }
2566
2567 fn reader_for(extension: &str) -> Option<TableFunction> {
2574 if extension.eq_ignore_ascii_case("parquet") {
2575 return Some(TableFunction::ReadParquet);
2576 }
2577 if extension.eq_ignore_ascii_case("csv") || extension.eq_ignore_ascii_case("tsv") {
2578 return Some(TableFunction::ReadCsv);
2579 }
2580 None
2581 }
2582
2583 fn table_function_source(
2593 &mut self,
2594 function: TableFunction,
2595 args: &[ExprRef],
2596 written: &[(&'static str, Value, ExprRef)],
2597 read: Read,
2598 label: &str,
2599 names: &[&str],
2600 ) -> Result<(NodeRef, Scope)> {
2601 let Read { fields, rows, distincts, zones } = read;
2602 let index = self.fresh_index();
2603 if rows.is_known() {
2608 self.plan.measure(index, rows);
2609 }
2610 for (column, distinct) in distincts {
2611 self.plan.measure_distinct(index, &column, distinct);
2612 }
2613 if let Some(zones) = zones {
2614 self.plan.set_zones(index, zones);
2615 }
2616 let mut scope = Scope::empty();
2617 for (at, field) in fields.iter().enumerate() {
2618 scope.push(Visible {
2619 table: label.to_string(),
2620 name: field.name.clone(),
2621 binding: ColumnBinding::new(index, at as u32),
2622 ty: field.ty.clone(),
2623 not_null: false,
2626 key: None,
2627 default: None,
2628 qualified: false,
2629 also: None,
2630 });
2631 }
2632 if !names.is_empty() {
2633 scope.rename(names, label)?;
2634 } else if matches!(function, TableFunction::Range | TableFunction::GenerateSeries) {
2635 for column in &mut scope.columns {
2638 column.also = Some(std::mem::replace(&mut column.name, label.to_string()));
2639 }
2640 }
2641 let function = self.plan.intern(function.name());
2642 let args = self.plan.add_expr_list(args);
2643 let named: Vec<u32> =
2644 written.iter().map(|(parameter, _, _)| self.plan.intern(parameter)).collect();
2645 let settings: Vec<ExprRef> = written.iter().map(|(_, _, expr)| *expr).collect();
2646 let options = self.plan.add_name_list(&named);
2647 let settings = self.plan.add_expr_list(&settings);
2648 let columns = self.plan.add_fields(&fields);
2649 let node = self.add_node(Node::TableFunction {
2650 index,
2651 function,
2652 args,
2653 options,
2654 settings,
2655 columns,
2656 });
2657 Ok((node, scope))
2658 }
2659
2660 fn file_paths(&self, expr: ExprRef, name: &str) -> Result<Vec<String>> {
2667 let mut paths = Vec::new();
2668 for pattern in self.file_patterns(expr, name)? {
2669 paths.extend(files(&pattern)?);
2670 }
2671 Ok(paths)
2672 }
2673
2674 fn file_patterns(&self, expr: ExprRef, name: &str) -> Result<Vec<String>> {
2693 let Some(value) = fold::value_of(&self.plan, expr)? else {
2694 return Err(Error::not_implemented(
2695 "a table function file name that is not a constant",
2696 ));
2697 };
2698 match value {
2699 Value::Varchar(path) => Ok(vec![path]),
2700 Value::Null => Err(Error::parser(format!("{name} cannot take NULL list as parameter"))),
2702 Value::List { values, .. } if values.is_empty() => {
2707 Err(Error::io(format!("\"{name}\" needs at least one file to read")))
2708 }
2709 Value::List { values, .. } => values
2710 .iter()
2711 .map(|value| match value {
2712 Value::Varchar(path) => Ok(path.clone()),
2713 _ => Err(Error::parser(format!(
2714 "{name} reader cannot take NULL input as parameter"
2715 ))),
2716 })
2717 .collect(),
2718 other => {
2719 Err(Error::internal(format!("a file name bound as VARCHAR arrived as {other}")))
2720 }
2721 }
2722 }
2723
2724 fn side_of(
2742 &self,
2743 pending: &PendingSubquery,
2744 left_tables: &[u32],
2745 right_tables: &[u32],
2746 ) -> Option<Side> {
2747 let mut needs_left = false;
2748 let mut needs_right = false;
2749 let mut note = |binding: ColumnBinding| {
2750 needs_left |= left_tables.contains(&binding.table);
2751 needs_right |= right_tables.contains(&binding.table);
2752 };
2753 for &binding in &pending.reads {
2754 note(binding);
2755 }
2756 for &condition in &pending.conditions {
2761 self.plan.read_columns(condition, &mut |_, binding| note(binding));
2762 }
2763 match (needs_left, needs_right) {
2764 (true, true) => None,
2765 (_, true) => Some(Side::Right),
2766 _ => Some(Side::Left),
2767 }
2768 }
2769
2770 #[allow(clippy::too_many_arguments)]
2791 fn bind_pair_dependent_join(
2792 &mut self,
2793 kind: ast::JoinKind,
2794 independent: bool,
2795 left: NodeRef,
2796 right: NodeRef,
2797 pair: Vec<PendingSubquery>,
2798 conditions: Vec<ExprRef>,
2799 scope: Scope,
2800 ) -> Result<(NodeRef, Scope)> {
2801 if kind != ast::JoinKind::Inner {
2802 return Err(Error::not_implemented(
2803 "a subquery that reads both sides of that join, written in the condition of a join \
2804 that is not an inner join"
2805 .to_string(),
2806 ));
2807 }
2808 if !independent {
2811 return Err(Error::not_implemented(
2812 "a subquery that reads both sides of that join, written in the condition of a join \
2813 whose right side is lateral"
2814 .to_string(),
2815 ));
2816 }
2817 let mut node = self.add_node(Node::CrossProduct { left, right });
2818 for pending in pair {
2819 node = self.attach_subquery(node, pending);
2820 }
2821 let mut conditions = conditions.into_iter();
2825 let mut predicate = conditions.next().expect("a join condition was bound");
2826 for next in conditions {
2827 let children = self.plan.add_expr_list(&[predicate, next]);
2828 let conjunction = Expr::Conjunction { op: ConjunctionOp::And, children };
2829 predicate = self.plan.add_expr(conjunction, LogicalType::Boolean);
2830 }
2831 let node = self.add_node(Node::Filter { input: node, predicate });
2832 Ok((node, scope))
2833 }
2834
2835 #[allow(clippy::too_many_arguments)]
2836 fn bind_join(
2837 &mut self,
2838 ast: &Ast,
2839 left: ast::SourceRef,
2840 right: ast::SourceRef,
2841 kind: ast::JoinKind,
2842 natural: bool,
2843 on: ast::ExprRef,
2844 using: ast::Slice,
2845 ) -> Result<(NodeRef, Scope)> {
2846 let (left_node, left_scope) = self.bind_source(ast, left)?;
2847 let (right_node, right_scope, correlated) = self.bind_lateral(ast, right, &left_scope)?;
2848 if !correlated.is_empty()
2852 && !matches!(kind, ast::JoinKind::Inner | ast::JoinKind::Cross | ast::JoinKind::Left)
2853 {
2854 return Err(Error::binder(
2855 "The combining JOIN type must be INNER or LEFT for a LATERAL reference",
2856 ));
2857 }
2858 let split = left_scope.len();
2859 let left_tables: Vec<u32> =
2865 left_scope.columns.iter().map(|column| column.binding.table).collect();
2866 let right_tables: Vec<u32> =
2867 right_scope.columns.iter().map(|column| column.binding.table).collect();
2868 let mut scope = left_scope.concat(right_scope);
2869
2870 let merged: Vec<String> = if natural {
2873 let mut names = Vec::new();
2874 for (at, column) in scope.columns.iter().enumerate().take(split) {
2875 if scope.columns[split..].iter().any(|right| same_name(&right.name, &column.name))
2876 && !names.iter().any(|held: &String| same_name(held, &column.name))
2877 {
2878 let _ = at;
2879 names.push(column.name.clone());
2880 }
2881 }
2882 names
2883 } else {
2884 let mut names: Vec<String> = Vec::new();
2890 for name in ast.name(using) {
2891 if !names.iter().any(|held| same_name(held, name)) {
2892 names.push(name.to_string());
2893 }
2894 }
2895 names
2896 };
2897
2898 let mut conditions = Vec::new();
2899 let mut dropped = Vec::new();
2900 for name in &merged {
2901 let left_at = scope.columns[..split]
2902 .iter()
2903 .position(|column| same_name(&column.name, name))
2904 .ok_or_else(|| {
2905 Error::binder(format!(
2906 "column \"{name}\" specified in USING clause does not exist in left table"
2907 ))
2908 })?;
2909 let right_at = scope.columns[split..]
2910 .iter()
2911 .position(|column| same_name(&column.name, name))
2912 .map(|at| at + split)
2913 .ok_or_else(|| {
2914 Error::binder(format!(
2915 "column \"{name}\" specified in USING clause does not exist in right table"
2916 ))
2917 })?;
2918 let left_column = &scope.columns[left_at];
2919 let (left_binding, left_type) = (left_column.binding, left_column.ty.clone());
2920 let right_column = &scope.columns[right_at];
2921 let (right_binding, right_type) = (right_column.binding, right_column.ty.clone());
2922 let left_expr = self.plan.add_expr(Expr::Column(left_binding), left_type);
2923 let right_expr = self.plan.add_expr(Expr::Column(right_binding), right_type);
2924 conditions.push(self.compare(rudb_plan::CompareOp::Equal, left_expr, right_expr)?);
2925 dropped.push(right_at);
2926 }
2927 dropped.sort_unstable();
2930 for at in dropped.into_iter().rev() {
2931 scope.remove(at);
2932 }
2933
2934 let mut left_node = left_node;
2935 let mut right_node = right_node;
2936 let mut pair = Vec::new();
2937 if on != NONE {
2938 if !merged.is_empty() {
2939 return Err(Error::binder("a join cannot have both ON and USING"));
2940 }
2941 self.clause = "JOIN condition";
2942 let waiting = self.scalar_subqueries.len();
2943 let predicate = self.bind_expr(ast, on, &scope)?;
2944 conditions.push(self.as_boolean(predicate, "JOIN")?);
2945 for pending in self.scalar_subqueries.split_off(waiting) {
2946 match self.side_of(&pending, &left_tables, &right_tables) {
2947 Some(Side::Right) => right_node = self.attach_subquery(right_node, pending),
2948 Some(Side::Left) => left_node = self.attach_subquery(left_node, pending),
2949 None => pair.push(pending),
2950 }
2951 }
2952 }
2953
2954 if kind == ast::JoinKind::Cross && !conditions.is_empty() {
2955 return Err(Error::binder("a CROSS JOIN cannot have a condition"));
2956 }
2957 if !pair.is_empty() {
2958 return self.bind_pair_dependent_join(
2959 kind,
2960 correlated.is_empty(),
2961 left_node,
2962 right_node,
2963 pair,
2964 conditions,
2965 scope,
2966 );
2967 }
2968 if correlated.is_empty()
2972 && conditions.is_empty()
2973 && matches!(kind, ast::JoinKind::Cross | ast::JoinKind::Inner)
2974 {
2975 let node = self.add_node(Node::CrossProduct { left: left_node, right: right_node });
2976 return Ok((node, scope));
2977 }
2978 if matches!(kind, ast::JoinKind::Semi | ast::JoinKind::Anti) {
2987 scope.truncate(split);
2988 }
2989 let kind = match kind {
2990 ast::JoinKind::Inner | ast::JoinKind::Cross => JoinKind::Inner,
2991 ast::JoinKind::Left => JoinKind::Left,
2992 ast::JoinKind::Right => JoinKind::Right,
2993 ast::JoinKind::Full => JoinKind::Full,
2994 ast::JoinKind::Semi => JoinKind::Semi,
2995 ast::JoinKind::Anti => JoinKind::Anti,
2996 ast::JoinKind::Positional => JoinKind::Positional,
2997 };
2998 let conditions = self.plan.add_expr_list(&conditions);
2999 let node = if correlated.is_empty() {
3000 self.add_node(Node::Join {
3001 left: left_node,
3002 right: right_node,
3003 kind,
3004 conditions,
3005 build: BuildSide::default(),
3006 })
3007 } else {
3008 self.add_node(Node::DependentJoin {
3009 left: left_node,
3010 right: right_node,
3011 kind,
3012 conditions,
3013 })
3014 };
3015 Ok((node, scope))
3016 }
3017
3018 fn bind_filter(
3026 &mut self,
3027 ast: &Ast,
3028 filter: ast::ExprRef,
3029 scope: &Scope,
3030 ) -> Result<Option<ExprRef>> {
3031 if filter == NONE {
3032 return Ok(None);
3033 }
3034 let bound = self.bind_expr(ast, filter, scope)?;
3035 Ok(Some(self.checked_cast_to(bound, &LogicalType::Boolean, false)?))
3036 }
3037
3038 pub(crate) fn bind_aggregate(
3043 &mut self,
3044 ast: &Ast,
3045 name: &str,
3046 args: &[ast::ExprRef],
3047 distinct: bool,
3048 filter: ast::ExprRef,
3049 scope: &Scope,
3050 ) -> Result<ExprRef> {
3051 let frames = std::mem::take(&mut self.lambda_frames);
3052 let bound = self.bind_aggregate_over_rows(ast, name, args, distinct, filter, scope);
3053 self.lambda_frames = frames;
3054 bound
3055 }
3056
3057 fn bind_aggregate_over_rows(
3058 &mut self,
3059 ast: &Ast,
3060 name: &str,
3061 args: &[ast::ExprRef],
3062 distinct: bool,
3063 filter: ast::ExprRef,
3064 scope: &Scope,
3065 ) -> Result<ExprRef> {
3066 if self.in_filter {
3067 return Err(Error::binder("aggregate functions are not allowed in FILTER"));
3068 }
3069 if self.in_aggregate {
3070 return Err(Error::binder(format!(
3071 "aggregate function calls cannot be nested, and {name}() is inside one"
3072 )));
3073 }
3074 if self.aggregation.is_none() {
3075 return Err(Error::binder(format!(
3076 "aggregate function calls cannot be used in the {}",
3077 self.clause
3078 )));
3079 }
3080 self.in_aggregate = true;
3085 self.in_filter = true;
3086 let filter = self.bind_filter(ast, filter, scope);
3087 self.in_filter = false;
3088 self.in_aggregate = false;
3089 let filter = filter?;
3090
3091 self.in_aggregate = true;
3092 let mut bound = Vec::with_capacity(args.len());
3093 let mut failure = None;
3094 for &arg in args {
3095 match self.bind_expr(ast, arg, scope) {
3096 Ok(expr) => bound.push(expr),
3097 Err(error) => {
3098 failure = Some(error);
3099 break;
3100 }
3101 }
3102 }
3103 self.in_aggregate = false;
3104 if let Some(error) = failure {
3105 return Err(error);
3106 }
3107
3108 let types: Vec<LogicalType> =
3109 bound.iter().map(|&arg| self.plan.expr_type(arg).clone()).collect();
3110 let resolved = resolve(name, &types)?;
3111 if resolved.name == "string_agg"
3114 && bound.len() == 2
3115 && !matches!(fold::value_of(&self.plan, bound[1]), Ok(Some(_)))
3116 {
3117 return Err(Error::binder(
3118 "The \"separator\" argument in function \"string_agg\" must be a constant expression",
3119 ));
3120 }
3121 let mut cast = Vec::with_capacity(bound.len());
3122 for (arg, wanted) in bound.iter().zip(&resolved.arguments) {
3123 cast.push(self.checked_cast_to(*arg, wanted, false)?);
3124 }
3125 let args = self.plan.add_expr_list(&cast);
3126 let name = self.plan.intern(resolved.name);
3127 let ty = resolved.returns;
3128 let call = self.plan.add_expr(Expr::Aggregate { name, args, distinct, filter }, ty.clone());
3129
3130 let existing = self.aggregation.as_ref().map(|held| held.aggregates.clone());
3133 let existing = existing.unwrap_or_default();
3134 let at = match existing.iter().position(|&held| self.same_expr(held, call)) {
3135 Some(at) => at,
3136 None => {
3137 let aggregation = self.aggregation.as_mut().expect("checked above");
3138 aggregation.aggregates.push(call);
3139 aggregation.aggregates.len() - 1
3140 }
3141 };
3142 let aggregation = self.aggregation.as_ref().expect("checked above");
3143 let (index, groups) = (aggregation.index, aggregation.groups.len());
3144 Ok(self.column(index, groups + at, ty))
3145 }
3146
3147 pub(crate) fn bind_window(
3158 &mut self,
3159 ast: &Ast,
3160 written: &WindowCall<'_>,
3161 scope: &Scope,
3162 ) -> Result<ExprRef> {
3163 let frames = std::mem::take(&mut self.lambda_frames);
3164 let bound = self.bind_window_over_rows(ast, written, scope);
3165 self.lambda_frames = frames;
3166 bound
3167 }
3168
3169 fn bind_window_over_rows(
3170 &mut self,
3171 ast: &Ast,
3172 written: &WindowCall<'_>,
3173 scope: &Scope,
3174 ) -> Result<ExprRef> {
3175 let WindowCall { name, args, distinct, filter, ignore_nulls, spec, .. } = *written;
3176 if self.in_aggregate {
3177 return Err(Error::binder(
3178 "aggregate function calls cannot contain window function calls",
3179 ));
3180 }
3181 if self.in_window {
3182 return Err(Error::binder("window function calls cannot be nested"));
3183 }
3184 let clause = if self.clause == "JOIN condition" { "WHERE clause" } else { self.clause };
3188 if clause != "SELECT clause" && clause != "ORDER BY clause" {
3189 return Err(Error::binder(format!("{clause} cannot contain window functions!")));
3190 }
3191
3192 let starred = args.iter().any(|&arg| {
3196 matches!(ast.expr(arg), ast::Expr::Star { qualifier, replacements }
3197 if qualifier.is_empty() && replacements.is_empty())
3198 });
3199 let (name, args): (&str, &[ast::ExprRef]) = if starred {
3200 if !same_name(name, "count") || args.len() != 1 {
3201 return Err(Error::binder(format!("* is not allowed in {name}()")));
3202 }
3203 ("count_star", &[])
3204 } else if same_name(name, "count") && args.is_empty() {
3205 ("count_star", &[])
3208 } else {
3209 (name, args)
3210 };
3211
3212 let held = ast.window(spec);
3213 self.in_window = true;
3214 let parts = self.window_parts(ast, written, args, held, scope);
3215 let filter = if parts.is_ok() { self.bind_filter(ast, filter, scope) } else { Ok(None) };
3220 self.in_window = false;
3221 let parts = parts?;
3222 let filter = filter?;
3223 let offsets = [parts.frame.start, parts.frame.end]
3226 .iter()
3227 .any(|end| matches!(end, WindowBound::Preceding(_) | WindowBound::Following(_)));
3228 if parts.frame.unit == WindowUnit::Range && offsets && parts.order.len() != 1 {
3229 return Err(Error::binder("RANGE frames must have only one ORDER BY expression"));
3230 }
3231
3232 let types: Vec<LogicalType> =
3233 parts.args.iter().map(|&arg| self.plan.expr_type(arg).clone()).collect();
3234 let resolved = window_signature(name, &types)?;
3235 if resolved.name == "fill" {
3238 let keys: Vec<LogicalType> =
3239 parts.order.iter().map(|key| self.plan.expr_type(key.expr).clone()).collect();
3240 refuse_fill(&types[0], &keys, distinct, ignore_nulls)?;
3241 }
3242 if distinct && kind_of(resolved.name) == Some(FunctionKind::Window) {
3246 return Err(Error::binder(format!(
3247 "DISTINCT is not implemented for the window function \"\"{name}\"\""
3248 )));
3249 }
3250 if filter.is_some() && kind_of(resolved.name) == Some(FunctionKind::Window) {
3253 return Err(Error::binder(format!(
3254 "FILTER is not implemented for the window function \"\"{name}\"\""
3255 )));
3256 }
3257 if !parts.inner.is_empty() && kind_of(resolved.name) == Some(FunctionKind::Window) {
3265 let counts = matches!(resolved.name, "first_value" | "last_value" | "nth_value");
3266 if !counts {
3267 if parts.frame.exclude != WindowExclude::NoOthers {
3268 return Err(Error::binder(format!(
3269 "EXCLUDE is not supported for the window function \"\"{}\"\"",
3270 resolved.name
3271 )));
3272 }
3273 return Err(Error::not_implemented(format!(
3274 "ORDER BY inside the arguments of the window function \"{}\"",
3275 resolved.name
3276 )));
3277 }
3278 }
3279 let mut cast = Vec::with_capacity(parts.args.len());
3280 for (arg, wanted) in parts.args.iter().zip(&resolved.arguments) {
3281 cast.push(self.checked_cast_to(*arg, wanted, false)?);
3282 }
3283 let args = self.plan.add_expr_list(&cast);
3284 let order = self.plan.add_sort_keys(&parts.inner);
3285 let name = self.plan.intern(resolved.name);
3286 let ty = resolved.returns;
3287 let call = self.plan.add_expr(
3288 Expr::Window { name, args, distinct, filter, ignore_nulls, order },
3289 ty.clone(),
3290 );
3291
3292 let at = self.window_run(parts.partition, parts.order, parts.frame, call);
3293 let index = self.windows.last().expect("the run was just filed").index;
3294 Ok(self.column(index, at, ty))
3295 }
3296
3297 fn window_run(
3304 &mut self,
3305 partition: Vec<ExprRef>,
3306 order: Vec<SortKey>,
3307 frame: WindowFrame,
3308 call: ExprRef,
3309 ) -> usize {
3310 let matches = self.windows.last().is_some_and(|run| {
3311 run.frame == frame
3312 && run.partition.len() == partition.len()
3313 && run.order.len() == order.len()
3314 && run.partition.iter().zip(&partition).all(|(&l, &r)| self.same_expr(l, r))
3315 && run.order.iter().zip(&order).all(|(l, r)| {
3316 l.descending == r.descending
3317 && l.nulls_first == r.nulls_first
3318 && self.same_expr(l.expr, r.expr)
3319 })
3320 });
3321 if !matches {
3322 let index = self.fresh_index();
3323 self.windows.push(WindowRun { index, partition, order, frame, calls: Vec::new() });
3324 }
3325 let calls = self.windows.last().expect("a run is open").calls.clone();
3328 if let Some(at) = calls.iter().position(|&held| self.same_expr(held, call)) {
3329 return at;
3330 }
3331 let run = self.windows.last_mut().expect("a run is open");
3332 run.calls.push(call);
3333 run.calls.len() - 1
3334 }
3335
3336 fn window_parts(
3342 &mut self,
3343 ast: &Ast,
3344 written: &WindowCall<'_>,
3345 args: &[ast::ExprRef],
3346 held: ast::WindowSpec,
3347 scope: &Scope,
3348 ) -> Result<WindowParts> {
3349 let mut bound = Vec::with_capacity(args.len());
3350 for &arg in args {
3351 let expr = self.bind_expr(ast, arg, scope)?;
3352 bound.push(self.over_aggregate(expr, scope)?);
3353 }
3354 let mut inner = Vec::new();
3358 for item in ast.order_list(written.order).to_vec() {
3359 let expr = self.bind_expr(ast, item.expr, scope)?;
3360 let expr = self.over_aggregate(expr, scope)?;
3361 inner.push(self.sort_key(expr, item));
3362 }
3363 let mut partition = Vec::new();
3364 for &key in ast.expr_list(held.partition) {
3365 let expr = self.bind_expr(ast, key, scope)?;
3366 partition.push(self.over_aggregate(expr, scope)?);
3367 }
3368 let mut order = Vec::new();
3369 for item in ast.order_list(held.order).to_vec() {
3370 let expr = self.bind_expr(ast, item.expr, scope)?;
3371 let expr = self.over_aggregate(expr, scope)?;
3372 order.push(self.sort_key(expr, item));
3373 }
3374 let frame = WindowFrame {
3375 unit: match held.unit {
3376 ast::WindowUnit::Rows => WindowUnit::Rows,
3377 ast::WindowUnit::Range => WindowUnit::Range,
3378 ast::WindowUnit::Groups => WindowUnit::Groups,
3379 },
3380 start: self.window_bound(ast, held.start, scope)?,
3381 end: self.window_bound(ast, held.end, scope)?,
3382 exclude: match held.exclude {
3383 ast::WindowExclude::NoOthers => WindowExclude::NoOthers,
3384 ast::WindowExclude::CurrentRow => WindowExclude::CurrentRow,
3385 ast::WindowExclude::Group => WindowExclude::Group,
3386 ast::WindowExclude::Ties => WindowExclude::Ties,
3387 },
3388 };
3389 Ok(WindowParts { args: bound, partition, order, inner, frame })
3390 }
3391
3392 fn window_bound(
3394 &mut self,
3395 ast: &Ast,
3396 bound: ast::WindowBound,
3397 scope: &Scope,
3398 ) -> Result<WindowBound> {
3399 let offset = |binder: &mut Self, written| {
3400 let expr = binder.bind_expr(ast, written, scope)?;
3401 binder.over_aggregate(expr, scope)
3402 };
3403 Ok(match bound {
3404 ast::WindowBound::UnboundedPreceding => WindowBound::UnboundedPreceding,
3405 ast::WindowBound::CurrentRow => WindowBound::CurrentRow,
3406 ast::WindowBound::UnboundedFollowing => WindowBound::UnboundedFollowing,
3407 ast::WindowBound::Preceding(written) => WindowBound::Preceding(offset(self, written)?),
3408 ast::WindowBound::Following(written) => WindowBound::Following(offset(self, written)?),
3409 })
3410 }
3411
3412 fn group_of(&self, read: ColumnBinding) -> Option<usize> {
3418 self.aggregation.as_ref()?.groups.iter().position(
3419 |group| matches!(*self.plan.expr(*group), Expr::Column(binding) if binding == read),
3420 )
3421 }
3422
3423 fn ungrouped_correlation(&self, binding: ColumnBinding) -> Option<ColumnBinding> {
3429 let pending =
3430 self.scalar_subqueries.iter().find(|pending| pending.index == binding.table)?;
3431 pending.reads.iter().copied().find(|read| self.group_of(*read).is_none())
3432 }
3433
3434 fn is_window_output(&self, binding: ColumnBinding) -> bool {
3436 self.windows.iter().any(|run| run.index == binding.table)
3437 }
3438
3439 fn is_correlation(&self, binding: ColumnBinding) -> bool {
3445 self.correlations.last().is_some_and(|frame| frame.contains(&binding))
3446 }
3447
3448 fn name_of(&self, binding: ColumnBinding, scope: &Scope) -> String {
3454 std::iter::once(scope)
3455 .chain(self.outer_scopes.iter().rev())
3456 .flat_map(|visible| visible.columns.iter())
3457 .find(|column| column.binding == binding)
3458 .map_or_else(|| "a column".to_string(), |column| format!("\"{}\"", column.name))
3459 }
3460
3461 pub(crate) fn over_aggregate(&mut self, expr: ExprRef, scope: &Scope) -> Result<ExprRef> {
3467 let Some(aggregation) = self.aggregation.as_ref() else {
3468 return Ok(expr);
3469 };
3470 let index = aggregation.index;
3471 let groups = aggregation.groups.clone();
3472 for (at, group) in groups.iter().enumerate() {
3473 if self.same_expr(expr, *group) {
3474 let ty = self.plan.expr_type(*group).clone();
3475 return Ok(self.column(index, at, ty));
3476 }
3477 }
3478 let ty = self.plan.expr_type(expr).clone();
3479 match self.plan.expr(expr).clone() {
3480 Expr::Column(binding) if binding.table == index => Ok(expr),
3481 Expr::Column(binding) if self.is_window_output(binding) => Ok(expr),
3486 Expr::Column(binding) if self.joined_above.contains(&binding.table) => Ok(expr),
3491 Expr::Column(binding) if self.is_correlation(binding) => Ok(expr),
3497 Expr::Column(binding) => {
3506 let read = self.ungrouped_correlation(binding).unwrap_or(binding);
3507 let name = self.name_of(read, scope);
3508 Err(Error::binder(format!(
3509 "column {name} must appear in the GROUP BY clause or must be part of an aggregate function"
3510 )))
3511 }
3512 Expr::Constant(_)
3513 | Expr::Aggregate { .. }
3514 | Expr::Window { .. }
3515 | Expr::LambdaParam(_) => Ok(expr),
3516 Expr::Lambda { table, params, body } => {
3520 let body = self.over_aggregate(body, scope)?;
3521 Ok(self.plan.add_expr(Expr::Lambda { table, params, body }, ty))
3522 }
3523 Expr::Cast { input, try_cast } => {
3524 let input = self.over_aggregate(input, scope)?;
3525 Ok(self.plan.add_expr(Expr::Cast { input, try_cast }, ty))
3526 }
3527 Expr::Compare { op, left, right } => {
3528 let left = self.over_aggregate(left, scope)?;
3529 let right = self.over_aggregate(right, scope)?;
3530 Ok(self.plan.add_expr(Expr::Compare { op, left, right }, ty))
3531 }
3532 Expr::Conjunction { op, children } => {
3533 let written = self.plan.expr_list(children).to_vec();
3534 let mut rewritten = Vec::with_capacity(written.len());
3535 for child in written {
3536 rewritten.push(self.over_aggregate(child, scope)?);
3537 }
3538 let children = self.plan.add_expr_list(&rewritten);
3539 Ok(self.plan.add_expr(Expr::Conjunction { op, children }, ty))
3540 }
3541 Expr::Function { name, args } => {
3542 let written = self.plan.expr_list(args).to_vec();
3543 let mut rewritten = Vec::with_capacity(written.len());
3544 for arg in written {
3545 rewritten.push(self.over_aggregate(arg, scope)?);
3546 }
3547 let args = self.plan.add_expr_list(&rewritten);
3548 Ok(self.plan.add_expr(Expr::Function { name, args }, ty))
3549 }
3550 Expr::Case { arms, otherwise } => {
3551 let written = self.plan.arm_list(arms).to_vec();
3552 let mut rewritten = Vec::with_capacity(written.len());
3553 for arm in written {
3554 let when = self.over_aggregate(arm.when, scope)?;
3555 let then = self.over_aggregate(arm.then, scope)?;
3556 rewritten.push(rudb_plan::Arm { when, then });
3557 }
3558 let otherwise = match otherwise {
3559 Some(expr) => Some(self.over_aggregate(expr, scope)?),
3560 None => None,
3561 };
3562 let arms = self.plan.add_arms(&rewritten);
3563 Ok(self.plan.add_expr(Expr::Case { arms, otherwise }, ty))
3564 }
3565 }
3566 }
3567
3568 pub(crate) fn same_expr(&self, left: ExprRef, right: ExprRef) -> bool {
3570 same_expr(&self.plan, left, right)
3571 }
3572}
3573
3574#[derive(Debug, Default)]
3584struct Options {
3585 binary_as_string: bool,
3588 all_varchar: bool,
3590 file_row_number: bool,
3595 given: Given,
3597}
3598
3599impl Options {
3600 fn of(written: &[(&'static str, Value, ExprRef)]) -> Result<Self> {
3607 let mut options = Self::default();
3608 for (parameter, value, _) in written {
3609 match (*parameter, value) {
3610 ("binary_as_string", Value::Boolean(on)) => options.binary_as_string = *on,
3611 ("all_varchar", Value::Boolean(on)) => options.all_varchar = *on,
3612 ("file_row_number", Value::Boolean(on)) => options.file_row_number = *on,
3613 _ => {}
3614 }
3615 }
3616 let named: Vec<(&str, Value)> =
3617 written.iter().map(|(parameter, value, _)| (*parameter, value.clone())).collect();
3618 options.given = csv_given(&named)?;
3619 Ok(options)
3620 }
3621}
3622
3623fn mirror_target(paths: &[String]) -> Option<(String, FileStamp)> {
3626 let [path] = paths else { return None };
3627 let canonical = std::fs::canonicalize(path).ok()?;
3628 let stamp = FileStamp::of(&canonical)?;
3629 Some((canonical.to_str()?.to_string(), stamp))
3630}
3631
3632#[derive(Clone, Copy)]
3634struct Operator {
3635 op: SetOp,
3637 quantifier: Quantifier,
3639 by_name: bool,
3641}
3642
3643struct Merged {
3645 name: String,
3647 ty: LogicalType,
3649 left: Option<usize>,
3651 right: Option<usize>,
3653}
3654
3655fn match_by_position(left: &Scope, right: &Scope) -> Result<Vec<Merged>> {
3659 if left.len() != right.len() {
3660 return Err(Error::binder(format!(
3661 "Set operations can only apply to expressions with the same number of result columns, but left side has {} and right side has {}",
3662 left.len(),
3663 right.len()
3664 )));
3665 }
3666 let mut merged = Vec::with_capacity(left.len());
3667 for (at, (held, other)) in left.columns.iter().zip(&right.columns).enumerate() {
3668 merged.push(Merged {
3669 name: held.name.clone(),
3670 ty: meet(&held.ty, &other.ty)?,
3671 left: Some(at),
3672 right: Some(at),
3673 });
3674 }
3675 Ok(merged)
3676}
3677
3678fn match_by_name(left: &Scope, right: &Scope) -> Result<Vec<Merged>> {
3686 named_once(left)?;
3687 named_once(right)?;
3688 let mut merged = Vec::with_capacity(left.len() + right.len());
3689 for (at, held) in left.columns.iter().enumerate() {
3690 let other = right.columns.iter().position(|column| same_name(&column.name, &held.name));
3691 let ty = match other {
3692 Some(other) => meet(&held.ty, &right.columns[other].ty)?,
3693 None => held.ty.clone(),
3694 };
3695 merged.push(Merged { name: held.name.clone(), ty, left: Some(at), right: other });
3696 }
3697 for (at, held) in right.columns.iter().enumerate() {
3698 if left.columns.iter().any(|column| same_name(&column.name, &held.name)) {
3699 continue;
3700 }
3701 merged.push(Merged {
3702 name: held.name.clone(),
3703 ty: held.ty.clone(),
3704 left: None,
3705 right: Some(at),
3706 });
3707 }
3708 Ok(merged)
3709}
3710
3711fn named_once(scope: &Scope) -> Result<()> {
3717 for (at, held) in scope.columns.iter().enumerate() {
3718 if scope.columns[..at].iter().any(|column| same_name(&column.name, &held.name)) {
3719 return Err(Error::binder(format!(
3720 "UNION (ALL) BY NAME operation doesn't support duplicate names in the SELECT list - the name \"\"{}\"\" occurs multiple times",
3721 held.name
3722 )));
3723 }
3724 }
3725 Ok(())
3726}
3727
3728fn meet(left: &LogicalType, right: &LogicalType) -> Result<LogicalType> {
3730 left.promote(right).ok_or_else(|| {
3731 Error::binder(format!(
3732 "Cannot combine a column of type {left} with a column of type {right} in a set operation"
3733 ))
3734 })
3735}
3736
3737fn null_parameter(function: TableFunction, parameter: &str) -> String {
3746 match parameter {
3747 "header" => format!("\"{parameter}\" expects a non-null boolean value (e.g. TRUE or 1)"),
3748 "all_varchar" => format!("{} \"{parameter}\" cannot be NULL", function.name()),
3749 _ => format!("Cannot use NULL as argument to \"{parameter}\""),
3750 }
3751}
3752
3753fn missing_replacement(name: &str, input: &Scope) -> Error {
3758 Error::binder(format!(
3759 "Column \"{name}\" in REPLACE list not found in FROM clause{}",
3760 input.candidates()
3761 ))
3762}
3763
3764fn subtractable(ty: &LogicalType, ordering: bool) -> bool {
3773 if ty.is_numeric() {
3774 return true;
3775 }
3776 match ty {
3777 LogicalType::Date
3778 | LogicalType::Time
3779 | LogicalType::Timestamp
3780 | LogicalType::TimestampS
3781 | LogicalType::TimestampMs
3782 | LogicalType::TimestampNs
3783 | LogicalType::TimestampTz => true,
3784 LogicalType::TimeTz => ordering,
3785 _ => false,
3786 }
3787}
3788
3789fn refuse_fill(
3798 argument: &LogicalType,
3799 order: &[LogicalType],
3800 distinct: bool,
3801 ignore_nulls: bool,
3802) -> Result<()> {
3803 if !subtractable(argument, false) {
3804 return Err(Error::binder("FILL argument must support subtraction"));
3805 }
3806 let [key] = order else {
3807 return Err(Error::binder("FILL functions must have only one ORDER BY expression"));
3808 };
3809 if !subtractable(key, true) {
3810 return Err(Error::binder("FILL ordering must support subtraction"));
3811 }
3812 if distinct {
3813 return Err(Error::binder(
3814 "DISTINCT is not implemented for the window function \"\"fill\"\"",
3815 ));
3816 }
3817 if ignore_nulls {
3818 return Err(Error::binder(
3819 "RESPECT/IGNORE NULLS is not supported for the window function \"fill\"",
3820 ));
3821 }
3822 Ok(())
3823}
3824
3825fn window_signature(name: &str, types: &[LogicalType]) -> Result<Resolved> {
3832 match kind_of(name) {
3833 Some(FunctionKind::Aggregate | FunctionKind::Window) => resolve(name, types),
3834 Some(FunctionKind::Scalar) => {
3835 Err(Error::catalog(format!("{name} is not an aggregate function")))
3836 }
3837 None => Err(Error::catalog(format!("Aggregate Function with name {name} does not exist!"))),
3838 }
3839}
3840
3841fn same_expr(plan: &Plan, left: ExprRef, right: ExprRef) -> bool {
3843 if left == right {
3844 return true;
3845 }
3846 if plan.expr_type(left) != plan.expr_type(right) {
3847 return false;
3848 }
3849 let lists = |left, right| {
3850 let left: &[ExprRef] = plan.expr_list(left);
3851 let right: &[ExprRef] = plan.expr_list(right);
3852 left.len() == right.len()
3853 && left.iter().zip(right).all(|(&left, &right)| same_expr(plan, left, right))
3854 };
3855 match (plan.expr(left), plan.expr(right)) {
3856 (Expr::Column(left), Expr::Column(right)) => left == right,
3857 (Expr::Constant(left), Expr::Constant(right)) => plan.value(*left) == plan.value(*right),
3858 (
3859 Expr::Cast { input: left, try_cast: left_try },
3860 Expr::Cast { input: right, try_cast: right_try },
3861 ) => left_try == right_try && same_expr(plan, *left, *right),
3862 (
3863 Expr::Compare { op: left_op, left: left_a, right: left_b },
3864 Expr::Compare { op: right_op, left: right_a, right: right_b },
3865 ) => {
3866 left_op == right_op
3867 && same_expr(plan, *left_a, *right_a)
3868 && same_expr(plan, *left_b, *right_b)
3869 }
3870 (
3871 Expr::Conjunction { op: left_op, children: left_children },
3872 Expr::Conjunction { op: right_op, children: right_children },
3873 ) => left_op == right_op && lists(*left_children, *right_children),
3874 (
3875 Expr::Function { name: left_name, args: left_args },
3876 Expr::Function { name: right_name, args: right_args },
3877 ) => plan.string(*left_name) == plan.string(*right_name) && lists(*left_args, *right_args),
3878 (
3879 Expr::Aggregate {
3880 name: left_name,
3881 args: left_args,
3882 distinct: left_distinct,
3883 filter: left_filter,
3884 },
3885 Expr::Aggregate {
3886 name: right_name,
3887 args: right_args,
3888 distinct: right_distinct,
3889 filter: right_filter,
3890 },
3891 ) => {
3892 plan.string(*left_name) == plan.string(*right_name)
3893 && left_distinct == right_distinct
3894 && match (left_filter, right_filter) {
3895 (None, None) => true,
3896 (Some(left), Some(right)) => same_expr(plan, *left, *right),
3897 _ => false,
3898 }
3899 && lists(*left_args, *right_args)
3900 }
3901 (
3905 Expr::Window {
3906 name: left_name,
3907 args: left_args,
3908 distinct: left_distinct,
3909 filter: left_filter,
3910 ignore_nulls: left_nulls,
3911 order: left_order,
3912 },
3913 Expr::Window {
3914 name: right_name,
3915 args: right_args,
3916 distinct: right_distinct,
3917 filter: right_filter,
3918 ignore_nulls: right_nulls,
3919 order: right_order,
3920 },
3921 ) => {
3922 let left_keys = plan.sort_key_list(*left_order);
3925 let right_keys = plan.sort_key_list(*right_order);
3926 plan.string(*left_name) == plan.string(*right_name)
3927 && left_distinct == right_distinct
3928 && left_nulls == right_nulls
3929 && left_keys.len() == right_keys.len()
3930 && left_keys.iter().zip(right_keys).all(|(left, right)| {
3931 left.descending == right.descending
3932 && left.nulls_first == right.nulls_first
3933 && same_expr(plan, left.expr, right.expr)
3934 })
3935 && match (left_filter, right_filter) {
3936 (None, None) => true,
3937 (Some(left), Some(right)) => same_expr(plan, *left, *right),
3938 _ => false,
3939 }
3940 && lists(*left_args, *right_args)
3941 }
3942 (
3943 Expr::Case { arms: left_arms, otherwise: left_otherwise },
3944 Expr::Case { arms: right_arms, otherwise: right_otherwise },
3945 ) => {
3946 let left_arms = plan.arm_list(*left_arms);
3947 let right_arms = plan.arm_list(*right_arms);
3948 left_arms.len() == right_arms.len()
3949 && left_arms.iter().zip(right_arms).all(|(left, right)| {
3950 same_expr(plan, left.when, right.when) && same_expr(plan, left.then, right.then)
3951 })
3952 && match (left_otherwise, right_otherwise) {
3953 (None, None) => true,
3954 (Some(left), Some(right)) => same_expr(plan, *left, *right),
3955 _ => false,
3956 }
3957 }
3958 _ => false,
3959 }
3960}