1use rudb_catalog::{Catalog, same_name};
16use rudb_common::{Error, Field, LogicalType, Result, Value};
17use rudb_functions::{
18 Columns, TableFunction, csv_fields, files, is_file, is_pattern, parquet_fields, resolve,
19 resolve_table,
20};
21use rudb_parse::ast::{self, Ast, Distinct, LiteralKind, Nulls, Order, Quantifier, SetOp};
22use rudb_parse::{NONE, parse_ast};
23use rudb_plan::{ColumnBinding, Expr, ExprRef, JoinKind, Node, NodeRef, Plan, SetOpKind, SortKey};
24
25use crate::expr::{describe, has_aggregate};
26use crate::scope::{Scope, Visible};
27
28pub fn bind(ast: &Ast, catalog: &Catalog) -> Result<Plan> {
35 let query = match ast.statements.as_slice() {
36 [ast::Statement::Query(query)] => *query,
37 [] => return Err(Error::binder("no statement to bind")),
38 _ => return Err(Error::not_implemented("a script of more than one statement")),
39 };
40 let mut binder = Binder::new(catalog);
41 let (root, _) = binder.bind_query(ast, query)?;
42 let mut plan = binder.into_plan();
43 plan.set_root(root);
44 plan.validate()?;
45 Ok(plan)
46}
47
48pub fn bind_sql(query: &str, catalog: &Catalog) -> Result<Plan> {
54 let ast = parse_ast(query)?;
55 bind(&ast, catalog)
56}
57
58#[derive(Debug)]
60pub(crate) struct Aggregation {
61 pub(crate) index: u32,
63 pub(crate) groups: Vec<ExprRef>,
65 pub(crate) aggregates: Vec<ExprRef>,
67}
68
69#[derive(Debug)]
71pub(crate) struct Binder<'a> {
72 catalog: &'a Catalog,
73 plan: Plan,
74 next_index: u32,
75 pub(crate) aggregation: Option<Aggregation>,
77 pub(crate) in_aggregate: bool,
79 pub(crate) clause: &'static str,
81}
82
83impl<'a> Binder<'a> {
84 pub(crate) fn new(catalog: &'a Catalog) -> Self {
85 Self {
86 catalog,
87 plan: Plan::new(),
88 next_index: 0,
89 aggregation: None,
90 in_aggregate: false,
91 clause: "SELECT clause",
92 }
93 }
94
95 pub(crate) fn plan(&self) -> &Plan {
96 &self.plan
97 }
98
99 pub(crate) fn plan_mut(&mut self) -> &mut Plan {
100 &mut self.plan
101 }
102
103 pub(crate) fn into_plan(self) -> Plan {
104 self.plan
105 }
106
107 pub(crate) fn fresh_index(&mut self) -> u32 {
109 let index = self.next_index;
110 self.next_index += 1;
111 index
112 }
113
114 fn column(&mut self, index: u32, position: usize, ty: LogicalType) -> ExprRef {
116 let binding = ColumnBinding::new(index, position as u32);
117 self.plan.add_expr(Expr::Column(binding), ty)
118 }
119
120 pub(crate) fn bind_query(
123 &mut self,
124 ast: &Ast,
125 query: ast::QueryRef,
126 ) -> Result<(NodeRef, Scope)> {
127 let written = ast.query(query);
128 match written.body {
129 ast::QueryBody::Select(select) => self.bind_select(ast, select, &written),
130 ast::QueryBody::SetOp { op, quantifier, by_name, left, right } => {
131 if by_name {
132 return Err(Error::not_implemented("UNION BY NAME"));
133 }
134 self.bind_set_op(ast, &written, op, quantifier, left, right)
135 }
136 ast::QueryBody::Values(rows) => self.bind_values(ast, &written, rows),
137 }
138 }
139
140 fn bind_values(
147 &mut self,
148 ast: &Ast,
149 query: &ast::Query,
150 rows: ast::Slice,
151 ) -> Result<(NodeRef, Scope)> {
152 let written = ast.rows(rows).to_vec();
153 let Some(first) = written.first() else {
154 return Err(Error::binder("VALUES needs at least one row"));
155 };
156 let width = first.len as usize;
157 for (at, row) in written.iter().enumerate() {
158 if row.len as usize != width {
159 return Err(Error::binder(format!(
160 "VALUES lists must all be the same length, expected {width} columns but row {} has {}",
161 at + 1,
162 row.len
163 )));
164 }
165 }
166 let empty = Scope::empty();
168 let previous = std::mem::replace(&mut self.clause, "VALUES clause");
169 let mut bound: Vec<Vec<ExprRef>> = Vec::with_capacity(written.len());
170 for row in &written {
171 let mut items = Vec::with_capacity(width);
172 for &expr in ast.expr_list(*row) {
173 items.push(self.bind_expr(ast, expr, &empty)?);
174 }
175 bound.push(items);
176 }
177 self.clause = previous;
178 let mut types = Vec::with_capacity(width);
179 for at in 0..width {
180 let mut ty = self.plan.expr_type(bound[0][at]).clone();
181 for row in &bound[1..] {
182 let other = self.plan.expr_type(row[at]).clone();
183 ty = ty.promote(&other).ok_or_else(|| {
184 Error::binder(format!(
185 "Cannot combine a value of type {ty} with a value of type {other} in column {} of a VALUES",
186 at + 1
187 ))
188 })?;
189 }
190 types.push(ty);
191 }
192 let mut slices = Vec::with_capacity(bound.len());
193 for row in &bound {
194 let items: Vec<ExprRef> =
195 row.iter().zip(&types).map(|(&expr, ty)| self.cast_to(expr, ty)).collect();
196 slices.push(self.plan.add_expr_list(&items));
197 }
198 let rows = self.plan.add_rows(&slices);
199 let fields: Vec<Field> = types
200 .iter()
201 .enumerate()
202 .map(|(at, ty)| Field::new(format!("col{at}"), ty.clone()))
203 .collect();
204 let columns = self.plan.add_fields(&fields);
205 let index = self.fresh_index();
206 let mut node = self.plan.add_node(Node::Values { index, columns, rows });
207 let mut scope = Scope::empty();
208 for (at, field) in fields.iter().enumerate() {
209 scope.push(Visible {
210 table: String::new(),
211 name: field.name.clone(),
212 binding: ColumnBinding::new(index, at as u32),
213 ty: field.ty.clone(),
214 });
215 }
216 let keys = self.sort_keys(ast, query, &scope, &[])?;
217 if !keys.is_empty() {
218 let keys = self.plan.add_sort_keys(&keys);
219 node = self.plan.add_node(Node::Sort { input: node, keys });
220 }
221 node = self.apply_limit(ast, query, node)?;
222 Ok((node, scope))
223 }
224
225 fn bind_set_op(
226 &mut self,
227 ast: &Ast,
228 query: &ast::Query,
229 op: SetOp,
230 quantifier: Quantifier,
231 left: ast::QueryRef,
232 right: ast::QueryRef,
233 ) -> Result<(NodeRef, Scope)> {
234 let (left_node, left_scope) = self.bind_query(ast, left)?;
235 let (right_node, right_scope) = self.bind_query(ast, right)?;
236 if left_scope.len() != right_scope.len() {
237 return Err(Error::binder(format!(
238 "Set operations can only apply to expressions with the same number of result columns, but left side has {} and right side has {}",
239 left_scope.len(),
240 right_scope.len()
241 )));
242 }
243 let mut types = Vec::with_capacity(left_scope.len());
245 for (left, right) in left_scope.columns.iter().zip(&right_scope.columns) {
246 let common = left.ty.promote(&right.ty).ok_or_else(|| {
247 Error::binder(format!(
248 "Cannot combine a column of type {} with a column of type {} in a set operation",
249 left.ty, right.ty
250 ))
251 })?;
252 types.push(common);
253 }
254 let left_node = self.conform(left_node, &left_scope, &types);
255 let right_node = self.conform(right_node, &right_scope, &types);
256 let index = self.fresh_index();
257 let kind = match op {
258 SetOp::Union => SetOpKind::Union,
259 SetOp::Except => SetOpKind::Except,
260 SetOp::Intersect => SetOpKind::Intersect,
261 };
262 let all = quantifier == Quantifier::All;
265 let mut node = self.plan.add_node(Node::SetOp {
266 left: left_node,
267 right: right_node,
268 kind,
269 all,
270 index,
271 });
272 let mut scope = Scope::empty();
273 for (at, (column, ty)) in left_scope.columns.iter().zip(&types).enumerate() {
274 scope.push(Visible {
275 table: String::new(),
276 name: column.name.clone(),
277 binding: ColumnBinding::new(index, at as u32),
278 ty: ty.clone(),
279 });
280 }
281 let keys = self.sort_keys(ast, query, &scope, &[])?;
285 if !keys.is_empty() {
286 let keys = self.plan.add_sort_keys(&keys);
287 node = self.plan.add_node(Node::Sort { input: node, keys });
288 }
289 node = self.apply_limit(ast, query, node)?;
290 Ok((node, scope))
291 }
292
293 fn conform(&mut self, node: NodeRef, scope: &Scope, types: &[LogicalType]) -> NodeRef {
295 if scope.columns.iter().zip(types).all(|(column, ty)| &column.ty == ty) {
296 return node;
297 }
298 let index = self.fresh_index();
299 let mut exprs = Vec::with_capacity(types.len());
300 let mut names = Vec::with_capacity(types.len());
301 for (column, ty) in scope.columns.iter().zip(types) {
302 let expr = self.plan.add_expr(Expr::Column(column.binding), column.ty.clone());
303 exprs.push(self.cast_to(expr, ty));
304 names.push(self.plan.intern(&column.name));
305 }
306 let exprs = self.plan.add_expr_list(&exprs);
307 let names = self.plan.add_name_list(&names);
308 self.plan.add_node(Node::Project { input: node, index, exprs, names })
309 }
310
311 fn bind_select(
314 &mut self,
315 ast: &Ast,
316 select: ast::SelectRef,
317 query: &ast::Query,
318 ) -> Result<(NodeRef, Scope)> {
319 let written = ast.select(select);
320 let (mut node, input) = self.bind_from(ast, written.from)?;
321
322 if written.filter != NONE {
323 self.clause = "WHERE clause";
324 let predicate = self.bind_expr(ast, written.filter, &input)?;
325 let predicate = self.as_boolean(predicate, "WHERE")?;
326 node = self.plan.add_node(Node::Filter { input: node, predicate });
327 }
328
329 let targets = ast.target_list(written.targets).to_vec();
330 if targets.is_empty() {
331 return Err(Error::binder("a SELECT needs at least one expression to select"));
332 }
333
334 let group_items = self.group_items(ast, &written, &targets)?;
335 let aggregating = !group_items.is_empty()
336 || written.having != NONE
337 || targets.iter().any(|target| has_aggregate(ast, target.expr));
338 if aggregating {
339 self.clause = "GROUP BY clause";
340 let mut groups = Vec::with_capacity(group_items.len());
341 for item in &group_items {
342 groups.push(self.bind_expr(ast, *item, &input)?);
343 }
344 let index = self.fresh_index();
345 self.aggregation = Some(Aggregation { index, groups, aggregates: Vec::new() });
346 }
347
348 self.clause = "SELECT clause";
349 let (mut exprs, mut names) = self.bind_targets(ast, &targets, &input)?;
350 let visible = exprs.len();
351
352 let mut having = None;
353 if written.having != NONE {
354 self.clause = "HAVING clause";
355 let predicate = self.bind_expr(ast, written.having, &input)?;
356 let predicate = self.over_aggregate(predicate, &input)?;
357 having = Some(self.as_boolean(predicate, "HAVING")?);
358 }
359
360 let project = self.fresh_index();
363 let mut output = Scope::empty();
364 for (at, (expr, name)) in exprs.iter().zip(&names).enumerate() {
365 output.push(Visible {
366 table: String::new(),
367 name: name.clone(),
368 binding: ColumnBinding::new(project, at as u32),
369 ty: self.plan.expr_type(*expr).clone(),
370 });
371 }
372
373 self.clause = "ORDER BY clause";
374 let mut extra = Vec::new();
375 let keys = self.select_sort_keys(
376 ast, query, &input, &output, project, &mut exprs, &mut names, &mut extra,
377 )?;
378 if !extra.is_empty() && written.distinct != Distinct::No {
379 return Err(Error::binder(
380 "For SELECT DISTINCT, ORDER BY expressions must appear in the select list",
381 ));
382 }
383 let on = self.distinct_on(ast, written.distinct, &output)?;
384
385 if let Some(aggregation) = self.aggregation.take() {
386 let index = aggregation.index;
387 let groups = self.plan.add_expr_list(&aggregation.groups);
388 let aggregates = self.plan.add_expr_list(&aggregation.aggregates);
389 node = self.plan.add_node(Node::Aggregate { input: node, index, groups, aggregates });
390 }
391 if let Some(predicate) = having {
392 node = self.plan.add_node(Node::Filter { input: node, predicate });
393 }
394
395 let interned: Vec<u32> = names.iter().map(|name| self.plan.intern(name)).collect();
396 let exprs_slice = self.plan.add_expr_list(&exprs);
397 let names_slice = self.plan.add_name_list(&interned);
398 node = self.plan.add_node(Node::Project {
399 input: node,
400 index: project,
401 exprs: exprs_slice,
402 names: names_slice,
403 });
404
405 if written.distinct != Distinct::No {
406 let on = self.plan.add_expr_list(&on);
407 node = self.plan.add_node(Node::Distinct { input: node, on });
408 }
409 if !keys.is_empty() {
410 let keys = self.plan.add_sort_keys(&keys);
411 node = self.plan.add_node(Node::Sort { input: node, keys });
412 }
413 node = self.apply_limit(ast, query, node)?;
414
415 if extra.is_empty() {
416 output.columns.truncate(visible);
417 return Ok((node, output));
418 }
419 let index = self.fresh_index();
422 let mut kept = Vec::with_capacity(visible);
423 let mut kept_names = Vec::with_capacity(visible);
424 let mut scope = Scope::empty();
425 for (at, name) in names.iter().enumerate().take(visible) {
426 let ty = output.columns[at].ty.clone();
427 kept.push(self.column(project, at, ty.clone()));
428 kept_names.push(self.plan.intern(name));
429 scope.push(Visible {
430 table: String::new(),
431 name: name.clone(),
432 binding: ColumnBinding::new(index, at as u32),
433 ty,
434 });
435 }
436 let exprs = self.plan.add_expr_list(&kept);
437 let names = self.plan.add_name_list(&kept_names);
438 node = self.plan.add_node(Node::Project { input: node, index, exprs, names });
439 Ok((node, scope))
440 }
441
442 fn bind_targets(
444 &mut self,
445 ast: &Ast,
446 targets: &[ast::Target],
447 input: &Scope,
448 ) -> Result<(Vec<ExprRef>, Vec<String>)> {
449 let mut exprs = Vec::with_capacity(targets.len());
450 let mut names = Vec::with_capacity(targets.len());
451 for target in targets {
452 if let ast::Expr::Star { qualifier } = ast.expr(target.expr) {
453 let table = ast.name(qualifier).last().map(str::to_string);
454 let expanded: Vec<Visible> =
455 input.star(table.as_deref())?.into_iter().cloned().collect();
456 for column in expanded {
457 let expr = self.plan.add_expr(Expr::Column(column.binding), column.ty);
458 exprs.push(self.over_aggregate(expr, input)?);
459 names.push(column.name);
460 }
461 continue;
462 }
463 let expr = self.bind_expr(ast, target.expr, input)?;
464 exprs.push(self.over_aggregate(expr, input)?);
465 names.push(if target.alias == NONE {
466 self.output_name(ast, target.expr, input)
467 } else {
468 ast.string(target.alias).to_string()
469 });
470 }
471 Ok((exprs, names))
472 }
473
474 fn output_name(&self, ast: &Ast, target: ast::ExprRef, input: &Scope) -> String {
480 if let ast::Expr::Column { name } = ast.expr(target) {
481 let parts: Vec<&str> = ast.name(name).collect();
482 if let Ok(found) = input.resolve(&parts) {
483 return found.name.clone();
484 }
485 }
486 describe(ast, target)
487 }
488
489 fn group_items(
491 &self,
492 ast: &Ast,
493 select: &ast::Select,
494 targets: &[ast::Target],
495 ) -> Result<Vec<ast::ExprRef>> {
496 if select.group_by_all {
497 return Ok(targets
500 .iter()
501 .filter(|target| !has_aggregate(ast, target.expr))
502 .map(|target| target.expr)
503 .collect());
504 }
505 let mut items = Vec::new();
506 for &item in ast.expr_list(select.group_by) {
507 items.push(self.output_reference(ast, item, targets, "GROUP BY")?.unwrap_or(item));
508 }
509 Ok(items)
510 }
511
512 fn output_reference(
514 &self,
515 ast: &Ast,
516 item: ast::ExprRef,
517 targets: &[ast::Target],
518 clause: &str,
519 ) -> Result<Option<ast::ExprRef>> {
520 match ast.expr(item) {
521 ast::Expr::Literal { kind: LiteralKind::Number, text } => {
522 let written = ast.string(text);
523 let position: usize = written.parse().map_err(|_| {
524 Error::binder(format!("{clause} term {written} is not a column"))
525 })?;
526 if position == 0 || position > targets.len() {
527 return Err(Error::binder(format!(
528 "{clause} term out of range - should be between 1 and {}",
529 targets.len()
530 )));
531 }
532 Ok(Some(targets[position - 1].expr))
533 }
534 ast::Expr::Column { name } => {
535 let parts: Vec<&str> = ast.name(name).collect();
536 let [written] = parts.as_slice() else { return Ok(None) };
537 let mut found = None;
538 for target in targets {
539 if target.alias != NONE && same_name(ast.string(target.alias), written) {
540 if found.is_some() {
541 return Ok(None);
542 }
543 found = Some(target.expr);
544 }
545 }
546 Ok(found)
547 }
548 _ => Ok(None),
549 }
550 }
551
552 #[allow(clippy::too_many_arguments)]
556 fn select_sort_keys(
557 &mut self,
558 ast: &Ast,
559 query: &ast::Query,
560 input: &Scope,
561 output: &Scope,
562 project: u32,
563 exprs: &mut Vec<ExprRef>,
564 names: &mut Vec<String>,
565 extra: &mut Vec<usize>,
566 ) -> Result<Vec<SortKey>> {
567 if query.order_by_all {
568 return Ok(self.every_column(output));
569 }
570 let items = ast.order_list(query.order_by).to_vec();
571 let mut keys = Vec::with_capacity(items.len());
572 for item in items {
573 let position = match self.output_position(ast, item.expr, output)? {
574 Some(position) => position,
575 None => {
576 let bound = self.bind_expr(ast, item.expr, input)?;
577 let bound = self.over_aggregate(bound, input)?;
578 match exprs.iter().position(|&held| self.same_expr(held, bound)) {
579 Some(position) => position,
580 None => {
581 exprs.push(bound);
582 names.push(describe(ast, item.expr));
583 extra.push(exprs.len() - 1);
584 exprs.len() - 1
585 }
586 }
587 }
588 };
589 let ty = self.plan.expr_type(exprs[position]).clone();
590 let expr = self.column(project, position, ty);
591 keys.push(sort_key(expr, item));
592 }
593 Ok(keys)
594 }
595
596 fn sort_keys(
598 &mut self,
599 ast: &Ast,
600 query: &ast::Query,
601 output: &Scope,
602 targets: &[ast::Target],
603 ) -> Result<Vec<SortKey>> {
604 if query.order_by_all {
605 return Ok(self.every_column(output));
606 }
607 let items = ast.order_list(query.order_by).to_vec();
608 let mut keys = Vec::with_capacity(items.len());
609 for item in items {
610 let expr = match self.output_position(ast, item.expr, output)? {
611 Some(position) => {
612 let column = &output.columns[position];
613 let (binding, ty) = (column.binding, column.ty.clone());
614 self.plan.add_expr(Expr::Column(binding), ty)
615 }
616 None => {
617 let _ = targets;
618 self.bind_expr(ast, item.expr, output)?
619 }
620 };
621 keys.push(sort_key(expr, item));
622 }
623 Ok(keys)
624 }
625
626 fn every_column(&mut self, output: &Scope) -> Vec<SortKey> {
627 let columns: Vec<(ColumnBinding, LogicalType)> =
628 output.columns.iter().map(|column| (column.binding, column.ty.clone())).collect();
629 columns
630 .into_iter()
631 .map(|(binding, ty)| {
632 let expr = self.plan.add_expr(Expr::Column(binding), ty);
633 SortKey { expr, descending: false, nulls_first: false }
634 })
635 .collect()
636 }
637
638 fn output_position(
640 &self,
641 ast: &Ast,
642 item: ast::ExprRef,
643 output: &Scope,
644 ) -> Result<Option<usize>> {
645 match ast.expr(item) {
646 ast::Expr::Literal { kind: LiteralKind::Number, text } => {
647 let written = ast.string(text);
648 if written.contains(['.', 'e', 'E']) {
649 return Ok(None);
650 }
651 let position: usize = written.parse().map_err(|_| {
652 Error::binder(format!("ORDER BY term {written} is not a column"))
653 })?;
654 if position == 0 || position > output.len() {
655 return Err(Error::binder(format!(
656 "ORDER BY term out of range - should be between 1 and {}",
657 output.len()
658 )));
659 }
660 Ok(Some(position - 1))
661 }
662 ast::Expr::Column { name } => {
663 let parts: Vec<&str> = ast.name(name).collect();
664 let [written] = parts.as_slice() else { return Ok(None) };
665 Ok(output.position_of(None, written))
666 }
667 _ => Ok(None),
668 }
669 }
670
671 fn distinct_on(
673 &mut self,
674 ast: &Ast,
675 distinct: Distinct,
676 output: &Scope,
677 ) -> Result<Vec<ExprRef>> {
678 let Distinct::On(items) = distinct else {
679 return Ok(Vec::new());
680 };
681 let items = ast.expr_list(items).to_vec();
682 let mut on = Vec::with_capacity(items.len());
683 for item in items {
684 let Some(position) = self.output_position(ast, item, output)? else {
685 return Err(Error::not_implemented(
686 "DISTINCT ON an expression that is not in the select list",
687 ));
688 };
689 let column = &output.columns[position];
690 let (binding, ty) = (column.binding, column.ty.clone());
691 on.push(self.plan.add_expr(Expr::Column(binding), ty));
692 }
693 Ok(on)
694 }
695
696 fn apply_limit(&mut self, ast: &Ast, query: &ast::Query, input: NodeRef) -> Result<NodeRef> {
697 if query.limit_percent {
698 return Err(Error::not_implemented("LIMIT with a percentage"));
699 }
700 let count = self.constant_count(ast, query.limit, "LIMIT")?;
701 let offset = self.constant_count(ast, query.offset, "OFFSET")?.unwrap_or(0);
702 if count.is_none() && offset == 0 {
703 return Ok(input);
704 }
705 Ok(self.plan.add_node(Node::Limit { input, count, offset }))
706 }
707
708 fn constant_count(
710 &mut self,
711 ast: &Ast,
712 written: ast::ExprRef,
713 clause: &str,
714 ) -> Result<Option<u64>> {
715 if written == NONE {
716 return Ok(None);
717 }
718 self.clause = "LIMIT clause";
719 let scope = Scope::empty();
720 let bound = self.bind_expr(ast, written, &scope)?;
721 let Expr::Constant(value) = *self.plan.expr(bound) else {
722 return Err(Error::not_implemented(format!("a {clause} that is not a constant")));
723 };
724 let count = match self.plan.value(value) {
725 Value::Null => return Ok(None),
726 Value::TinyInt(count) => i128::from(*count),
727 Value::SmallInt(count) => i128::from(*count),
728 Value::Integer(count) => i128::from(*count),
729 Value::BigInt(count) => i128::from(*count),
730 Value::HugeInt(count) => *count,
731 other => {
732 return Err(Error::binder(format!(
733 "{clause} takes a whole number of rows, not a value of type {}",
734 other.logical_type()
735 )));
736 }
737 };
738 u64::try_from(count)
739 .map(Some)
740 .map_err(|_| Error::binder(format!("{clause} must not be negative")))
741 }
742
743 fn bind_from(&mut self, ast: &Ast, from: ast::Slice) -> Result<(NodeRef, Scope)> {
746 let sources = ast.source_list(from).to_vec();
747 let Some((first, rest)) = sources.split_first() else {
748 return Ok((self.plan.add_node(Node::Dummy), Scope::empty()));
751 };
752 let (mut node, mut scope) = self.bind_source(ast, *first)?;
753 for source in rest {
754 let (right, right_scope) = self.bind_source(ast, *source)?;
755 node = self.plan.add_node(Node::CrossProduct { left: node, right });
756 scope = scope.concat(right_scope);
757 }
758 Ok((node, scope))
759 }
760
761 fn bind_source(&mut self, ast: &Ast, source: ast::SourceRef) -> Result<(NodeRef, Scope)> {
762 match ast.source(source) {
763 ast::Source::Table { name, alias, columns } => {
764 self.bind_table(ast, name, alias, columns)
765 }
766 ast::Source::Function { name, args, alias, columns } => {
767 self.bind_table_function(ast, name, args, alias, columns)
768 }
769 ast::Source::Subquery { query, alias, columns } => {
770 let (node, mut scope) = self.bind_query(ast, query)?;
771 let label = if alias == NONE {
772 "unnamed_subquery".to_string()
773 } else {
774 ast.string(alias).to_string()
775 };
776 scope.relabel(&label);
777 if !columns.is_empty() {
778 let names: Vec<&str> = ast.name(columns).collect();
779 scope.rename(&names, &label)?;
780 }
781 Ok((node, scope))
782 }
783 ast::Source::Values { rows, alias, columns } => {
784 let bare = ast::Query::bare(ast::QueryBody::Values(rows));
785 let (node, mut scope) = self.bind_values(ast, &bare, rows)?;
786 let label =
787 if alias == NONE { String::new() } else { ast.string(alias).to_string() };
788 scope.relabel(&label);
789 if !columns.is_empty() {
790 let names: Vec<&str> = ast.name(columns).collect();
791 scope.rename(&names, &label)?;
792 }
793 Ok((node, scope))
794 }
795 ast::Source::Join { left, right, kind, natural, on, using } => {
796 self.bind_join(ast, left, right, kind, natural, on, using)
797 }
798 }
799 }
800
801 fn bind_table(
802 &mut self,
803 ast: &Ast,
804 name: ast::Slice,
805 alias: ast::StrRef,
806 columns: ast::Slice,
807 ) -> Result<(NodeRef, Scope)> {
808 let parts: Vec<&str> = ast.name(name).collect();
809 let catalog = self.catalog;
810 let found = catalog.resolve(&parts).and_then(|resolved| {
813 let table = catalog.table(&resolved)?;
814 Ok((resolved, table))
815 });
816 let (resolved, table) = match found {
817 Ok(found) => found,
818 Err(missing) => {
819 return self.bind_replacement_scan(ast, &parts, alias, columns, missing);
820 }
821 };
822 let fields: Vec<Field> = table.columns().to_vec();
823 let label =
824 if alias == NONE { resolved.table.clone() } else { ast.string(alias).to_string() };
825 let index = self.fresh_index();
826 let mut scope = Scope::empty();
827 for (at, field) in fields.iter().enumerate() {
828 scope.push(Visible {
829 table: label.clone(),
830 name: field.name.clone(),
831 binding: ColumnBinding::new(index, at as u32),
832 ty: field.ty.clone(),
833 });
834 }
835 if !columns.is_empty() {
836 let names: Vec<&str> = ast.name(columns).collect();
837 scope.rename(&names, &label)?;
838 }
839 let catalog_name = self.plan.intern(&resolved.catalog);
840 let schema = self.plan.intern(&resolved.schema);
841 let table_name = self.plan.intern(&resolved.table);
842 let alias = self.plan.intern(&label);
843 let columns = self.plan.add_fields(&fields);
844 let node = self.plan.add_node(Node::Get {
845 catalog: catalog_name,
846 schema,
847 table: table_name,
848 alias,
849 index,
850 columns,
851 });
852 Ok((node, scope))
853 }
854
855 fn bind_table_function(
863 &mut self,
864 ast: &Ast,
865 name: ast::Slice,
866 args: ast::Slice,
867 alias: ast::StrRef,
868 columns: ast::Slice,
869 ) -> Result<(NodeRef, Scope)> {
870 let parts: Vec<&str> = ast.name(name).collect();
871 let function_name = *parts.last().unwrap_or(&"");
875 if let Some(schema) = parts.iter().rev().nth(1) {
876 if !schema.eq_ignore_ascii_case("main") && !schema.eq_ignore_ascii_case("system") {
877 return Err(Error::catalog(format!(
878 "Table Function with name {} does not exist!",
879 parts.join(".")
880 )));
881 }
882 }
883 if TableFunction::lookup(function_name).is_none() {
887 return Err(Error::catalog(format!(
888 "Table Function with name {function_name} does not exist!"
889 )));
890 }
891 let written = ast.expr_list(args).to_vec();
892 let empty = Scope::empty();
893 let previous = std::mem::replace(&mut self.clause, "table function arguments");
894 let mut bound = Vec::with_capacity(written.len());
895 for expr in written {
896 bound.push(self.bind_expr(ast, expr, &empty)?);
897 }
898 self.clause = previous;
899
900 let given: Vec<LogicalType> =
903 bound.iter().map(|&expr| self.plan.expr_type(expr).clone()).collect();
904 let resolved = resolve_table(function_name, &given)?;
905 let mut cast: Vec<ExprRef> = bound
906 .iter()
907 .zip(&resolved.arguments)
908 .map(|(&expr, ty)| self.cast_to(expr, ty))
909 .collect();
910
911 let fields = match resolved.columns {
912 Columns::Fixed(fields) => fields,
913 columns => {
914 let paths = files(&self.file_argument(cast[0])?)?;
919 let first = paths.first().map_or("", String::as_str);
920 let fields = match columns {
921 Columns::Csv => csv_fields(first)?,
922 _ => parquet_fields(first)?,
923 };
924 cast = paths.iter().map(|path| self.path_constant(path)).collect();
925 fields
926 }
927 };
928 let label = if alias == NONE {
929 resolved.function.name().to_string()
930 } else {
931 ast.string(alias).to_string()
932 };
933 let names: Vec<&str> = ast.name(columns).collect();
934 self.table_function_source(resolved.function, &cast, fields, &label, &names)
935 }
936
937 fn bind_replacement_scan(
948 &mut self,
949 ast: &Ast,
950 parts: &[&str],
951 alias: ast::StrRef,
952 columns: ast::Slice,
953 missing: Error,
954 ) -> Result<(NodeRef, Scope)> {
955 let [path] = parts else { return Err(missing) };
956 let path = *path;
957 let extension = path.rsplit_once('.').map(|(_, after)| after).unwrap_or_default();
958 let Some(function) = Self::reader_for(extension) else {
959 if is_file(path) {
960 return Err(Error::binder(format!(
965 "No extension found that is capable of reading the file \"{path}\"\n* If this \
966 file is a supported file format you can explicitly use the reader functions, \
967 such as read_csv, read_json or read_parquet"
968 )));
969 }
970 return Err(missing);
971 };
972 let paths = files(path)?;
977 let first = paths.first().map_or("", String::as_str);
978 let fields = match function {
979 TableFunction::ReadParquet => parquet_fields(first)?,
980 _ => csv_fields(first)?,
981 };
982 let label = if alias == NONE {
988 if is_pattern(path) {
989 path.to_string()
990 } else {
991 let file = path.rsplit_once('/').map_or(path, |(_, file)| file);
992 file.rsplit_once('.').map_or(file, |(stem, _)| stem).to_string()
993 }
994 } else {
995 ast.string(alias).to_string()
996 };
997 let arguments: Vec<ExprRef> = paths.iter().map(|path| self.path_constant(path)).collect();
998 let names: Vec<&str> = ast.name(columns).collect();
999 self.table_function_source(function, &arguments, fields, &label, &names)
1000 }
1001
1002 fn path_constant(&mut self, path: &str) -> ExprRef {
1004 let value = self.plan.add_value(Value::Varchar(path.to_string()));
1005 self.plan.add_expr(Expr::Constant(value), LogicalType::Varchar)
1006 }
1007
1008 fn reader_for(extension: &str) -> Option<TableFunction> {
1015 if extension.eq_ignore_ascii_case("parquet") {
1016 return Some(TableFunction::ReadParquet);
1017 }
1018 if extension.eq_ignore_ascii_case("csv") || extension.eq_ignore_ascii_case("tsv") {
1019 return Some(TableFunction::ReadCsv);
1020 }
1021 None
1022 }
1023
1024 fn table_function_source(
1029 &mut self,
1030 function: TableFunction,
1031 args: &[ExprRef],
1032 fields: Vec<Field>,
1033 label: &str,
1034 names: &[&str],
1035 ) -> Result<(NodeRef, Scope)> {
1036 let index = self.fresh_index();
1037 let mut scope = Scope::empty();
1038 for (at, field) in fields.iter().enumerate() {
1039 scope.push(Visible {
1040 table: label.to_string(),
1041 name: field.name.clone(),
1042 binding: ColumnBinding::new(index, at as u32),
1043 ty: field.ty.clone(),
1044 });
1045 }
1046 if !names.is_empty() {
1047 scope.rename(names, label)?;
1048 }
1049 let function = self.plan.intern(function.name());
1050 let args = self.plan.add_expr_list(args);
1051 let columns = self.plan.add_fields(&fields);
1052 let node = self.plan.add_node(Node::TableFunction { index, function, args, columns });
1053 Ok((node, scope))
1054 }
1055
1056 fn file_argument(&self, expr: ExprRef) -> Result<String> {
1065 let Expr::Constant(reference) = *self.plan.expr(expr) else {
1066 return Err(Error::not_implemented(
1067 "a table function file name that is not a constant",
1068 ));
1069 };
1070 match self.plan.value(reference) {
1071 Value::Varchar(path) => Ok(path.clone()),
1072 Value::Null => Err(Error::parser("read_parquet cannot take NULL list as parameter")),
1074 other => {
1075 Err(Error::internal(format!("a file name bound as VARCHAR arrived as {other}")))
1076 }
1077 }
1078 }
1079
1080 #[allow(clippy::too_many_arguments)]
1081 fn bind_join(
1082 &mut self,
1083 ast: &Ast,
1084 left: ast::SourceRef,
1085 right: ast::SourceRef,
1086 kind: ast::JoinKind,
1087 natural: bool,
1088 on: ast::ExprRef,
1089 using: ast::Slice,
1090 ) -> Result<(NodeRef, Scope)> {
1091 let (left_node, left_scope) = self.bind_source(ast, left)?;
1092 let (right_node, right_scope) = self.bind_source(ast, right)?;
1093 let split = left_scope.len();
1094 let mut scope = left_scope.concat(right_scope);
1095
1096 let merged: Vec<String> = if natural {
1099 let mut names = Vec::new();
1100 for (at, column) in scope.columns.iter().enumerate().take(split) {
1101 if scope.columns[split..].iter().any(|right| same_name(&right.name, &column.name))
1102 && !names.iter().any(|held: &String| same_name(held, &column.name))
1103 {
1104 let _ = at;
1105 names.push(column.name.clone());
1106 }
1107 }
1108 names
1109 } else {
1110 ast.name(using).map(str::to_string).collect()
1111 };
1112
1113 let mut conditions = Vec::new();
1114 let mut dropped = Vec::new();
1115 for name in &merged {
1116 let left_at = scope.columns[..split]
1117 .iter()
1118 .position(|column| same_name(&column.name, name))
1119 .ok_or_else(|| {
1120 Error::binder(format!(
1121 "column \"{name}\" specified in USING clause does not exist in left table"
1122 ))
1123 })?;
1124 let right_at = scope.columns[split..]
1125 .iter()
1126 .position(|column| same_name(&column.name, name))
1127 .map(|at| at + split)
1128 .ok_or_else(|| {
1129 Error::binder(format!(
1130 "column \"{name}\" specified in USING clause does not exist in right table"
1131 ))
1132 })?;
1133 let left_column = &scope.columns[left_at];
1134 let (left_binding, left_type) = (left_column.binding, left_column.ty.clone());
1135 let right_column = &scope.columns[right_at];
1136 let (right_binding, right_type) = (right_column.binding, right_column.ty.clone());
1137 let left_expr = self.plan.add_expr(Expr::Column(left_binding), left_type);
1138 let right_expr = self.plan.add_expr(Expr::Column(right_binding), right_type);
1139 conditions.push(self.compare(rudb_plan::CompareOp::Equal, left_expr, right_expr)?);
1140 dropped.push(right_at);
1141 }
1142 dropped.sort_unstable();
1145 for at in dropped.into_iter().rev() {
1146 scope.remove(at);
1147 }
1148
1149 if on != NONE {
1150 if !merged.is_empty() {
1151 return Err(Error::binder("a join cannot have both ON and USING"));
1152 }
1153 self.clause = "JOIN condition";
1154 let predicate = self.bind_expr(ast, on, &scope)?;
1155 conditions.push(self.as_boolean(predicate, "JOIN")?);
1156 }
1157
1158 if kind == ast::JoinKind::Cross {
1159 if !conditions.is_empty() {
1160 return Err(Error::binder("a CROSS JOIN cannot have a condition"));
1161 }
1162 let node =
1163 self.plan.add_node(Node::CrossProduct { left: left_node, right: right_node });
1164 return Ok((node, scope));
1165 }
1166 if conditions.is_empty() && kind == ast::JoinKind::Inner {
1167 let node =
1168 self.plan.add_node(Node::CrossProduct { left: left_node, right: right_node });
1169 return Ok((node, scope));
1170 }
1171 let kind = match kind {
1172 ast::JoinKind::Inner | ast::JoinKind::Cross => JoinKind::Inner,
1173 ast::JoinKind::Left => JoinKind::Left,
1174 ast::JoinKind::Right => JoinKind::Right,
1175 ast::JoinKind::Full => JoinKind::Full,
1176 ast::JoinKind::Semi => JoinKind::Semi,
1177 ast::JoinKind::Anti => JoinKind::Anti,
1178 ast::JoinKind::Positional => JoinKind::Positional,
1179 };
1180 let conditions = self.plan.add_expr_list(&conditions);
1181 let node =
1182 self.plan.add_node(Node::Join { left: left_node, right: right_node, kind, conditions });
1183 Ok((node, scope))
1184 }
1185
1186 pub(crate) fn bind_aggregate(
1190 &mut self,
1191 ast: &Ast,
1192 name: &str,
1193 args: &[ast::ExprRef],
1194 distinct: bool,
1195 scope: &Scope,
1196 ) -> Result<ExprRef> {
1197 if self.in_aggregate {
1198 return Err(Error::binder(format!(
1199 "aggregate function calls cannot be nested, and {name}() is inside one"
1200 )));
1201 }
1202 if self.aggregation.is_none() {
1203 return Err(Error::binder(format!(
1204 "aggregate function calls cannot be used in the {}",
1205 self.clause
1206 )));
1207 }
1208 self.in_aggregate = true;
1209 let mut bound = Vec::with_capacity(args.len());
1210 let mut failure = None;
1211 for &arg in args {
1212 match self.bind_expr(ast, arg, scope) {
1213 Ok(expr) => bound.push(expr),
1214 Err(error) => {
1215 failure = Some(error);
1216 break;
1217 }
1218 }
1219 }
1220 self.in_aggregate = false;
1221 if let Some(error) = failure {
1222 return Err(error);
1223 }
1224
1225 let types: Vec<LogicalType> =
1226 bound.iter().map(|&arg| self.plan.expr_type(arg).clone()).collect();
1227 let resolved = resolve(name, &types)?;
1228 let mut cast = Vec::with_capacity(bound.len());
1229 for (arg, wanted) in bound.iter().zip(&resolved.arguments) {
1230 cast.push(self.cast_to(*arg, wanted));
1231 }
1232 let args = self.plan.add_expr_list(&cast);
1233 let name = self.plan.intern(resolved.name);
1234 let ty = resolved.returns;
1235 let call =
1236 self.plan.add_expr(Expr::Aggregate { name, args, distinct, filter: None }, ty.clone());
1237
1238 let existing = self.aggregation.as_ref().map(|held| held.aggregates.clone());
1241 let existing = existing.unwrap_or_default();
1242 let at = match existing.iter().position(|&held| self.same_expr(held, call)) {
1243 Some(at) => at,
1244 None => {
1245 let aggregation = self.aggregation.as_mut().expect("checked above");
1246 aggregation.aggregates.push(call);
1247 aggregation.aggregates.len() - 1
1248 }
1249 };
1250 let aggregation = self.aggregation.as_ref().expect("checked above");
1251 let (index, groups) = (aggregation.index, aggregation.groups.len());
1252 Ok(self.column(index, groups + at, ty))
1253 }
1254
1255 pub(crate) fn over_aggregate(&mut self, expr: ExprRef, scope: &Scope) -> Result<ExprRef> {
1261 let Some(aggregation) = self.aggregation.as_ref() else {
1262 return Ok(expr);
1263 };
1264 let index = aggregation.index;
1265 let groups = aggregation.groups.clone();
1266 for (at, group) in groups.iter().enumerate() {
1267 if self.same_expr(expr, *group) {
1268 let ty = self.plan.expr_type(*group).clone();
1269 return Ok(self.column(index, at, ty));
1270 }
1271 }
1272 let ty = self.plan.expr_type(expr).clone();
1273 match self.plan.expr(expr).clone() {
1274 Expr::Column(binding) if binding.table == index => Ok(expr),
1275 Expr::Column(binding) => {
1276 let name =
1277 scope.columns.iter().find(|column| column.binding == binding).map_or_else(
1278 || "a column".to_string(),
1279 |column| format!("\"{}\"", column.name),
1280 );
1281 Err(Error::binder(format!(
1282 "column {name} must appear in the GROUP BY clause or must be part of an aggregate function"
1283 )))
1284 }
1285 Expr::Constant(_) | Expr::Aggregate { .. } => Ok(expr),
1286 Expr::Cast { input, try_cast } => {
1287 let input = self.over_aggregate(input, scope)?;
1288 Ok(self.plan.add_expr(Expr::Cast { input, try_cast }, ty))
1289 }
1290 Expr::Compare { op, left, right } => {
1291 let left = self.over_aggregate(left, scope)?;
1292 let right = self.over_aggregate(right, scope)?;
1293 Ok(self.plan.add_expr(Expr::Compare { op, left, right }, ty))
1294 }
1295 Expr::Conjunction { op, children } => {
1296 let written = self.plan.expr_list(children).to_vec();
1297 let mut rewritten = Vec::with_capacity(written.len());
1298 for child in written {
1299 rewritten.push(self.over_aggregate(child, scope)?);
1300 }
1301 let children = self.plan.add_expr_list(&rewritten);
1302 Ok(self.plan.add_expr(Expr::Conjunction { op, children }, ty))
1303 }
1304 Expr::Function { name, args } => {
1305 let written = self.plan.expr_list(args).to_vec();
1306 let mut rewritten = Vec::with_capacity(written.len());
1307 for arg in written {
1308 rewritten.push(self.over_aggregate(arg, scope)?);
1309 }
1310 let args = self.plan.add_expr_list(&rewritten);
1311 Ok(self.plan.add_expr(Expr::Function { name, args }, ty))
1312 }
1313 Expr::Case { arms, otherwise } => {
1314 let written = self.plan.arm_list(arms).to_vec();
1315 let mut rewritten = Vec::with_capacity(written.len());
1316 for arm in written {
1317 let when = self.over_aggregate(arm.when, scope)?;
1318 let then = self.over_aggregate(arm.then, scope)?;
1319 rewritten.push(rudb_plan::Arm { when, then });
1320 }
1321 let otherwise = match otherwise {
1322 Some(expr) => Some(self.over_aggregate(expr, scope)?),
1323 None => None,
1324 };
1325 let arms = self.plan.add_arms(&rewritten);
1326 Ok(self.plan.add_expr(Expr::Case { arms, otherwise }, ty))
1327 }
1328 }
1329 }
1330
1331 pub(crate) fn same_expr(&self, left: ExprRef, right: ExprRef) -> bool {
1333 same_expr(&self.plan, left, right)
1334 }
1335}
1336
1337fn sort_key(expr: ExprRef, item: ast::OrderItem) -> SortKey {
1343 let descending = item.order == Order::Descending;
1344 let nulls_first = match item.nulls {
1345 Nulls::First => true,
1346 Nulls::Last => false,
1347 Nulls::Unstated => descending,
1348 };
1349 SortKey { expr, descending, nulls_first }
1350}
1351
1352fn same_expr(plan: &Plan, left: ExprRef, right: ExprRef) -> bool {
1354 if left == right {
1355 return true;
1356 }
1357 if plan.expr_type(left) != plan.expr_type(right) {
1358 return false;
1359 }
1360 let lists = |left, right| {
1361 let left: &[ExprRef] = plan.expr_list(left);
1362 let right: &[ExprRef] = plan.expr_list(right);
1363 left.len() == right.len()
1364 && left.iter().zip(right).all(|(&left, &right)| same_expr(plan, left, right))
1365 };
1366 match (plan.expr(left), plan.expr(right)) {
1367 (Expr::Column(left), Expr::Column(right)) => left == right,
1368 (Expr::Constant(left), Expr::Constant(right)) => plan.value(*left) == plan.value(*right),
1369 (
1370 Expr::Cast { input: left, try_cast: left_try },
1371 Expr::Cast { input: right, try_cast: right_try },
1372 ) => left_try == right_try && same_expr(plan, *left, *right),
1373 (
1374 Expr::Compare { op: left_op, left: left_a, right: left_b },
1375 Expr::Compare { op: right_op, left: right_a, right: right_b },
1376 ) => {
1377 left_op == right_op
1378 && same_expr(plan, *left_a, *right_a)
1379 && same_expr(plan, *left_b, *right_b)
1380 }
1381 (
1382 Expr::Conjunction { op: left_op, children: left_children },
1383 Expr::Conjunction { op: right_op, children: right_children },
1384 ) => left_op == right_op && lists(*left_children, *right_children),
1385 (
1386 Expr::Function { name: left_name, args: left_args },
1387 Expr::Function { name: right_name, args: right_args },
1388 ) => plan.string(*left_name) == plan.string(*right_name) && lists(*left_args, *right_args),
1389 (
1390 Expr::Aggregate {
1391 name: left_name,
1392 args: left_args,
1393 distinct: left_distinct,
1394 filter: left_filter,
1395 },
1396 Expr::Aggregate {
1397 name: right_name,
1398 args: right_args,
1399 distinct: right_distinct,
1400 filter: right_filter,
1401 },
1402 ) => {
1403 plan.string(*left_name) == plan.string(*right_name)
1404 && left_distinct == right_distinct
1405 && match (left_filter, right_filter) {
1406 (None, None) => true,
1407 (Some(left), Some(right)) => same_expr(plan, *left, *right),
1408 _ => false,
1409 }
1410 && lists(*left_args, *right_args)
1411 }
1412 (
1413 Expr::Case { arms: left_arms, otherwise: left_otherwise },
1414 Expr::Case { arms: right_arms, otherwise: right_otherwise },
1415 ) => {
1416 let left_arms = plan.arm_list(*left_arms);
1417 let right_arms = plan.arm_list(*right_arms);
1418 left_arms.len() == right_arms.len()
1419 && left_arms.iter().zip(right_arms).all(|(left, right)| {
1420 same_expr(plan, left.when, right.when) && same_expr(plan, left.then, right.then)
1421 })
1422 && match (left_otherwise, right_otherwise) {
1423 (None, None) => true,
1424 (Some(left), Some(right)) => same_expr(plan, *left, *right),
1425 _ => false,
1426 }
1427 }
1428 _ => false,
1429 }
1430}