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