rudb_exec/
build.rs

1//! Turning a bound plan into a tree of operators.
2//!
3//! One match, one arm per logical operator, and nothing else. There is no physical plan and no cost
4//! based choice between two ways of running the same node, which is the honest description of tier
5//! 0: there is one implementation of each operator so there is nothing to choose between. The
6//! physical planner that section 9.6 describes goes here, and the reason this is a separate module
7//! from the operators is so that it can grow into one without any of them moving.
8//!
9//! The tree borrows the plan and the catalog for as long as it exists. A scan reads its rows out of
10//! the catalog's table rather than copying them, and an expression reads its constants, its function
11//! names and its types out of the plan's arena, so a plan that outlives the query it built is the
12//! whole of the lifetime story here.
13
14use rudb_catalog::{Catalog, QualifiedName};
15use rudb_common::{Cancel, Memory, Result};
16use rudb_functions::TableFunction;
17use rudb_plan::{Node, NodeRef, Plan};
18
19use crate::cancel::Guarded;
20use crate::group::{Aggregate, Distinct};
21use crate::join::{CrossProduct, Join};
22use crate::operator::Operator;
23use crate::setop::SetOp;
24use crate::sort::Sort;
25use crate::source::{Dummy, FileScan, Scan, Series, Values};
26use crate::strategies::Strategies;
27use crate::stream::{Filter, Limit, Project};
28use crate::topn::TopN;
29
30/// Builds the operator tree for a plan's root, for a query nothing will stop.
31///
32/// # Errors
33///
34/// If the plan names a table or a column the catalog does not have, if an expression is malformed
35/// in a way [`Plan::validate`] would have caught, or anything an operator's construction reports.
36pub fn build<'a>(plan: &'a Plan, catalog: &'a Catalog) -> Result<Box<dyn Operator + 'a>> {
37    build_with(plan, catalog, &Cancel::new(), &Memory::unlimited())
38}
39
40/// Builds the operator tree for a plan's root, stoppable through this token and held to this
41/// budget.
42///
43/// Every node in the tree is wrapped in a check, so the query stops at the first chunk boundary
44/// after the token says to. See the `cancel` module for why the check is uniform rather than
45/// placed in the operators that can loop.
46///
47/// The budget is not uniform, and that is the difference between the two. A streaming operator
48/// holds one chunk and gives it away again, so charging every node would count the same megabyte
49/// once per level of the tree. Only the operators that buffer without bound take a reservation, and
50/// [`rudb_common::Memory`] lists which ones those are.
51///
52/// # Errors
53///
54/// The same as [`build`].
55pub fn build_with<'a>(
56    plan: &'a Plan,
57    catalog: &'a Catalog,
58    cancel: &Cancel,
59    memory: &Memory,
60) -> Result<Box<dyn Operator + 'a>> {
61    node(plan, catalog, cancel, memory, plan.root())
62}
63
64fn node<'a>(
65    plan: &'a Plan,
66    catalog: &'a Catalog,
67    cancel: &Cancel,
68    memory: &Memory,
69    reference: NodeRef,
70) -> Result<Box<dyn Operator + 'a>> {
71    let inner: Box<dyn Operator + 'a> = match *plan.node(reference) {
72        Node::Get { catalog: database, schema, table, index, columns, .. } => {
73            let name =
74                QualifiedName::new(plan.string(database), plan.string(schema), plan.string(table));
75            Box::new(Scan::new(plan, catalog.table(&name)?, index, columns)?)
76        }
77        Node::Dummy => Box::new(Dummy::new()),
78        Node::Values { index, columns, rows } => Box::new(Values::new(plan, index, columns, rows)?),
79        Node::TableFunction { index, function, args, options, settings, columns } => {
80            match TableFunction::lookup(plan.string(function)) {
81                Some(function @ (TableFunction::ReadParquet | TableFunction::ReadCsv)) => Box::new(
82                    FileScan::new(plan, index, function, args, options, settings, columns)?,
83                ),
84                Some(TableFunction::RudbStrategies) => {
85                    Box::new(Strategies::new(plan, index, columns)?)
86                }
87                _ => Box::new(Series::new(plan, index, plan.string(function), args)?),
88            }
89        }
90        Node::Filter { input, predicate } => {
91            Box::new(Filter::new(plan, node(plan, catalog, cancel, memory, input)?, predicate)?)
92        }
93        Node::Project { input, index, exprs, names } => Box::new(Project::new(
94            plan,
95            node(plan, catalog, cancel, memory, input)?,
96            index,
97            exprs,
98            names,
99        )?),
100        Node::Aggregate { input, index, groups, aggregates } => Box::new(Aggregate::new(
101            plan,
102            node(plan, catalog, cancel, memory, input)?,
103            index,
104            groups,
105            aggregates,
106            memory,
107        )?),
108        Node::Sort { input, keys } => {
109            Box::new(Sort::new(plan, node(plan, catalog, cancel, memory, input)?, keys, memory))
110        }
111        Node::Limit { input, count, offset } => {
112            Box::new(Limit::new(node(plan, catalog, cancel, memory, input)?, count, offset))
113        }
114        Node::TopN { input, keys, count, offset } => Box::new(TopN::new(
115            plan,
116            node(plan, catalog, cancel, memory, input)?,
117            keys,
118            count,
119            offset,
120            memory,
121        )),
122        Node::Distinct { input, on } => {
123            Box::new(Distinct::new(plan, node(plan, catalog, cancel, memory, input)?, on, memory))
124        }
125        Node::Join { left, right, kind, conditions } => Box::new(Join::new(
126            plan,
127            node(plan, catalog, cancel, memory, left)?,
128            node(plan, catalog, cancel, memory, right)?,
129            kind,
130            conditions,
131            cancel,
132            memory,
133        )),
134        Node::CrossProduct { left, right } => Box::new(CrossProduct::new(
135            node(plan, catalog, cancel, memory, left)?,
136            node(plan, catalog, cancel, memory, right)?,
137            memory,
138        )),
139        Node::SetOp { left, right, kind, all, index } => Box::new(SetOp::new(
140            node(plan, catalog, cancel, memory, left)?,
141            node(plan, catalog, cancel, memory, right)?,
142            kind,
143            all,
144            index,
145            memory,
146        )),
147    };
148    Ok(Box::new(Guarded::new(inner, cancel.clone())))
149}
rudb_exec/build.rs

rudb_exec/
build.rs
