rudb_exec/prepared.rs
1//! An expression prepared once for a pipeline and then evaluated over every chunk.
2//!
3//! `spec/engine/04-expressions.md`. [`evaluate`](crate::evaluate) walks the plan's expression tree
4//! on every chunk, which means it does four things per chunk that depend on nothing about the
5//! chunk: it recurses, it resolves every column reference by a linear search through the schema, it
6//! clones a [`LogicalType`] for every node, and it copies the whole column a [`Expr::Column`] names.
7//! Over `hits` at a hundred thousand chunks that is a hundred thousand schema searches per column
8//! reference and a hundred thousand copies of every column any expression mentions.
9//!
10//! This type does all four once. The tree is flattened into a post order array, so evaluating it is
11//! a loop over that array and the recursion is gone with it. Column references are resolved to
12//! positions when the pipeline is built. Types are held here rather than cloned out of the plan.
13//! And a column reference is not a step that produces anything: it is read straight out of the chunk
14//! at the point an operand is wanted, so the column is never copied at all.
15//!
16//! # What is shared and what is not
17//!
18//! [`Prepared`] is immutable after it is built and is `Send` and `Sync`, so one of them serves every
19//! thread running a copy of the pipeline. [`Scratch`] is the per chunk working space and there is
20//! one per pipeline instance. That split is not for this layer's benefit. It is the same split every
21//! operator needs at layer eight, where the scheduler runs one pipeline on as many threads as it has
22//! morsels for, and building it here means the operators above are written against it from the start
23//! rather than retrofitted onto it.
24//!
25//! # What is still allocated per chunk
26//!
27//! Two things, and both are named rather than hidden. A node with four or more operands gathers
28//! references to them into a `Vec<&Vector>` so a kernel can take a slice, which is one allocation of
29//! pointers rather than a copy of any data, and which a node of one, two or three operands does on
30//! the stack instead. And every kernel allocates the vector it returns, because no kernel in
31//! `rudb-kernels` takes an output parameter. The second is much the larger of the two and it is the
32//! one tier 1 fusion removes, which is scheduled after layer six for the reason
33//! `spec/engine/04-expressions.md` gives: once the tree walk is gone what is left to save is pass
34//! count, and at 1024 rows the intermediate vectors are eight kilobytes and stay in L1.
35
36use rudb_common::{
37 Error, LogicalType, PhysicalType, Result, Session, SessionTimeZone, Span, Value,
38};
39use rudb_kernels::{
40 Comparison, Connective, Found, Held, Lookup, Members, Recipe, cast_in_time_zone, combine,
41 compare_prepared, in_set, is_true, refine_flags, refine_prepared, select_prepared, selection,
42};
43use rudb_plan::{CompareOp, ConjunctionOp, Expr, ExprRef, Plan};
44use rudb_vector::{Assembly, Chunk, Selection, Vector};
45use std::collections::HashMap;
46use std::sync::Arc;
47
48use crate::fused::Fused;
49use crate::lambda::{Lambda, lambda_call};
50use crate::ordering::Ordering;
51use crate::schema::Schema;
52use crate::written::written;
53
54/// The scheduler's half of the expression contract, imposed now rather than at layer eight.
55///
56/// A prepared expression is the immutable half of a pipeline and layer eight hands one of them to
57/// every thread running that pipeline. That is only sound if it holds nothing thread local, and the
58/// way to find out on the commit that breaks it rather than eight layers later is to ask the
59/// compiler here, exactly as [`Chunk`] does for the data plane.
60const _: () = {
61 const fn assert_shareable<T: Send + Sync>() {}
62 assert_shareable::<Prepared>();
63};
64
65/// One or more bound expressions, flattened and resolved against a schema.
66///
67/// Built once per pipeline with [`Prepared::new`] and evaluated per chunk with
68/// [`Prepared::evaluate`] or [`Prepared::evaluate_one`], each of which wants the [`Scratch`] that
69/// [`Prepared::scratch`] hands out.
70#[derive(Debug)]
71pub struct Prepared {
72 /// The nodes in post order, so every node's operands have already been computed when it runs.
73 steps: Vec<Step>,
74 /// The type each step produces, indexed the same way as `steps`.
75 ///
76 /// A parallel array rather than a field in the variant, for the reason [`Expr`] gives: a
77 /// [`LogicalType`] owns a `Vec` for its nested cases and putting one in every variant would make
78 /// the common variants several times larger for the benefit of the rare ones.
79 types: Vec<LogicalType>,
80 /// The source range each step came from, indexed the same way as `steps`.
81 spans: Vec<Span>,
82 /// The operand lists of the steps that have one, as runs of step indices.
83 operands: Vec<usize>,
84 /// The last step that reads each step's slot, or `usize::MAX` for one nothing reads.
85 ///
86 /// A slot is emptied as soon as the step that was the last to read it has run. Keeping every
87 /// intermediate alive to the end of the array instead is what the first measured version of this
88 /// did, and a chain of eight additions was slower prepared than walked because of it: nine live
89 /// intermediates at eight kilobytes each is seventy two kilobytes of working set where the tree
90 /// walk has two, and two is the pair the allocator hands back and forth and that stays in L1.
91 /// Everything else about the prepared form was faster and this one thing paid all of it back.
92 last_use: Vec<usize>,
93 /// The step index each expression this was built from ends at.
94 roots: Vec<usize>,
95 /// The step already compiled for each shared plan expression.
96 shared: HashMap<ExprRef, usize>,
97 share: bool,
98 /// Whether a tree of decimal arithmetic is run as one [`Fused`] step. Off only for the steps a
99 /// fused one falls back to, which would otherwise fuse themselves again.
100 fuse: bool,
101 /// The parsed zone used only by casts whose answer depends on the session.
102 time_zone: SessionTimeZone,
103}
104
105/// One node of a flattened expression.
106///
107/// A step refers to its operands by their index in [`Prepared::steps`], which is always smaller than
108/// its own because the array is in post order.
109#[derive(Debug)]
110enum Step {
111 /// A column of the chunk, by resolved position.
112 ///
113 /// This step computes nothing. Its slot stays empty and an operand that names it is read out of
114 /// the chunk, which is the whole of what makes a column reference free rather than a copy.
115 Column(usize),
116 /// A literal, materialized into a constant vector as long as the chunk.
117 Constant(Value),
118 /// A cast to this step's own type.
119 Cast {
120 /// The step being cast.
121 input: usize,
122 /// Whether a failed cast yields null instead of raising.
123 try_cast: bool,
124 },
125 /// A binary comparison.
126 Compare {
127 /// Which comparison.
128 op: Comparison,
129 /// The left operand's step.
130 left: usize,
131 /// The right operand's step.
132 right: usize,
133 /// The side that is a literal, in the one row column the comparison loops read it through,
134 /// and `None` when neither side is one.
135 ///
136 /// Built here because the loops read both sides through a slice, so the constant side has
137 /// to become a column somewhere, and the plan says which side that is. For a string it is
138 /// also where the four byte prefix comes from, which is what almost every row of a string
139 /// comparison is decided by.
140 held: Option<Held>,
141 },
142 /// An `AND` or `OR` over a run of [`Prepared::operands`].
143 Conjunction {
144 /// Which connective.
145 op: Connective,
146 /// Where the operand list starts.
147 start: usize,
148 /// How many operands it has.
149 len: usize,
150 },
151 /// A scalar function over a run of [`Prepared::operands`].
152 Function {
153 /// The call, with the name resolved and whatever the kernel could work out from the
154 /// arguments that were literals already worked out.
155 ///
156 /// Held here so the plan is not consulted per chunk, and built here so that a regular
157 /// expression is compiled once for the query rather than once for each of the hundred
158 /// thousand chunks a pipeline over `hits` runs.
159 recipe: Recipe,
160 /// How the call is written, for the one error message that quotes it.
161 ///
162 /// Rendered when the pipeline is built rather than when a chunk arrives, because the plan
163 /// is here and is not there. It is a short string per function node in the query and it is
164 /// built once, which is a different cost from the tree walk's, where the plan is still to
165 /// hand and the rendering can wait until the row that fails.
166 written: String,
167 /// Where the argument list starts.
168 start: usize,
169 /// How many arguments it has.
170 len: usize,
171 },
172 /// A membership test over a list the query wrote out.
173 ///
174 /// The binder has no `IN` node: `x IN (1, 2, 3)` arrives as an `OR` of three equalities and
175 /// `x NOT IN (1, 2, 3)` as an `AND` of three inequalities. That is the right shape for a binder
176 /// to produce, because nothing after it then needs a second set of rules for null, and it is the
177 /// wrong shape to run, because it is a pass over the column and an output vector per entry.
178 /// This is that shape folded back up, and folding it here rather than after the operands are
179 /// pushed is what keeps the equalities from being run anyway.
180 InSet {
181 /// The step being tested.
182 input: usize,
183 /// The list, as a set, with the null rule and the direction it is read in.
184 members: Members,
185 },
186 /// A searched `CASE`, whose branches are prepared expressions of their own.
187 ///
188 /// Nested rather than flattened into the same array because a branch is not evaluated over the
189 /// chunk, it is evaluated over the rows no earlier arm claimed, and a step in the outer array
190 /// would have no way to say that. The selection threaded form in #57 replaces this whole
191 /// variant, and when it does the branches stop being separate arrays.
192 Case {
193 /// The `WHEN`/`THEN` pairs, in order.
194 arms: Vec<PreparedArm>,
195 /// The `ELSE`, if there is one. Absent means null.
196 otherwise: Option<Prepared>,
197 /// How to answer it as codes, for the shape that can be. Absent means read the values.
198 blend: Option<Blend>,
199 },
200 /// A tree of decimal arithmetic over columns and literals, run as one loop when the columns'
201 /// ranges prove it cannot overflow.
202 ///
203 /// The fallback is the same tree prepared the ordinary way, nested for the reason a case's
204 /// branches are, and it is what runs over a chunk the ranges do not settle.
205 Fused {
206 /// The program.
207 fused: Box<Fused>,
208 /// The steps it replaced.
209 fallback: Box<Prepared>,
210 },
211 /// A call to a function that takes a lambda, whose body is a prepared expression of its own.
212 ///
213 /// Nested for the reason a case's branches are: the body does not run over the chunk, it runs
214 /// over a chunk with a row per element that [`Lambda`] builds, and a step in the outer array has
215 /// no way to say that.
216 Lambda {
217 /// The steps of the call's other arguments: the list and `list_reduce`'s initial value, or
218 /// `invoke`'s parameters.
219 inputs: Vec<usize>,
220 /// The layout of what the body runs over and what to do with its answers.
221 runner: Box<Lambda>,
222 /// The body, prepared against the runner's schema.
223 body: Box<Prepared>,
224 },
225}
226
227/// One `WHEN`/`THEN` pair of a prepared [`Step::Case`].
228#[derive(Debug)]
229struct PreparedArm {
230 /// The condition.
231 when: Prepared,
232 /// The result if the condition is true.
233 then: Prepared,
234}
235
236/// A `CASE` over text whose every branch is a column or a literal, answered as codes.
237///
238/// What the general path does with the branches is read their values and write them into a vector of
239/// their own, which for a text column out of a native file decodes a compressed dictionary block per
240/// row and then throws the dictionary away. An operator above that has to work with strings even
241/// though every string it sees came out of one dictionary it could have kept.
242///
243/// It does not have to. The branches here name values rather than compute them, so if they all name
244/// values of one dictionary then so does the answer, and the answer is the codes: one code per row
245/// copied from the branch that claimed the row, and a literal is one code for all of its rows once
246/// the dictionary has been searched for it. Nothing is read and the dictionary comes out the other
247/// side, so a group by over the `CASE` groups on codes the way a group by over the bare column does.
248///
249/// ClickBench 39 is the query this is for. It groups by `CASE WHEN (SearchEngineID = 0 AND
250/// AdvEngineID = 0) THEN Referer ELSE '' END` beside `URL`, and writing that one column out as
251/// strings was a quarter of the query.
252///
253/// The shape is narrow on purpose. A branch that computes anything is not here, because then the
254/// answer is a value that no dictionary holds. A literal the dictionary does not hold is not here
255/// either, for the same reason, and that is decided per dictionary at run time rather than when the
256/// expression is prepared. And a `CASE` with no `ELSE` is not here, because the rows nothing claims
257/// are null and a null is not a code.
258#[derive(Debug)]
259struct Blend {
260 /// Where each branch takes its value from: one per arm in order, and the `ELSE` last.
261 branches: Vec<Branch>,
262 /// The literals the branches name, each with the search that finds it in a dictionary.
263 literals: Vec<(String, Lookup)>,
264}
265
266/// Where one branch of a [`Blend`] takes its value from.
267#[derive(Debug, Clone, Copy)]
268enum Branch {
269 /// A column of the chunk, by resolved position. Its rows keep the codes they arrived with.
270 Column(usize),
271 /// The literal at this index of [`Blend::literals`]. Its rows all get one code.
272 Literal(usize),
273}
274
275/// The per chunk working space of one [`Prepared`].
276///
277/// One per pipeline instance and never shared, which is the mutable half of the split the module
278/// documentation describes. It is handed back in rather than made inside [`Prepared::evaluate`] so
279/// that the array of slots survives from one chunk to the next instead of being allocated a hundred
280/// thousand times over a scan.
281#[derive(Debug, Default)]
282pub struct Scratch {
283 /// What each step produced, or `None` for a step that produces nothing and for one that has not
284 /// run yet.
285 slots: Vec<Option<Vector>>,
286 /// What each connective step has learned about its operands, indexed by step.
287 ///
288 /// Empty for every step that is not a connective and for a connective a filter has not reached
289 /// yet, since it is built the first time one runs and the shape it needs is not known before
290 /// then. This is the mutable half of the adaptive ordering and it is here rather than in
291 /// [`Prepared`] because a prepared expression is shared by every thread running the pipeline.
292 orders: Vec<Option<Ordering>>,
293}
294
295impl Scratch {
296 /// The order a connective's operands are run in.
297 ///
298 /// For the tests that say the learning reached the walk. Nothing in the engine asks a scratch
299 /// this, because the walk is the only thing that reads an ordering and it reads its own.
300 #[cfg(test)]
301 fn order(&self, step: usize) -> Option<&[usize]> {
302 self.orders[step].as_ref().map(Ordering::order)
303 }
304}
305
306impl Prepared {
307 /// Prepares `exprs` against `schema`.
308 ///
309 /// # Errors
310 ///
311 /// If a column reference names a binding the schema does not have, or if an aggregate appears
312 /// where an ordinary expression was expected. Both are failures of the plan rather than of the
313 /// data, which is why they are found here, once, rather than on some chunk in the middle of a
314 /// scan.
315 pub fn new(plan: &Plan, exprs: &[ExprRef], schema: &Schema) -> Result<Self> {
316 Self::build(plan, exprs, schema, false)
317 }
318
319 /// Prepares expressions whose caller can evaluate a shared expression graph as one unit.
320 pub(crate) fn shared(plan: &Plan, exprs: &[ExprRef], schema: &Schema) -> Result<Self> {
321 Self::build(plan, exprs, schema, true)
322 }
323
324 fn build(plan: &Plan, exprs: &[ExprRef], schema: &Schema, share: bool) -> Result<Self> {
325 Self::built(plan, exprs, schema, share, true)
326 }
327
328 fn built(
329 plan: &Plan,
330 exprs: &[ExprRef],
331 schema: &Schema,
332 share: bool,
333 fuse: bool,
334 ) -> Result<Self> {
335 let mut prepared = Self {
336 steps: Vec::new(),
337 types: Vec::new(),
338 spans: Vec::new(),
339 operands: Vec::new(),
340 last_use: Vec::new(),
341 roots: Vec::new(),
342 shared: HashMap::new(),
343 share,
344 fuse,
345 time_zone: SessionTimeZone::default(),
346 };
347 for &expr in exprs {
348 let root = prepared.push(plan, expr, schema)?;
349 prepared.roots.push(root);
350 }
351 prepared.last_use = prepared.last_uses();
352 Ok(prepared)
353 }
354
355 /// Uses the zone of the session that owns this prepared expression.
356 #[must_use]
357 pub fn in_session(mut self, session: &Session) -> Self {
358 self.set_time_zone(session.session_time_zone());
359 self
360 }
361
362 /// Sets the zone here and in every lambda body, which is prepared before the session is known.
363 fn set_time_zone(&mut self, time_zone: SessionTimeZone) {
364 self.time_zone = time_zone;
365 for step in &mut self.steps {
366 match step {
367 Step::Lambda { body, .. } => body.set_time_zone(time_zone),
368 Step::Fused { fallback, .. } => fallback.set_time_zone(time_zone),
369 _ => {}
370 }
371 }
372 }
373
374 /// Which step is the last to read each step, computed once when the expression is prepared.
375 ///
376 /// A root is never freed, because the whole point of running the array was to produce it. A
377 /// step nothing reads and that is not a root cannot happen, since every step is pushed by the
378 /// node that wanted it, but saying `usize::MAX` rather than asserting that keeps this a fact
379 /// about the array rather than a claim about the builder.
380 fn last_uses(&self) -> Vec<usize> {
381 let mut last = vec![usize::MAX; self.steps.len()];
382 for index in 0..self.steps.len() {
383 self.for_each_operand(index, |operand| last[operand] = index);
384 }
385 for &root in &self.roots {
386 last[root] = usize::MAX;
387 }
388 last
389 }
390
391 /// Visits the steps one step reads, whatever shape its operands are held in.
392 fn for_each_operand(&self, index: usize, mut visit: impl FnMut(usize)) {
393 match &self.steps[index] {
394 // A case's branches are arrays of their own and read nothing out of this one, and a
395 // fused tree reads its columns straight out of the chunk.
396 Step::Column(_) | Step::Constant(_) | Step::Case { .. } | Step::Fused { .. } => {}
397 Step::Cast { input, .. } | Step::InSet { input, .. } => visit(*input),
398 Step::Lambda { inputs, .. } => inputs.iter().for_each(|&input| visit(input)),
399 Step::Compare { left, right, .. } => {
400 visit(*left);
401 visit(*right);
402 }
403 Step::Conjunction { start, len, .. } | Step::Function { start, len, .. } => {
404 for &operand in &self.operands[*start..*start + *len] {
405 visit(operand);
406 }
407 }
408 }
409 }
410
411 /// Prepares one expression, which is the common case and saves the caller a slice.
412 ///
413 /// # Errors
414 ///
415 /// Whatever [`Prepared::new`] reports.
416 pub fn one(plan: &Plan, expr: ExprRef, schema: &Schema) -> Result<Self> {
417 Self::new(plan, &[expr], schema)
418 }
419
420 /// Working space sized for this expression.
421 #[must_use]
422 pub fn scratch(&self) -> Scratch {
423 Scratch {
424 slots: (0..self.steps.len()).map(|_| None).collect(),
425 orders: (0..self.steps.len()).map(|_| None).collect(),
426 }
427 }
428
429 /// How many expressions this was built from.
430 #[must_use]
431 pub fn len(&self) -> usize {
432 self.roots.len()
433 }
434
435 /// How many comparisons have their literal side already built.
436 ///
437 /// For the tests, for the same reason as [`Self::sets`]: an answer that moved would be a bug,
438 /// so the only thing a test can look at is whether the building happened.
439 #[cfg(test)]
440 fn literals_built(&self) -> usize {
441 self.steps.iter().filter(|step| matches!(step, Step::Compare { held: Some(_), .. })).count()
442 }
443
444 /// How many of the steps are an `IN` list folded back up.
445 ///
446 /// For the tests, which cannot see the fold in an answer because an answer that changed would
447 /// be a bug.
448 #[cfg(test)]
449 fn sets(&self) -> usize {
450 self.steps.iter().filter(|step| matches!(step, Step::InSet { .. })).count()
451 }
452
453 /// How many of the steps are a tree of decimal arithmetic run as one loop.
454 #[cfg(test)]
455 fn fused(&self) -> usize {
456 self.steps.iter().filter(|step| matches!(step, Step::Fused { .. })).count()
457 }
458
459 /// How many of the function steps worked something out when this was built.
460 ///
461 /// For the tests, which cannot see the hoisting in an answer because an answer that changed
462 /// would be a bug.
463 #[cfg(test)]
464 fn hoisted(&self) -> usize {
465 self.steps
466 .iter()
467 .filter(|step| matches!(step, Step::Function { recipe, .. } if recipe.hoists()))
468 .count()
469 }
470
471 /// Whether it was built from no expressions at all.
472 #[must_use]
473 pub fn is_empty(&self) -> bool {
474 self.roots.is_empty()
475 }
476
477 /// How many of the steps do something to a row.
478 ///
479 /// A column reference and a literal are not among them. A column reference computes nothing at
480 /// all, which is what makes a step that names one free rather than a copy, and a literal is
481 /// materialized once for the whole chunk rather than once a row. What is left is a pass over
482 /// the rows each, so this is roughly what one row costs, counted in the same unit the scan's
483 /// own reading of that row is counted in.
484 ///
485 /// What reads it is the scan, through the weight an operator reports to the pipeline. See
486 /// [`Stream::weight`](rudb_pipeline::Stream::weight).
487 #[must_use]
488 pub fn passes(&self) -> usize {
489 self.steps
490 .iter()
491 .filter(|step| !matches!(step, Step::Column(_) | Step::Constant(_)))
492 .count()
493 }
494
495 /// Evaluates every expression over `chunk`, appending one vector each to `out`.
496 ///
497 /// Appends rather than returns a `Vec`, so a caller in a loop reuses one buffer.
498 ///
499 /// # Errors
500 ///
501 /// Anything a kernel reports, on the first expression that reports it.
502 pub fn evaluate(
503 &self,
504 chunk: &Chunk,
505 scratch: &mut Scratch,
506 out: &mut Vec<Vector>,
507 ) -> Result<()> {
508 self.run(chunk, scratch)?;
509 let mut remaining: HashMap<usize, usize> = HashMap::new();
510 for &root in &self.roots {
511 *remaining.entry(root).or_default() += 1;
512 }
513 for &root in &self.roots {
514 // The one place a column is copied, and it is copied because the caller is taking
515 // ownership of a vector that has to outlive the chunk it came from. `SELECT a` is that
516 // shape and a projection of a bare column is the only expression where it happens.
517 match self.steps[root] {
518 Step::Column(position) => out.push(chunk.column(position)?.clone()),
519 _ => {
520 let Some(left) = remaining.get_mut(&root) else {
521 return Err(Error::internal("a prepared root was not counted"));
522 };
523 *left -= 1;
524 if *left == 0 {
525 out.push(scratch.slots[root].take().ok_or_else(|| missing(root))?);
526 } else {
527 out.push(
528 scratch.slots[root].as_ref().ok_or_else(|| missing(root))?.clone(),
529 );
530 }
531 }
532 }
533 }
534 Ok(())
535 }
536
537 /// Evaluates a single expression over `chunk`, handing back a reference to the answer.
538 ///
539 /// A reference rather than a vector, because the caller of this is a filter, which reads the
540 /// flags to build a selection and then drops them. Nothing about that wants ownership, and a
541 /// predicate that is a bare column reference, which `WHERE flag` is, would otherwise copy the
542 /// column to hand it over.
543 ///
544 /// # Errors
545 ///
546 /// Anything a kernel reports, and an internal error if this was not built from exactly one
547 /// expression.
548 pub fn evaluate_one<'s>(
549 &'s self,
550 chunk: &'s Chunk,
551 scratch: &'s mut Scratch,
552 ) -> Result<&'s Vector> {
553 let [root] = self.roots[..] else {
554 return Err(Error::internal(format!(
555 "evaluate_one over a prepared expression of {} roots",
556 self.roots.len()
557 )));
558 };
559 self.run(chunk, scratch)?;
560 self.operand(root, chunk, &scratch.slots)
561 }
562
563 /// Evaluates a single expression as a filter, handing back the rows it keeps.
564 ///
565 /// The difference between this and [`evaluate_one`](Self::evaluate_one) followed by
566 /// [`selection`] is the whole of what a threaded filter is. An `AND` evaluated as an expression
567 /// runs every conjunct over every row and then combines the flag vectors, so a predicate of four
568 /// conjuncts that each pass a fifth of the rows does five times the work of one that stops
569 /// looking at a row as soon as a conjunct rejects it. TPC-H Q6 is exactly that predicate.
570 ///
571 /// So the conjuncts of a top level `AND` are run one at a time, each over the rows the ones
572 /// before it left, and the moment nothing is left the rest of the predicate is not run at all.
573 /// The order they run in starts as the order the plan gives and then moves, because which
574 /// conjunct is worth running first is a question about the data and the scan is the thing
575 /// holding the answer. The `ordering` module has what is measured and how.
576 ///
577 /// A top level `OR` is threaded the same way against the complement. A row the first branch
578 /// accepts is a row the filter keeps whatever the rest of the predicate says about it, so each
579 /// branch is run over the rows no branch before it accepted, and the moment every row has been
580 /// accepted the rest of the predicate is not run either. That is the mirror of the `AND` case
581 /// and not an approximation of it: the answer is the same set of rows, because `OR` over three
582 /// valued logic is true wherever any branch is true and nothing a later branch says can take a
583 /// row back. It is worth less than the `AND` case in practice, since an `OR` of selective
584 /// branches leaves almost every row in play for the branch after, and it is worth having anyway
585 /// because the cost of finding that out is one merge per branch.
586 ///
587 /// What is threaded is the operand's own comparison rather than the whole of its subtree. A
588 /// conjunct of `a + b > 5` still adds over the whole chunk, because the scalar kernels take a
589 /// vector rather than a selection, and it is the comparison and everything downstream of it that
590 /// reads only the rows still in play. An operand that is a bare column or a function produces
591 /// flags over the chunk and is narrowed with [`refine_flags`], which is what keeps one awkward
592 /// operand from putting the others back on the unthreaded path. An operand that is itself a
593 /// connective recurses, so the two conjuncts of each half of `(a AND b) OR (c AND d)` are
594 /// threaded the same way the halves are.
595 ///
596 /// None of this is available to a projection. `SELECT a > 5 AND b LIKE 'x%'` wants a value per
597 /// row and the rows a selection dropped have no value in it, so [`evaluate`](Self::evaluate) and
598 /// [`evaluate_one`](Self::evaluate_one) evaluate the whole tree over the whole chunk and combine
599 /// flags. The two are separate entry points picked when the pipeline is built rather than one
600 /// path with a flag in it, because conflating them is a wrong answer rather than a slow one.
601 ///
602 /// # Errors
603 ///
604 /// Anything a kernel reports, and an internal error if this was not built from exactly one
605 /// expression.
606 pub fn evaluate_filter(&self, chunk: &Chunk, scratch: &mut Scratch) -> Result<Selection> {
607 let [root] = self.roots[..] else {
608 return Err(Error::internal(format!(
609 "evaluate_filter over a prepared expression of {} roots",
610 self.roots.len()
611 )));
612 };
613 scratch.slots.clear();
614 scratch.slots.resize_with(self.steps.len(), || None);
615 // A predicate that is not a connective at all is the same walk over one operand, which is
616 // where [`thread`](Self::thread) starts: it runs the tree and turns the flags into a
617 // selection, with no narrowing to do because nothing has narrowed anything yet.
618 self.thread(root, 0, chunk, scratch, None)
619 }
620
621 /// The operands of one connective, run in order, each over the rows the ones before it left.
622 ///
623 /// `live` is the rows this connective has to decide about and `None` means every row of the
624 /// chunk, which is not the same as a selection of all of them: it lets the first operand take
625 /// the unthreaded kernel rather than a pass over an identity selection. The answer is the rows
626 /// out of `live` the connective is true for.
627 ///
628 /// The walk is the same for both connectives and only the bookkeeping differs. `AND` carries the
629 /// rows every operand so far has kept, so each answer replaces it. `OR` carries the rows no
630 /// operand so far has accepted, so each answer comes out of it and the rows the connective keeps
631 /// are the ones that went missing along the way.
632 ///
633 /// The operand is not `steps[begin..=operand]` evaluated and then narrowed. Its subtree is run
634 /// over the whole chunk and it is the operand itself that reads only the rows in play, except
635 /// where the operand is another connective, which recurses and threads its own operands from
636 /// here rather than falling back to a flag vector. That is what makes `(a AND b) OR (c AND d)`
637 /// four threaded comparisons rather than two threaded ones and two flag passes.
638 fn branches(
639 &self,
640 index: usize,
641 begin: usize,
642 chunk: &Chunk,
643 scratch: &mut Scratch,
644 live: Option<&Selection>,
645 ) -> Result<Selection> {
646 let Step::Conjunction { op, start, len } = self.steps[index] else {
647 return Err(Error::internal("a connective walk over a step that is not a connective"));
648 };
649 let operands = &self.operands[start..start + len];
650 let rows = chunk.len();
651 // Out of the scratch for the length of the walk, because the walk runs steps and running a
652 // step wants the scratch. It goes back at the end, which is also where it learns. A walk
653 // that fails leaves the slot empty and the next chunk starts the connective over, which is
654 // a history lost on a query that is about to stop running anyway.
655 let mut order = scratch.orders[index]
656 .take()
657 .unwrap_or_else(|| Ordering::new(op, self.weights(operands, begin)));
658 let mut carried: Option<Selection> = live.cloned();
659 for slot in 0..len {
660 if carried.as_ref().is_some_and(Selection::is_empty) {
661 break;
662 }
663 let which = order.at(slot);
664 let operand = operands[which];
665 // The array is in post order and an operand's whole subtree sits between the operand
666 // before it and the operand itself, which is a range the run order cannot move. That is
667 // what lets the operands run in any order at all without a second structure to say
668 // where each one starts.
669 let from = if which == 0 { begin } else { operands[which - 1] + 1 };
670 let given = carried.as_ref().map_or(rows, Selection::len);
671 let answered = self.thread(operand, from, chunk, scratch, carried.as_ref())?;
672 order.observed(which, given, answered.len());
673 carried = Some(match (op, carried) {
674 (Connective::And, _) => answered,
675 (Connective::Or, None) => answered.complement(rows),
676 (Connective::Or, Some(carried)) => carried.without(&answered),
677 });
678 // Keep a shared step alive when a later operand still reads it.
679 for step in from..=operand {
680 if self.last_use[step] <= operand {
681 scratch.slots[step] = None;
682 }
683 }
684 }
685 order.relearn();
686 scratch.orders[index] = Some(order);
687 Ok(match (op, carried) {
688 // A connective with no operands, which the binder does not build and which is answered
689 // here rather than left to index arithmetic: an empty `AND` is every row and an empty
690 // `OR` is none.
691 (Connective::And, None) => live.cloned().unwrap_or_else(|| Selection::identity(rows)),
692 (Connective::And, Some(kept)) => kept,
693 (Connective::Or, None) => Selection::empty(),
694 (Connective::Or, Some(missed)) => match live {
695 None => missed.complement(rows),
696 Some(live) => live.without(&missed),
697 },
698 })
699 }
700
701 /// What each operand of a connective costs to run over a chunk, for the ordering to divide by.
702 ///
703 /// An operand costs what its whole subtree costs, which is the steps from where the operand
704 /// before it ended up to the operand itself.
705 fn weights(&self, operands: &[usize], begin: usize) -> Vec<f64> {
706 let mut costs = Vec::with_capacity(operands.len());
707 let mut from = begin;
708 for &operand in operands {
709 costs.push((from..=operand).map(|step| self.weight(step)).sum());
710 from = operand + 1;
711 }
712 costs
713 }
714
715 /// Roughly what one step costs to run over a chunk, against a comparison of two fixed width
716 /// columns as the unit.
717 ///
718 /// A ranking rather than a prediction. Nothing downstream reads the number itself, only which
719 /// of two of them is larger, and the differences that decide an order are the big ones: a
720 /// column reference costs nothing because it is read in place, a string function costs many
721 /// times what an integer comparison costs, and a comparison over a variable length type costs
722 /// several times what the same comparison over a fixed width one costs. Everything finer than
723 /// that is below the noise of what the window is measuring anyway.
724 fn weight(&self, index: usize) -> f64 {
725 match &self.steps[index] {
726 // Read straight out of the chunk at the point an operand is wanted, so there is no step
727 // to run and nothing to charge for.
728 Step::Column(_) => 0.0,
729 // One vector built per chunk, however many rows the chunk has.
730 Step::Constant(_) => 0.25,
731 // The operands carry the cost of a connective, and they are steps of their own.
732 Step::Conjunction { .. } => 0.0,
733 Step::Cast { input, .. } => 2.0 * touching(&self.types[*input]),
734 Step::Compare { left, .. } => touching(&self.types[*left]),
735 // One hash and one probe a row, whatever the list holds, which is the point of it. It
736 // is dearer than a comparison and much cheaper than the chain of them it replaced.
737 Step::InSet { input, .. } => 2.0 * touching(&self.types[*input]),
738 Step::Function { start, len, .. } => {
739 let widest = self.operands[*start..*start + *len]
740 .iter()
741 .map(|&argument| touching(&self.types[argument]))
742 .fold(1.0, f64::max);
743 4.0 * widest
744 }
745 // A branch per arm, each of which is a prepared expression of its own that this does
746 // not look inside. Charging for the arms alone understates it and says the right thing
747 // about the order, which is that a `CASE` is not what you want in front.
748 Step::Case { arms, .. } => 4.0 * arms.len() as f64,
749 // A run of the body per element, which is several a row, and a list to take apart and
750 // put back together around it.
751 Step::Lambda { .. } => 16.0,
752 // An integer operation a row per node and no check, which is a quarter of what the
753 // function steps it replaced cost each.
754 Step::Fused { fused, .. } => fused.len() as f64,
755 }
756 }
757
758 /// One operand of a connective, over the rows it is still worth asking about.
759 ///
760 /// `begin` is the first step of the operand's subtree, which the caller knows because the steps
761 /// are in post order.
762 fn thread(
763 &self,
764 index: usize,
765 begin: usize,
766 chunk: &Chunk,
767 scratch: &mut Scratch,
768 live: Option<&Selection>,
769 ) -> Result<Selection> {
770 if matches!(self.steps[index], Step::Conjunction { .. }) {
771 return self.branches(index, begin, chunk, scratch, live);
772 }
773 for step in begin..index {
774 self.run_step(step, chunk, scratch)?;
775 }
776 if let Step::Compare { op, left, right, held } = &self.steps[index] {
777 let one = self.operand(*left, chunk, &scratch.slots)?;
778 let other = self.operand(*right, chunk, &scratch.slots)?;
779 let held = held.as_ref();
780 return match live {
781 // The first operand has every row in play, and asking the threaded kernel for that
782 // would be a pass over an identity selection the unthreaded one does not need.
783 None => select_prepared(*op, one, other, held),
784 Some(live) => refine_prepared(*op, one, other, live, held),
785 };
786 }
787 // A later LIKE in a threaded filter often sees only a handful of survivors.
788 // Gather its arguments, not the whole chunk, while preserving the stable
789 // dictionary behind a gathered string column. The ordinary full-vector
790 // path remains cheaper when most rows are still live.
791 if let (Some(live), Step::Function { recipe, written, start, len }) =
792 (live, &self.steps[index])
793 {
794 if matches!(recipe.name(), "~~" | "!~~" | "~~*" | "!~~*")
795 && live.len().saturating_mul(4) <= chunk.len()
796 {
797 let flags = self
798 .with_operands(*start, *len, chunk, &scratch.slots, |args| {
799 let gathered = args
800 .iter()
801 .map(|arg| arg.gather(live.indices()))
802 .collect::<Result<Vec<_>>>()?;
803 let narrowed = gathered.iter().collect::<Vec<_>>();
804 rudb_kernels::call_prepared(
805 recipe,
806 &narrowed,
807 &self.types[index],
808 Some(&|| written.clone()),
809 )
810 })
811 .map_err(|error| error.with_fallback_span(self.spans[index]))?;
812 return Ok(selection(&flags, live.len()).compose(live));
813 }
814 }
815 self.run_step(index, chunk, scratch)?;
816 let flags = self.operand(index, chunk, &scratch.slots)?;
817 match live {
818 None => Ok(selection(flags, chunk.len())),
819 Some(live) => refine_flags(flags, live),
820 }
821 }
822
823 /// Runs every step in order, filling the slots.
824 fn run(&self, chunk: &Chunk, scratch: &mut Scratch) -> Result<()> {
825 scratch.slots.clear();
826 scratch.slots.resize_with(self.steps.len(), || None);
827 for index in 0..self.steps.len() {
828 self.run_step(index, chunk, scratch)?;
829 }
830 Ok(())
831 }
832
833 /// Runs one step and empties the slot of every operand this was the last step to read.
834 fn run_step(&self, index: usize, chunk: &Chunk, scratch: &mut Scratch) -> Result<()> {
835 let produced = self
836 .step(index, chunk, &scratch.slots)
837 .map_err(|error| error.with_fallback_span(self.spans[index]))?;
838 scratch.slots[index] = produced;
839 let slots = &mut scratch.slots;
840 self.for_each_operand(index, |operand| {
841 if self.last_use[operand] == index {
842 slots[operand] = None;
843 }
844 });
845 Ok(())
846 }
847
848 /// Runs one step, given what the steps before it produced.
849 fn step(
850 &self,
851 index: usize,
852 chunk: &Chunk,
853 slots: &[Option<Vector>],
854 ) -> Result<Option<Vector>> {
855 let ty = &self.types[index];
856 let produced = match &self.steps[index] {
857 Step::Column(_) => None,
858 Step::Constant(value) => Some(Vector::constant(ty.clone(), value.clone(), chunk.len())),
859 Step::Cast { input, try_cast } => Some(cast_in_time_zone(
860 self.operand(*input, chunk, slots)?,
861 ty,
862 *try_cast,
863 Some(self.time_zone),
864 )?),
865 Step::Compare { op, left, right, held } => Some(compare_prepared(
866 *op,
867 self.operand(*left, chunk, slots)?,
868 self.operand(*right, chunk, slots)?,
869 held.as_ref(),
870 )?),
871 Step::Conjunction { op, start, len } => {
872 Some(
873 self.with_operands(*start, *len, chunk, slots, |children| {
874 combine(*op, children)
875 })?,
876 )
877 }
878 Step::Function { recipe, written, start, len } => {
879 Some(self.with_operands(*start, *len, chunk, slots, |args| {
880 rudb_kernels::call_prepared(recipe, args, ty, Some(&|| written.clone()))
881 })?)
882 }
883 Step::InSet { input, members } => {
884 Some(in_set(self.operand(*input, chunk, slots)?, members, ty)?)
885 }
886 Step::Case { arms, otherwise, blend } => {
887 Some(self.case(chunk, arms, otherwise.as_ref(), blend.as_ref(), ty)?)
888 }
889 Step::Fused { fused, fallback } => Some(match fused.run(chunk) {
890 Some(answer) => answer,
891 None => fallback.evaluate_one(chunk, &mut fallback.scratch())?.clone(),
892 }),
893 Step::Lambda { inputs, runner, body } => {
894 let mut operands = Vec::with_capacity(inputs.len());
895 for &input in inputs {
896 operands.push(self.operand(input, chunk, slots)?);
897 }
898 let mut scratch = body.scratch();
899 Some(runner.run(&operands, chunk, &mut |inner| {
900 body.evaluate_one(inner, &mut scratch).cloned()
901 })?)
902 }
903 };
904 Ok(produced)
905 }
906
907 /// The vector a step produced, or the chunk's column if the step is a column reference.
908 fn operand<'v>(
909 &self,
910 index: usize,
911 chunk: &'v Chunk,
912 slots: &'v [Option<Vector>],
913 ) -> Result<&'v Vector> {
914 if let Step::Column(position) = self.steps[index] {
915 return chunk.column(position);
916 }
917 slots[index].as_ref().ok_or_else(|| missing(index))
918 }
919
920 /// Hands a kernel the references to an operand list, without allocating for the usual widths.
921 ///
922 /// One, two and three because those are what a bound tree is made of: every scalar function in
923 /// the catalog is unary or binary, a comparison is binary, and a conjunction is two or three
924 /// often enough to be worth a line. A stack array for those means a chain of eight additions
925 /// makes zero allocations for its operand lists over a chunk instead of eight, and eight
926 /// allocations a chunk at the rate a pipeline produces chunks is a real number rather than a
927 /// tidiness argument. Anything wider falls back to [`gather`](Self::gather), which is a `Vec`
928 /// of pointers and still moves no data.
929 fn with_operands<'v, T>(
930 &self,
931 start: usize,
932 len: usize,
933 chunk: &'v Chunk,
934 slots: &'v [Option<Vector>],
935 run: impl FnOnce(&[&'v Vector]) -> Result<T>,
936 ) -> Result<T> {
937 match self.operands[start..start + len] {
938 [a] => run(&[self.operand(a, chunk, slots)?]),
939 [a, b] => run(&[self.operand(a, chunk, slots)?, self.operand(b, chunk, slots)?]),
940 [a, b, c] => run(&[
941 self.operand(a, chunk, slots)?,
942 self.operand(b, chunk, slots)?,
943 self.operand(c, chunk, slots)?,
944 ]),
945 _ => {
946 let gathered = self.gather(start, len, chunk, slots)?;
947 run(&gathered)
948 }
949 }
950 }
951
952 /// References to an operand list, for a kernel that takes a slice of them.
953 ///
954 /// The `Vec` here is the allocation the module documentation names: it holds pointers rather
955 /// than vectors, so it is a dozen bytes an operand and no data moves.
956 fn gather<'v>(
957 &self,
958 start: usize,
959 len: usize,
960 chunk: &'v Chunk,
961 slots: &'v [Option<Vector>],
962 ) -> Result<Vec<&'v Vector>> {
963 let mut gathered = Vec::with_capacity(len);
964 for &operand in &self.operands[start..start + len] {
965 gathered.push(self.operand(operand, chunk, slots)?);
966 }
967 Ok(gathered)
968 }
969
970 /// A searched `CASE` over the rows no earlier arm claimed.
971 ///
972 /// The same shape [`evaluate`](crate::evaluate) has, because the thing that makes it that shape
973 /// is a correctness rule rather than a performance one: `CASE WHEN x <> 0 THEN 1 / x ELSE 0 END`
974 /// divides by zero on the rows the arm excludes if the arm is evaluated for them.
975 ///
976 /// Each arm answers the rows no earlier arm claimed, so the answers come back short and out of
977 /// order and have to be put back in the order the rows arrived in. That is what [`Assembly`] is:
978 /// the arms are laid end to end into one run of data and the interleave is a single typed copy
979 /// over it. It used to be a `Vec<Value>` filled a row at a time and handed to
980 /// `Vector::from_values`, which is a heap allocation and a drop for every string in the answer.
981 /// On the ClickBench query that groups by a `CASE` over `Referer` that was about a quarter of
982 /// the whole query.
983 ///
984 /// What is left of #57 here is the narrowing. An arm still narrows the whole chunk rather than
985 /// the columns it reads, and the selection threading that replaces the narrowing entirely is
986 /// the item this one was carved out of.
987 fn case(
988 &self,
989 chunk: &Chunk,
990 arms: &[PreparedArm],
991 otherwise: Option<&Prepared>,
992 blend: Option<&Blend>,
993 ty: &LogicalType,
994 ) -> Result<Vector> {
995 let claimed = self.claims(chunk, arms)?;
996 if let Some(blend) = blend {
997 if let Some(blended) = blended(chunk, &claimed, blend)? {
998 return Ok(blended);
999 }
1000 }
1001 let mut built = Assembly::new(ty.clone(), chunk.len())?;
1002 let branches = arms.iter().map(|arm| &arm.then).map(Some).chain([otherwise]);
1003 for (branch, rows) in branches.zip(&claimed) {
1004 let (Some(branch), false) = (branch, rows.is_empty()) else { continue };
1005 // The same cut the conditions skip above, skipped here for the same reason: a branch
1006 // that claimed every row claimed them in order, so narrowing to them is a copy of every
1007 // column in the chunk to arrive back at the chunk.
1008 let cut;
1009 let matched = if rows.len() == chunk.len() {
1010 chunk
1011 } else {
1012 cut = narrow(chunk, rows)?;
1013 &cut
1014 };
1015 let mut scratch = branch.scratch();
1016 let results = branch.evaluate_one(matched, &mut scratch)?;
1017 built.place(&placed(rows)?, results)?;
1018 }
1019 built.finish()
1020 }
1021
1022 /// The rows each branch of a `CASE` answers, one list per arm in order and the `ELSE` last.
1023 ///
1024 /// Only the conditions are run here, which is what keeps the rule the doc above states: an arm's
1025 /// condition is evaluated over the rows no earlier arm claimed, so a condition that would raise
1026 /// on a row an earlier arm took is never asked about it. The results are worked out afterwards,
1027 /// once, from these lists, and both ways of working them out want the same thing, which is the
1028 /// rows of one branch in the order they arrived in.
1029 fn claims(&self, chunk: &Chunk, arms: &[PreparedArm]) -> Result<Vec<Vec<usize>>> {
1030 let mut claimed = Vec::with_capacity(arms.len() + 1);
1031 let mut pending: Vec<usize> = (0..chunk.len()).collect();
1032 for arm in arms {
1033 if pending.is_empty() {
1034 claimed.push(Vec::new());
1035 continue;
1036 }
1037 // `pending` starts as every row in order and only ever shrinks, so the same length is
1038 // the same rows in the same order and there is nothing to cut. That is the whole of the
1039 // first arm of a one armed `CASE`, which is the shape of the ClickBench query this was
1040 // measured on, and cutting it was a copy of every column in the chunk for nothing.
1041 let cut;
1042 let narrowed = if pending.len() == chunk.len() {
1043 chunk
1044 } else {
1045 cut = narrow(chunk, &pending)?;
1046 &cut
1047 };
1048 let mut scratch = arm.when.scratch();
1049 let flags = arm.when.evaluate_one(narrowed, &mut scratch)?;
1050 let mut taken = Vec::new();
1051 let mut still = Vec::new();
1052 // row at a time: splitting the rows an arm claims from the ones it leaves is a test per
1053 // row, and what replaces it is the selection threading the rest of #57 asks for rather
1054 // than anything that can be done here.
1055 for (at, &row) in pending.iter().enumerate() {
1056 if is_true(&flags.value_at(at)) {
1057 taken.push(row);
1058 } else {
1059 still.push(row);
1060 }
1061 }
1062 claimed.push(taken);
1063 pending = still;
1064 }
1065 claimed.push(pending);
1066 Ok(claimed)
1067 }
1068
1069 /// Flattens one expression, appending its steps and returning the index of its last one.
1070 fn push(&mut self, plan: &Plan, expr: ExprRef, schema: &Schema) -> Result<usize> {
1071 if self.share {
1072 if let Some(&step) = self.shared.get(&expr) {
1073 return Ok(step);
1074 }
1075 }
1076 let ty = plan.expr_type(expr).clone();
1077 if self.fuse {
1078 if let Some(fused) = Fused::compile(plan, expr, schema) {
1079 let fallback = Self::built(plan, &[expr], schema, false, false)?;
1080 let step = Step::Fused { fused: Box::new(fused), fallback: Box::new(fallback) };
1081 return Ok(self.place(plan, expr, step, ty));
1082 }
1083 }
1084 if let Some((stamp, count)) = stamped_seconds(plan, expr) {
1085 let (start, len) = self.push_list(plan, &[stamp, count], schema)?;
1086 let step = Step::Function {
1087 recipe: Recipe::new("__rudb_stamp_seconds", &self.literals(start, len)),
1088 written: written(plan, expr, schema),
1089 start,
1090 len,
1091 };
1092 return Ok(self.place(plan, expr, step, ty));
1093 }
1094 let step = match *plan.expr(expr) {
1095 Expr::Column(binding) => {
1096 let position = schema.position_of(binding).ok_or_else(|| {
1097 Error::internal(format!(
1098 "column #{}.{} is not in the schema this operator was given",
1099 binding.table, binding.column
1100 ))
1101 })?;
1102 Step::Column(position)
1103 }
1104 Expr::Constant(reference) => Step::Constant(plan.value(reference).clone()),
1105 Expr::Cast { input, try_cast } => {
1106 Step::Cast { input: self.push(plan, input, schema)?, try_cast }
1107 }
1108 Expr::Compare { op, left, right } => {
1109 let left = self.push(plan, left, schema)?;
1110 let right = self.push(plan, right, schema)?;
1111 Step::Compare { op: comparison(op), left, right, held: self.held(left, right) }
1112 }
1113 Expr::Conjunction { op, children } => {
1114 let list = plan.expr_list(children).to_vec();
1115 match self.membership(plan, connective(op), &list, schema)? {
1116 Some(step) => step,
1117 None => {
1118 let (start, len) = self.push_list(plan, &list, schema)?;
1119 Step::Conjunction { op: connective(op), start, len }
1120 }
1121 }
1122 }
1123 Expr::Function { name, args } if lambda_call(plan, args).is_some() => {
1124 let Some((lambda, inputs)) = lambda_call(plan, args) else {
1125 return Err(Error::internal("a lambda call without a lambda"));
1126 };
1127 let Expr::Lambda { body, .. } = *plan.expr(lambda) else {
1128 return Err(Error::internal("a lambda call without a lambda"));
1129 };
1130 let runner = Lambda::new(plan, plan.string(name), lambda, &inputs, schema)?;
1131 let body = Self::one(plan, body, runner.schema())?;
1132 let mut steps = Vec::with_capacity(inputs.len());
1133 for &input in &inputs {
1134 steps.push(self.push(plan, input, schema)?);
1135 }
1136 Step::Lambda { inputs: steps, runner: Box::new(runner), body: Box::new(body) }
1137 }
1138 Expr::LambdaParam(binding) => {
1139 let position = schema.position_of(binding).ok_or_else(|| {
1140 Error::internal(format!(
1141 "lambda parameter @{}.{} is not in the schema its body was given",
1142 binding.table, binding.column
1143 ))
1144 })?;
1145 Step::Column(position)
1146 }
1147 Expr::Lambda { .. } => {
1148 return Err(Error::internal(
1149 "a lambda was evaluated outside the function that takes it",
1150 ));
1151 }
1152 Expr::Function { name, args } => {
1153 let (start, len) = self.push_list(plan, plan.expr_list(args), schema)?;
1154 Step::Function {
1155 recipe: Recipe::new(plan.string(name), &self.literals(start, len)),
1156 written: written(plan, expr, schema),
1157 start,
1158 len,
1159 }
1160 }
1161 Expr::Aggregate { name, .. } => {
1162 return Err(Error::internal(format!(
1163 "the {} aggregate was evaluated as an ordinary expression",
1164 plan.string(name)
1165 )));
1166 }
1167 Expr::Window { name, .. } => {
1168 return Err(Error::internal(format!(
1169 "the {} window function was evaluated as an ordinary expression",
1170 plan.string(name)
1171 )));
1172 }
1173 Expr::Case { arms, otherwise } => {
1174 let mut prepared = Vec::new();
1175 for &arm in plan.arm_list(arms) {
1176 prepared.push(PreparedArm {
1177 when: Self::one(plan, arm.when, schema)?,
1178 then: Self::one(plan, arm.then, schema)?,
1179 });
1180 }
1181 let otherwise = match otherwise {
1182 Some(otherwise) => Some(Self::one(plan, otherwise, schema)?),
1183 None => None,
1184 };
1185 let blend = blending(&ty, &prepared, otherwise.as_ref());
1186 Step::Case { arms: prepared, otherwise, blend }
1187 }
1188 };
1189 Ok(self.place(plan, expr, step, ty))
1190 }
1191
1192 /// Appends a built step and answers its index.
1193 fn place(&mut self, plan: &Plan, expr: ExprRef, step: Step, ty: LogicalType) -> usize {
1194 self.steps.push(step);
1195 self.types.push(ty);
1196 self.spans.push(plan.expr_span(expr));
1197 let step = self.steps.len() - 1;
1198 if self.share {
1199 self.shared.insert(expr, step);
1200 }
1201 step
1202 }
1203
1204 /// Flattens a list of expressions and records where its operand run starts and how long it is.
1205 ///
1206 /// The operand run is written after every child has been flattened rather than as they go,
1207 /// because a child that is itself a list would otherwise interleave its run with this one.
1208 fn push_list(
1209 &mut self,
1210 plan: &Plan,
1211 exprs: &[ExprRef],
1212 schema: &Schema,
1213 ) -> Result<(usize, usize)> {
1214 let mut indices = Vec::with_capacity(exprs.len());
1215 for &expr in exprs {
1216 indices.push(self.push(plan, expr, schema)?);
1217 }
1218 let start = self.operands.len();
1219 let len = indices.len();
1220 self.operands.extend(indices);
1221 Ok((start, len))
1222 }
1223
1224 /// This connective folded back into the `IN` the user wrote, or `None` when it is not one.
1225 ///
1226 /// What the binder writes for `x IN (1, 2, 3)` is `x = 1 OR x = 2 OR x = 3`, and for
1227 /// `x NOT IN (1, 2, 3)` it is `x <> 1 AND x <> 2 AND x <> 3`. So the shape looked for is every
1228 /// child a comparison of the one direction, every left the same expression, and every right a
1229 /// literal. Anything else is left alone, which covers the `OR` that was written as an `OR` and
1230 /// the one where an `IN` has been flattened together with another branch. The second is a fold
1231 /// this could make and does not, and it is worth having later out of a query that wants it
1232 /// rather than now out of a guess.
1233 ///
1234 /// This runs before the children are pushed, and that is the whole reason it is here rather than
1235 /// as a pass over the finished array. A step that nothing reads is still a step the walk runs,
1236 /// because the walk over a subtree is a range and not a graph, so folding after the fact would
1237 /// leave every equality in place and running.
1238 fn membership(
1239 &mut self,
1240 plan: &Plan,
1241 op: Connective,
1242 children: &[ExprRef],
1243 schema: &Schema,
1244 ) -> Result<Option<Step>> {
1245 let wanted = match op {
1246 Connective::Or => CompareOp::Equal,
1247 Connective::And => CompareOp::NotEqual,
1248 };
1249 let mut subject: Option<ExprRef> = None;
1250 let mut values = Vec::with_capacity(children.len());
1251 for &child in children {
1252 let Expr::Compare { op: found, left, right } = *plan.expr(child) else {
1253 return Ok(None);
1254 };
1255 if found != wanted || !same(plan, *subject.get_or_insert(left), left) {
1256 return Ok(None);
1257 }
1258 let Expr::Constant(reference) = *plan.expr(right) else {
1259 return Ok(None);
1260 };
1261 values.push(plan.value(reference).clone());
1262 }
1263 let (Some(subject), Some(members)) = (subject, Members::of(&values, op == Connective::And))
1264 else {
1265 return Ok(None);
1266 };
1267 Ok(Some(Step::InSet { input: self.push(plan, subject, schema)?, members }))
1268 }
1269
1270 /// The literal side of a comparison, in the one row column the comparison reads it through.
1271 ///
1272 /// The right side first, because that is the side the binder puts a literal on and the side the
1273 /// loops are written for. Two literals is a comparison the optimizer folded, and if it did not
1274 /// then the kernel answers it once for the whole vector and never reads either column, so
1275 /// neither side is built here.
1276 fn held(&self, left: usize, right: usize) -> Option<Held> {
1277 let (at, other) = match (&self.steps[left], &self.steps[right]) {
1278 (Step::Constant(_), Step::Constant(_)) => return None,
1279 (_, Step::Constant(value)) => (right, value),
1280 (Step::Constant(value), _) => (left, value),
1281 _ => return None,
1282 };
1283 Held::of(&self.types[at], other)
1284 }
1285
1286 /// The literal behind each argument in a run of the operand list, and `None` for an argument
1287 /// that is anything else.
1288 ///
1289 /// This is what a [`Recipe`] hoists from. An argument that is a literal in the plan arrives as a
1290 /// constant vector holding exactly this value on every chunk, so what a kernel reads here is
1291 /// what it would have read per chunk. An argument that is a cast of a literal reads as `None`,
1292 /// which is a call the kernel decides per chunk as it always did, and the optimizer folds most
1293 /// of those before the plan gets here anyway.
1294 fn literals(&self, start: usize, len: usize) -> Vec<Option<Value>> {
1295 self.operands[start..start + len]
1296 .iter()
1297 .map(|&operand| match &self.steps[operand] {
1298 Step::Constant(value) => Some(value.clone()),
1299 _ => None,
1300 })
1301 .collect()
1302 }
1303}
1304
1305/// Whether two expressions of one plan are the same expression, written once or written twice.
1306///
1307/// The binder binds the subject of an `IN` once and points every comparison it writes at that one
1308/// reference, so the answer is almost always the first line. A plan that has been through a rewrite,
1309/// and a plan read back from its own text, hold two copies of the same tree instead, and for the
1310/// fold in [`Prepared::membership`] those are the same expression.
1311///
1312/// The four shapes handled are what an `IN` is written over: a column, a literal, a cast of either,
1313/// and a call, which is TPC-H query 22 asking whether the first two digits of a phone number are in
1314/// a list. Anything else answers no, which costs a fold that could have happened rather than a wrong
1315/// one. The walk is bounded by the size of the subject and a subject is small.
1316/// The timestamp and the whole count of `stamp + to_seconds(CAST(count AS DOUBLE))`, the shape the
1317/// benchmark view writes `INTERVAL (EventTime) SECOND` in, and `None` for anything else.
1318///
1319/// It runs as one call, [`rudb_kernels`]'s `__rudb_stamp_seconds`, rather than as a cast to a
1320/// double, an interval per row and a shift by it.
1321fn stamped_seconds(plan: &Plan, expr: ExprRef) -> Option<(ExprRef, ExprRef)> {
1322 let Expr::Function { name, args } = *plan.expr(expr) else { return None };
1323 if plan.string(name) != "+" || plan.expr_type(expr) != &LogicalType::Timestamp {
1324 return None;
1325 }
1326 let &[one, other] = plan.expr_list(args) else { return None };
1327 let (stamp, interval) =
1328 if plan.expr_type(one) == &LogicalType::Timestamp { (one, other) } else { (other, one) };
1329 if plan.expr_type(stamp) != &LogicalType::Timestamp {
1330 return None;
1331 }
1332 let Expr::Function { name, args } = *plan.expr(interval) else { return None };
1333 let &[cast] = plan.expr_list(args) else { return None };
1334 let Expr::Cast { input, try_cast: false } = *plan.expr(cast) else { return None };
1335 let whole = matches!(
1336 plan.expr_type(input),
1337 LogicalType::TinyInt
1338 | LogicalType::SmallInt
1339 | LogicalType::Integer
1340 | LogicalType::BigInt
1341 | LogicalType::UTinyInt
1342 | LogicalType::USmallInt
1343 | LogicalType::UInteger
1344 );
1345 (plan.string(name) == "to_seconds" && plan.expr_type(cast) == &LogicalType::Double && whole)
1346 .then_some((stamp, input))
1347}
1348
1349fn same(plan: &Plan, left: ExprRef, right: ExprRef) -> bool {
1350 if left == right {
1351 return true;
1352 }
1353 if plan.expr_type(left) != plan.expr_type(right) {
1354 return false;
1355 }
1356 match (plan.expr(left), plan.expr(right)) {
1357 (Expr::Column(one), Expr::Column(other)) => one == other,
1358 (Expr::Constant(one), Expr::Constant(other)) => plan.value(*one) == plan.value(*other),
1359 (
1360 Expr::Cast { input: one, try_cast: first },
1361 Expr::Cast { input: other, try_cast: second },
1362 ) => first == second && same(plan, *one, *other),
1363 (
1364 Expr::Function { name: one, args: first },
1365 Expr::Function { name: other, args: second },
1366 ) => {
1367 let (first, second) = (plan.expr_list(*first), plan.expr_list(*second));
1368 plan.string(*one) == plan.string(*other)
1369 && first.len() == second.len()
1370 && first.iter().zip(second).all(|(&one, &other)| same(plan, one, other))
1371 }
1372 _ => false,
1373 }
1374}
1375
1376/// What touching a value of this type costs, against a fixed width one as the unit.
1377///
1378/// A variable length value is a pointer to follow and a length that is not the same twice, and a
1379/// nested one is that per element. Four is not measured, and what it has to be is large enough that
1380/// the ordering puts a fixed width comparison in front of a string one and small enough that it does
1381/// not put one in front of a string comparison that rejects every row.
1382fn touching(ty: &LogicalType) -> f64 {
1383 match ty.physical() {
1384 PhysicalType::Varlen => 4.0,
1385 PhysicalType::List | PhysicalType::Array | PhysicalType::Struct => 8.0,
1386 _ => 1.0,
1387 }
1388}
1389
1390/// The error for a slot that should have held something and did not.
1391///
1392/// This cannot happen while the array is in post order, since every operand's index is smaller than
1393/// the index of the step using it and every step runs in order. It is an error rather than a panic
1394/// because the property it depends on is a property of [`Prepared::push`], and the day somebody
1395/// writes a pass that reorders the array is the day it stops holding.
1396fn missing(index: usize) -> Error {
1397 Error::internal(format!("step {index} was used as an operand before it produced anything"))
1398}
1399
1400/// Chunk rows as the positions an [`Assembly`] places a piece at.
1401///
1402/// A chunk is at most [`VECTOR_SIZE`](rudb_vector::VECTOR_SIZE) rows, so the conversion cannot fail
1403/// in practice. It is checked rather than cast because a silent truncation here would put a value in
1404/// the wrong row, and a wrong row is the one kind of bug nothing downstream can notice.
1405fn placed(rows: &[usize]) -> Result<Vec<u32>> {
1406 rows.iter()
1407 .map(|&row| {
1408 u32::try_from(row).map_err(|_| Error::internal("a chunk of more than u32 rows"))
1409 })
1410 .collect()
1411}
1412
1413/// A `CASE` answered as codes over the dictionary its branches share, or `None` for a chunk that
1414/// cannot be.
1415///
1416/// Declined per chunk rather than once, because whether a column arrives coded is a fact about the
1417/// chunk and not about the expression. The same query reads codes out of a native file and plain
1418/// strings out of rows held in memory, and one file can hand a column over as a dictionary in one
1419/// part and as plain data in the next. Everything that declines does so before a code is written, so
1420/// the caller starts the general path from nothing rather than from a half filled answer.
1421fn blended(chunk: &Chunk, claimed: &[Vec<usize>], blend: &Blend) -> Result<Option<Vector>> {
1422 let Some((dictionary, literals)) = agreed(chunk, blend)? else { return Ok(None) };
1423 let mut codes = vec![0; chunk.len()];
1424 for (branch, rows) in blend.branches.iter().zip(claimed) {
1425 match *branch {
1426 Branch::Column(position) => {
1427 let Some((from, _)) = chunk.column(position)?.stable_dictionary_parts() else {
1428 return Ok(None);
1429 };
1430 for &row in rows {
1431 codes[row] = from[row];
1432 }
1433 }
1434 Branch::Literal(at) => {
1435 for &row in rows {
1436 codes[row] = literals[at];
1437 }
1438 }
1439 }
1440 }
1441 Vector::stable_dictionary(codes, dictionary).map(Some)
1442}
1443
1444/// The one dictionary every branch of a blend names values in, and the code each literal sits at.
1445///
1446/// Three things say no. A column that did not arrive as a stable dictionary has no codes to copy. A
1447/// second column over a different dictionary would have codes that mean something else, and a code
1448/// is a position in one dictionary and nothing anywhere else. And a literal the dictionary does not
1449/// hold has no code at all, which for `ELSE ''` over a column where no row is empty is the honest
1450/// answer rather than a missing one.
1451///
1452/// The null check is the fourth. A dictionary keeps its nulls in the values it points at rather than
1453/// beside its codes, so a column carrying its own validity is one whose codes do not say everything
1454/// the column says, and copying them would turn its nulls into whatever their codes happen to name.
1455fn agreed(chunk: &Chunk, blend: &Blend) -> Result<Option<(Arc<Vector>, Vec<u32>)>> {
1456 let mut held: Option<(&Vector, &Arc<Vector>)> = None;
1457 for branch in &blend.branches {
1458 let Branch::Column(position) = *branch else { continue };
1459 let column = chunk.column(position)?;
1460 let Some((_, dictionary)) = column.stable_dictionary_parts() else { return Ok(None) };
1461 if column.validity().has_nulls(chunk.len()) {
1462 return Ok(None);
1463 }
1464 match held {
1465 Some((_, first)) if !Arc::ptr_eq(first, dictionary) => return Ok(None),
1466 Some(_) => {}
1467 None => held = Some((column, dictionary)),
1468 }
1469 }
1470 let Some((column, dictionary)) = held else { return Ok(None) };
1471 let mut codes = Vec::with_capacity(blend.literals.len());
1472 for (text, lookup) in &blend.literals {
1473 match lookup.find(column, text.as_bytes()) {
1474 Some(Ok(Found::At(code))) => codes.push(code),
1475 Some(Err(error)) => return Err(error),
1476 Some(Ok(Found::Absent)) | None => return Ok(None),
1477 }
1478 }
1479 Ok(Some((Arc::clone(dictionary), codes)))
1480}
1481
1482/// The blend a `CASE` can be answered by, or `None` for one that has to read its branches' values.
1483fn blending(ty: &LogicalType, arms: &[PreparedArm], otherwise: Option<&Prepared>) -> Option<Blend> {
1484 if !matches!(ty, LogicalType::Varchar) {
1485 return None;
1486 }
1487 let otherwise = otherwise?;
1488 let mut branches = Vec::with_capacity(arms.len() + 1);
1489 let mut literals = Vec::new();
1490 for branch in arms.iter().map(|arm| &arm.then).chain([otherwise]) {
1491 branches.push(named(branch, &mut literals)?);
1492 }
1493 // All of them literals means there is no dictionary to name any of them in, and a `CASE` whose
1494 // every branch is a constant is not a thing anybody writes.
1495 let any = branches.iter().any(|branch| matches!(branch, Branch::Column(_)));
1496 any.then_some(Blend { branches, literals })
1497}
1498
1499/// The branch a prepared expression stands for, when it names a value rather than computing one.
1500fn named(prepared: &Prepared, literals: &mut Vec<(String, Lookup)>) -> Option<Branch> {
1501 match prepared.steps.as_slice() {
1502 [Step::Column(position)] => Some(Branch::Column(*position)),
1503 [Step::Constant(Value::Varchar(text))] => {
1504 literals.push((text.clone(), Lookup::default()));
1505 Some(Branch::Literal(literals.len() - 1))
1506 }
1507 _ => None,
1508 }
1509}
1510
1511/// The chunk cut down to the given rows.
1512///
1513/// The reason `CASE` is written with this rather than by evaluating every arm over the whole chunk
1514/// and picking afterwards. `CASE WHEN x <> 0 THEN 1 // x ELSE 0 END` divides by zero on the rows the
1515/// arm does not apply to if the arm is evaluated for them, and a `CASE` that raises on a row it was
1516/// written to exclude is the classic wrong answer this shape prevents.
1517pub(crate) fn narrow(chunk: &Chunk, rows: &[usize]) -> Result<Chunk> {
1518 let mut selection = Selection::with_capacity(rows.len());
1519 for &row in rows {
1520 selection.push(row);
1521 }
1522 chunk.clone().select(&selection)
1523}
1524
1525/// The kernels' comparison for the plan's.
1526///
1527/// A translation rather than one shared enum, because the kernels are rank 3 and the plan is rank
1528/// 9. This function is the whole of what that separation costs.
1529pub(crate) fn comparison(op: CompareOp) -> Comparison {
1530 match op {
1531 CompareOp::Equal => Comparison::Equal,
1532 CompareOp::NotEqual => Comparison::NotEqual,
1533 CompareOp::Less => Comparison::Less,
1534 CompareOp::LessOrEqual => Comparison::LessOrEqual,
1535 CompareOp::Greater => Comparison::Greater,
1536 CompareOp::GreaterOrEqual => Comparison::GreaterOrEqual,
1537 CompareOp::DistinctFrom => Comparison::DistinctFrom,
1538 CompareOp::NotDistinctFrom => Comparison::NotDistinctFrom,
1539 }
1540}
1541
1542/// The kernels' connective for the plan's.
1543pub(crate) fn connective(op: ConjunctionOp) -> Connective {
1544 match op {
1545 ConjunctionOp::And => Connective::And,
1546 ConjunctionOp::Or => Connective::Or,
1547 }
1548}
1549
1550#[cfg(test)]
1551mod tests {
1552 use rudb_common::{Field, LogicalType, Value};
1553 use rudb_kernels::is_true;
1554 use rudb_plan::{ExprRef, Node, Plan};
1555 use rudb_vector::{Chunk, Selection, Vector};
1556
1557 use super::{Prepared, narrow};
1558 use crate::expr::evaluate;
1559 use crate::schema::Schema;
1560
1561 /// Two columns with a null in each, because every disagreement between these two evaluators
1562 /// that is worth finding is a disagreement about which rows are null.
1563 fn input() -> (Schema, Chunk) {
1564 let schema = Schema::numbered(
1565 vec![Field::new("x", LogicalType::Integer), Field::new("s", LogicalType::Varchar)],
1566 0,
1567 );
1568 let x = Vector::from_values(
1569 LogicalType::Integer,
1570 &[Value::Integer(3), Value::Integer(1), Value::Null, Value::Integer(2)],
1571 )
1572 .expect("four integers");
1573 let s = Vector::from_values(
1574 LogicalType::Varchar,
1575 &[
1576 Value::Varchar("a".to_string()),
1577 Value::Null,
1578 Value::Varchar("c".to_string()),
1579 Value::Varchar("a".to_string()),
1580 ],
1581 )
1582 .expect("four strings");
1583 (schema, Chunk::new(vec![x, s]).expect("two columns of four rows"))
1584 }
1585
1586 /// The expressions of a projection written in the plan's textual form, over the two columns
1587 /// [`input`] produces.
1588 ///
1589 /// Going through the text rather than the arena builders for the reason the other test module
1590 /// gives: a test that says what it evaluates in the notation a plan dump uses is a test whose
1591 /// failure can be pasted into a plan and vice versa.
1592 fn projection(exprs: &str) -> (Plan, Vec<ExprRef>) {
1593 let text =
1594 format!("Project #1 [{exprs}]\n Get memory.main.t AS t #0 [x::INTEGER, s::VARCHAR]");
1595 let plan = Plan::parse(&text).expect("a well formed plan");
1596 let Node::Project { exprs, .. } = *plan.node(plan.root()) else {
1597 panic!("the root of that text is a projection");
1598 };
1599 let list = plan.expr_list(exprs).to_vec();
1600 (plan, list)
1601 }
1602
1603 /// Every expression shape, evaluated both ways over the same chunk.
1604 ///
1605 /// This is the agreement the module documentation claims and it is the only thing that makes
1606 /// the prepared form safe to put in front of the tree walk. The generated well typed trees the
1607 /// test gate of #57 asks for are a wider version of this and are worth building once the
1608 /// selection threaded shapes exist to disagree about.
1609 fn agrees(exprs: &str) {
1610 let (schema, chunk) = input();
1611 let (plan, list) = projection(exprs);
1612 let prepared = Prepared::new(&plan, &list, &schema).expect("the expressions resolve");
1613 let mut scratch = prepared.scratch();
1614 let mut fast = Vec::new();
1615 prepared.evaluate(&chunk, &mut scratch, &mut fast).expect("the prepared form runs");
1616 for (at, &expr) in list.iter().enumerate() {
1617 let slow = evaluate(&plan, expr, &schema, &chunk).expect("the tree walk runs");
1618 for row in 0..chunk.len() {
1619 assert_eq!(
1620 fast[at].value_at(row),
1621 slow.value_at(row),
1622 "expression {at} of `{exprs}` at row {row}"
1623 );
1624 }
1625 }
1626 }
1627
1628 /// Three decimal columns of TPC-H's shape, in the form `form` puts them in.
1629 fn decimals(prices: &[i128], form: fn(Vector) -> Vector) -> (Schema, Chunk) {
1630 let ty = LogicalType::Decimal { width: 15, scale: 2 };
1631 let schema = Schema::numbered(
1632 vec![
1633 Field::new("p", ty.clone()),
1634 Field::new("d", ty.clone()),
1635 Field::new("t", ty.clone()),
1636 ],
1637 0,
1638 );
1639 let column =
1640 |values: Vec<Value>| form(Vector::from_values(ty.clone(), &values).expect("decimals"));
1641 let decimal = |unscaled| Value::Decimal { unscaled, width: 15, scale: 2 };
1642 let p = column(prices.iter().map(|&v| decimal(v)).collect());
1643 let d = column((0..prices.len() as i128).map(|v| decimal(v % 11)).collect());
1644 let t = column((0..prices.len() as i128).map(|v| decimal(v % 9)).collect());
1645 (schema, Chunk::new(vec![p, d, t]).expect("three columns"))
1646 }
1647
1648 /// q01's charge, as the binder writes it.
1649 const CHARGE: &str = "\"*\"(\"*\"(CAST(#0.0::DECIMAL(15,2))::DECIMAL(18,2), \
1650 CAST(\"-\"(1.00::DECIMAL(16,2), CAST(#0.1::DECIMAL(15,2))::DECIMAL(16,2))::DECIMAL(16,2))\
1651 ::DECIMAL(18,2))::DECIMAL(18,4), CAST(\"+\"(1.00::DECIMAL(16,2), \
1652 CAST(#0.2::DECIMAL(15,2))::DECIMAL(16,2))::DECIMAL(16,2))::DECIMAL(18,2))::DECIMAL(18,6) AS a";
1653
1654 /// The fused answer, the unfused one and the tree walk's, over one chunk.
1655 fn three_ways(chunk: &Chunk, schema: &Schema) -> [rudb_common::Result<Vec<Value>>; 3] {
1656 let text = format!(
1657 "Project #1 [{CHARGE}]\n Get memory.main.t AS t #0 \
1658 [p::DECIMAL(15,2), d::DECIMAL(15,2), t::DECIMAL(15,2)]"
1659 );
1660 let plan = Plan::parse(&text).expect("a well formed plan");
1661 let Node::Project { exprs, .. } = *plan.node(plan.root()) else {
1662 panic!("the root of that text is a projection");
1663 };
1664 let expr = plan.expr_list(exprs)[0];
1665 let values = |vector: &Vector| (0..chunk.len()).map(|row| vector.value_at(row)).collect();
1666 let fused = Prepared::one(&plan, expr, schema).expect("resolves");
1667 assert_eq!(fused.fused(), 1, "the whole tree is one step");
1668 let unfused = Prepared::built(&plan, &[expr], schema, false, false).expect("resolves");
1669 assert_eq!(unfused.fused(), 0);
1670 let run = |prepared: &Prepared| {
1671 prepared.evaluate_one(chunk, &mut prepared.scratch()).map(&values)
1672 };
1673 [run(&fused), run(&unfused), evaluate(&plan, expr, schema, chunk).map(|v| values(&v))]
1674 }
1675
1676 fn all_agree(chunk: &Chunk, schema: &Schema) {
1677 let [fused, unfused, walked] = three_ways(chunk, schema);
1678 let fused = fused.expect("fits");
1679 assert_eq!(fused, unfused.expect("fits"));
1680 assert_eq!(fused, walked.expect("fits"));
1681 }
1682
1683 /// The epoch plus a whole count of seconds runs as one call, and agrees with the cast, the
1684 /// interval and the shift it stands for, on both sides of the count where the double stops
1685 /// being exact and on a count that takes the answer out of range.
1686 #[test]
1687 fn a_timestamp_plus_whole_seconds_agrees_with_the_interval_it_stands_for() {
1688 let schema = Schema::numbered(vec![Field::new("x", LogicalType::BigInt)], 0);
1689 let counts = [
1690 Value::BigInt(1_373_000_000),
1691 Value::BigInt(-5),
1692 Value::Null,
1693 Value::BigInt(9_007_199_254),
1694 Value::BigInt(9_007_199_255),
1695 Value::BigInt(9_000_000_000_123),
1696 ];
1697 let x = Vector::from_values(LogicalType::BigInt, &counts).expect("six counts");
1698 let chunk = Chunk::new(vec![x]).expect("one column");
1699 let text = "Project #1 [\"+\"(0::TIMESTAMP, to_seconds(CAST(#0.0::BIGINT)::DOUBLE)::INTERVAL)::TIMESTAMP AS e]\n Get memory.main.t AS t #0 [x::BIGINT]";
1700 let plan = Plan::parse(text).expect("a well formed plan");
1701 let Node::Project { exprs, .. } = *plan.node(plan.root()) else {
1702 panic!("the root of that text is a projection");
1703 };
1704 let list = plan.expr_list(exprs).to_vec();
1705 let prepared = Prepared::new(&plan, &list, &schema).expect("the expression resolves");
1706 assert!(
1707 prepared.steps.iter().any(
1708 |step| matches!(step, super::Step::Function { recipe, .. } if recipe.name() == "__rudb_stamp_seconds")
1709 ),
1710 "the shift is one call"
1711 );
1712 let mut scratch = prepared.scratch();
1713 let mut fast = Vec::new();
1714 prepared.evaluate(&chunk, &mut scratch, &mut fast).expect("the prepared form runs");
1715 let slow = evaluate(&plan, list[0], &schema, &chunk).expect("the tree walk runs");
1716 for row in 0..chunk.len() {
1717 assert_eq!(fast[0].value_at(row), slow.value_at(row), "row {row}");
1718 }
1719 assert_eq!(fast[0].value_at(0), Value::Timestamp(1_373_000_000_000_000));
1720
1721 let far = Vector::from_values(LogicalType::BigInt, &[Value::BigInt(9_300_000_000_000)])
1722 .expect("one count");
1723 let chunk = Chunk::new(vec![far]).expect("one column");
1724 let mut fast = Vec::new();
1725 let fused = prepared.evaluate(&chunk, &mut scratch, &mut fast);
1726 let slow = evaluate(&plan, list[0], &schema, &chunk).map(|_| ());
1727 assert!(fused.is_err() && slow.is_err(), "past the last timestamp both raise");
1728 }
1729
1730 #[test]
1731 fn decimal_arithmetic_run_as_one_loop_agrees_in_every_form() {
1732 let prices: Vec<i128> = (0..2500).map(|v| 90_000 + v * 37).collect();
1733 let packed = |vector: Vector| vector.bit_packed().expect("packs");
1734 let coded = |vector: Vector| {
1735 let rows = vector.len();
1736 let codes = (0..rows as u32).rev().collect();
1737 Vector::dictionary(codes, vector.bit_packed().expect("packs")).expect("in range")
1738 };
1739 // Codes too far apart for a block to unpack the run they cover.
1740 let scattered = |vector: Vector| {
1741 let rows = vector.len() as u32;
1742 let codes = (0..rows).map(|row| row * 997 % rows).collect();
1743 Vector::dictionary(codes, vector.bit_packed().expect("packs")).expect("in range")
1744 };
1745 for form in [std::convert::identity, packed, coded, scattered] {
1746 let (schema, chunk) = decimals(&prices, form);
1747 all_agree(&chunk, &schema);
1748 }
1749 }
1750
1751 #[test]
1752 fn a_chunk_the_ranges_cannot_prove_raises_what_the_steps_raise() {
1753 // The large price in the second block, so a flat column gets as far as running the first.
1754 let mut prices = vec![5; 300];
1755 prices.push(999_999_999_999_999);
1756 let packed = |vector: Vector| vector.bit_packed().expect("packs");
1757 for form in [std::convert::identity, packed] {
1758 let (schema, chunk) = decimals(&prices, form);
1759 let [fused, unfused, _] = three_ways(&chunk, &schema);
1760 let (fused, unfused) = (fused.expect_err("overflows"), unfused.expect_err("overflows"));
1761 assert_eq!(fused.message(), unfused.message());
1762 }
1763 }
1764
1765 #[test]
1766 fn a_chunk_with_a_null_goes_through_the_steps() {
1767 let ty = LogicalType::Decimal { width: 15, scale: 2 };
1768 let (schema, mut chunk) = decimals(&[100, 200, 300], std::convert::identity);
1769 let with_null = Vector::from_values(
1770 ty,
1771 &[Value::Decimal { unscaled: 5, width: 15, scale: 2 }, Value::Null, Value::Null],
1772 )
1773 .expect("decimals");
1774 chunk = Chunk::new(vec![
1775 chunk.column(0).expect("p").clone(),
1776 with_null,
1777 chunk.column(2).expect("t").clone(),
1778 ])
1779 .expect("three columns");
1780 all_agree(&chunk, &schema);
1781 }
1782
1783 #[test]
1784 fn a_column_reference_agrees() {
1785 agrees("#0.0::INTEGER AS a, #0.1::VARCHAR AS b");
1786 }
1787
1788 #[test]
1789 fn a_constant_agrees() {
1790 agrees("7::INTEGER AS a, NULL::INTEGER AS b");
1791 }
1792
1793 #[test]
1794 fn a_cast_agrees() {
1795 agrees("CAST(#0.0::INTEGER)::BIGINT AS a, CAST(#0.0::INTEGER)::VARCHAR AS b");
1796 }
1797
1798 #[test]
1799 fn a_comparison_agrees() {
1800 agrees("(#0.0::INTEGER > 1::INTEGER)::BOOLEAN AS a");
1801 }
1802
1803 #[test]
1804 fn a_conjunction_agrees() {
1805 agrees(
1806 "((#0.0::INTEGER > 1::INTEGER)::BOOLEAN AND (#0.0::INTEGER < 3::INTEGER)::BOOLEAN)\
1807 ::BOOLEAN AS a",
1808 );
1809 }
1810
1811 #[test]
1812 fn a_function_agrees() {
1813 agrees("\"+\"(#0.0::INTEGER, 1::INTEGER)::INTEGER AS a");
1814 }
1815
1816 /// The two evaluators quote the same expression when a divisor is zero. Per #262.
1817 ///
1818 /// This is the one message in the engine that depends on how an expression is written rather
1819 /// than on what it computes, and the two evaluators render it at different times: the prepared
1820 /// form when the pipeline is built, the tree walk on the row that fails. Same renderer, so the
1821 /// same sentence, and this is what says so.
1822 #[test]
1823 fn both_evaluators_quote_the_same_expression_when_a_divisor_is_zero() {
1824 let (schema, chunk) = input();
1825 let (plan, list) = projection("\"//\"(#0.0::INTEGER, 0::INTEGER)::INTEGER AS a");
1826 let prepared = Prepared::new(&plan, &list, &schema).expect("the expression resolves");
1827 let mut scratch = prepared.scratch();
1828 let mut out = Vec::new();
1829 let fast = prepared.evaluate(&chunk, &mut scratch, &mut out).expect_err("divides by zero");
1830 let slow = evaluate(&plan, list[0], &schema, &chunk).expect_err("divides by zero");
1831 assert_eq!(fast.message(), slow.message());
1832 assert!(fast.message().starts_with("Division by zero in expression (x // 0)."), "{fast}");
1833 }
1834
1835 #[test]
1836 fn a_case_agrees() {
1837 agrees(
1838 "CASE WHEN (#0.0::INTEGER > 1::INTEGER)::BOOLEAN THEN 10::INTEGER \
1839 ELSE 20::INTEGER END::INTEGER AS a",
1840 );
1841 }
1842
1843 /// A second arm, which is the first one that sees a cut chunk rather than the whole one.
1844 ///
1845 /// The first arm of any `CASE` runs over every row, so it takes the path that does not cut at
1846 /// all, and a `CASE` of one arm never exercises the other one. Two arms and an `ELSE` puts a
1847 /// different set of rows in front of each of the three.
1848 ///
1849 /// That this is the only test here reaching the cut was checked rather than assumed, by gating a
1850 /// panic on it and rerunning the seven. This one failed and the other six did not.
1851 #[test]
1852 fn a_case_of_two_arms_agrees() {
1853 agrees(
1854 "CASE WHEN (#0.0::INTEGER > 2::INTEGER)::BOOLEAN THEN 10::INTEGER \
1855 WHEN (#0.0::INTEGER > 1::INTEGER)::BOOLEAN THEN 20::INTEGER \
1856 ELSE 30::INTEGER END::INTEGER AS a",
1857 );
1858 }
1859
1860 /// No `ELSE`, so the rows no arm claims are null rather than anything.
1861 ///
1862 /// The case a run of data with a hole in it gets wrong: a null still occupies a position, and an
1863 /// assembly that skipped it would put every value after it one row early.
1864 #[test]
1865 fn a_case_with_no_else_agrees() {
1866 agrees(
1867 "CASE WHEN (#0.0::INTEGER > 2::INTEGER)::BOOLEAN THEN 10::INTEGER \
1868 END::INTEGER AS a",
1869 );
1870 }
1871
1872 /// An arm no row takes, so it contributes nothing to the answer and must not shift it.
1873 #[test]
1874 fn a_case_whose_arm_claims_nothing_agrees() {
1875 agrees(
1876 "CASE WHEN (#0.0::INTEGER > 99::INTEGER)::BOOLEAN THEN 10::INTEGER \
1877 ELSE 20::INTEGER END::INTEGER AS a",
1878 );
1879 }
1880
1881 /// Strings, which is the case that used to allocate one of them per row and drop it afterwards.
1882 ///
1883 /// The arm reads a column and the `ELSE` is a constant, which is the shape of the ClickBench
1884 /// query this path was rewritten for: the arm arrives as views over an arena and the `ELSE` as
1885 /// one value repeated, and the two have to be laid end to end into a single arena.
1886 #[test]
1887 fn a_case_over_strings_agrees() {
1888 agrees(
1889 "CASE WHEN (#0.0::INTEGER > 1::INTEGER)::BOOLEAN THEN #0.1::VARCHAR \
1890 ELSE ''::VARCHAR END::VARCHAR AS a",
1891 );
1892 }
1893
1894 /// A null inside an arm, which is a different thing from a row no arm claimed.
1895 ///
1896 /// Both come out null and they reach the validity mask by different routes, so a mask built for
1897 /// one of them and not the other reads correct on whichever test only has the other in it.
1898 #[test]
1899 fn a_case_whose_arm_answers_null_agrees() {
1900 agrees(
1901 "CASE WHEN (#0.0::INTEGER > 1::INTEGER)::BOOLEAN THEN #0.1::VARCHAR \
1902 ELSE NULL::VARCHAR END::VARCHAR AS a",
1903 );
1904 }
1905
1906 /// A `WHEN` over a column that is null on some rows, which is neither true nor false there.
1907 ///
1908 /// A three valued `WHEN` is what decides whether a row goes to the arm or falls through, and
1909 /// treating unknown as true would claim a row the `ELSE` should have had.
1910 #[test]
1911 fn a_case_whose_test_is_null_on_some_rows_agrees() {
1912 agrees(
1913 "CASE WHEN (#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN THEN 10::INTEGER \
1914 ELSE 20::INTEGER END::INTEGER AS a",
1915 );
1916 }
1917
1918 /// The same expression twice, which is where the tree walk copies the column twice and this
1919 /// does not, and the answers still have to be identical.
1920 #[test]
1921 fn a_column_mentioned_three_times_agrees() {
1922 agrees("\"+\"(\"+\"(#0.0::INTEGER, #0.0::INTEGER)::INTEGER, #0.0::INTEGER)::INTEGER AS a");
1923 }
1924
1925 /// The intermediates of a chain are not all held to the end of it.
1926 ///
1927 /// This is the whole difference between the prepared form being faster than the tree walk on a
1928 /// deep chain and being slower than it, and it is a property of the slot array rather than of
1929 /// any answer, so it is asserted here rather than left to the benchmark to catch.
1930 #[test]
1931 fn a_chain_holds_one_intermediate_at_a_time() {
1932 let (schema, chunk) = input();
1933 let mut expr = "#0.0::INTEGER".to_string();
1934 for _ in 0..8 {
1935 expr = format!("\"+\"({expr}, 1::INTEGER)::INTEGER");
1936 }
1937 let (plan, list) = projection(&format!("{expr} AS a"));
1938 let prepared = Prepared::new(&plan, &list, &schema).expect("the chain resolves");
1939 let mut scratch = prepared.scratch();
1940 prepared.run(&chunk, &mut scratch).expect("the chain runs");
1941 let live = scratch.slots.iter().filter(|slot| slot.is_some()).count();
1942 assert_eq!(live, 1, "a chain that has run should be holding its answer and nothing else");
1943 }
1944
1945 /// The rows a threaded filter keeps are the rows the tree walk says the predicate is true for.
1946 ///
1947 /// Every threaded conjunct is a chance to disagree with the unthreaded answer about a null,
1948 /// about a row an earlier conjunct had already dropped, or about a chunk nothing survives, and
1949 /// the answer is a set of row numbers rather than a vector, so this is checked against the tree
1950 /// walk read a row at a time rather than against the prepared form it is part of.
1951 fn filters(predicate: &str) {
1952 let (schema, chunk) = input();
1953 let (plan, list) = projection(&format!("{predicate} AS p"));
1954 let prepared = Prepared::new(&plan, &list, &schema).expect("the predicate resolves");
1955 let mut scratch = prepared.scratch();
1956 let threaded = prepared.evaluate_filter(&chunk, &mut scratch).expect("the filter runs");
1957 let flags = evaluate(&plan, list[0], &schema, &chunk).expect("the tree walk runs");
1958 let expected = Selection::from_predicate(chunk.len(), |row| is_true(&flags.value_at(row)));
1959 assert_eq!(threaded, expected, "`{predicate}`");
1960 // And running it again over the same scratch is the same answer, because a pipeline calls
1961 // this once a chunk and a slot left behind by the conjunct before would show up here.
1962 let again = prepared.evaluate_filter(&chunk, &mut scratch).expect("the filter runs again");
1963 assert_eq!(again, expected, "`{predicate}` a second time");
1964 }
1965
1966 /// A predicate with no `AND` in it is not threaded and has to keep saying the same thing.
1967 #[test]
1968 fn a_single_comparison_filters_the_same_rows() {
1969 filters("(#0.0::INTEGER > 1::INTEGER)::BOOLEAN");
1970 filters("(#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN");
1971 filters("(#0.0::INTEGER IS NOT DISTINCT FROM NULL::INTEGER)::BOOLEAN");
1972 }
1973
1974 #[test]
1975 fn a_chain_of_conjuncts_keeps_what_all_of_them_keep() {
1976 filters(
1977 "((#0.0::INTEGER > 1::INTEGER)::BOOLEAN AND (#0.0::INTEGER < 3::INTEGER)::BOOLEAN)\
1978 ::BOOLEAN",
1979 );
1980 filters(
1981 "((#0.0::INTEGER >= 1::INTEGER)::BOOLEAN AND (#0.0::INTEGER <= 3::INTEGER)::BOOLEAN \
1982 AND (#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN AND (#0.0::INTEGER <> 2::INTEGER)\
1983 ::BOOLEAN)::BOOLEAN",
1984 );
1985 }
1986
1987 /// A conjunct that rejects every row, in front of one that would have kept some. The rows are
1988 /// the same either way and the point of the shape is that the second conjunct never runs.
1989 #[test]
1990 fn a_conjunct_that_keeps_nothing_ends_the_predicate() {
1991 filters(
1992 "((#0.0::INTEGER > 9::INTEGER)::BOOLEAN AND (#0.0::INTEGER < 9::INTEGER)::BOOLEAN)\
1993 ::BOOLEAN",
1994 );
1995 }
1996
1997 /// A conjunct whose operands are computed rather than read, which is the shape where the
1998 /// comparison is threaded and the arithmetic under it is not.
1999 #[test]
2000 fn a_conjunct_over_a_computed_operand_keeps_the_same_rows() {
2001 filters(
2002 "((#0.0::INTEGER > 1::INTEGER)::BOOLEAN AND \
2003 (\"+\"(#0.0::INTEGER, 1::INTEGER)::INTEGER < 4::INTEGER)::BOOLEAN)::BOOLEAN",
2004 );
2005 }
2006
2007 /// A conjunct that is not a comparison at all, which is the one that goes through the flag
2008 /// kernel rather than the comparison kernel.
2009 #[test]
2010 fn a_conjunct_that_is_not_a_comparison_is_threaded_too() {
2011 filters(
2012 "((#0.0::INTEGER > 1::INTEGER)::BOOLEAN AND ((#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN \
2013 OR (#0.0::INTEGER = 1::INTEGER)::BOOLEAN)::BOOLEAN)::BOOLEAN",
2014 );
2015 filters(
2016 "(((#0.1::VARCHAR = 'c'::VARCHAR)::BOOLEAN OR (#0.0::INTEGER = 3::INTEGER)::BOOLEAN)\
2017 ::BOOLEAN AND (#0.0::INTEGER <> 1::INTEGER)::BOOLEAN)::BOOLEAN",
2018 );
2019 }
2020
2021 #[test]
2022 fn a_selective_conjunct_evaluates_later_like_on_its_survivors() {
2023 filters(
2024 "((#0.0::INTEGER > 2::INTEGER)::BOOLEAN AND \
2025 \"~~\"(#0.1::VARCHAR, '%a%'::VARCHAR)::BOOLEAN)::BOOLEAN",
2026 );
2027 filters(
2028 "((#0.0::INTEGER > 2::INTEGER)::BOOLEAN AND \
2029 \"!~~\"(#0.1::VARCHAR, '%a%'::VARCHAR)::BOOLEAN)::BOOLEAN",
2030 );
2031 }
2032
2033 /// An `OR` at the top threads the complement: the second branch only sees the rows the first
2034 /// one did not accept, and the rows it accepts are added to them rather than replacing them.
2035 ///
2036 /// The input has a row where the first branch is true, one where the second is, one where both
2037 /// are false and one where the first is null and the second is true, which is the row that says
2038 /// whether the complement was taken over "not true" or over "false".
2039 #[test]
2040 fn an_or_at_the_top_threads_the_complement() {
2041 filters(
2042 "((#0.0::INTEGER > 2::INTEGER)::BOOLEAN OR (#0.1::VARCHAR = 'c'::VARCHAR)::BOOLEAN)\
2043 ::BOOLEAN",
2044 );
2045 filters(
2046 "((#0.0::INTEGER = 1::INTEGER)::BOOLEAN OR (#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN \
2047 OR (#0.0::INTEGER > 2::INTEGER)::BOOLEAN)::BOOLEAN",
2048 );
2049 }
2050
2051 /// A branch that accepts every row, in front of one that would have accepted none. The rows are
2052 /// the same either way and the point of the shape is that the second branch never runs.
2053 #[test]
2054 fn a_branch_that_keeps_everything_ends_the_predicate() {
2055 filters(
2056 "((#0.0::INTEGER IS NOT DISTINCT FROM #0.0::INTEGER)::BOOLEAN OR \
2057 (#0.0::INTEGER > 9::INTEGER)::BOOLEAN)::BOOLEAN",
2058 );
2059 }
2060
2061 /// The branches after one that has accepted every row really are skipped.
2062 ///
2063 /// Every other test here says the threaded answer matches the unthreaded one, which it would
2064 /// even if nothing were threaded at all. This one puts a division by zero behind a branch that
2065 /// accepts everything, so the predicate raises if the second branch runs and does not if the
2066 /// walk stopped where it was supposed to.
2067 #[test]
2068 fn a_branch_behind_one_that_accepted_every_row_does_not_run() {
2069 let (schema, chunk) = input();
2070 let predicate = "((#0.0::INTEGER IS NOT DISTINCT FROM #0.0::INTEGER)::BOOLEAN OR \
2071 (\"//\"(#0.0::INTEGER, 0::INTEGER)::INTEGER > 0::INTEGER)::BOOLEAN)\
2072 ::BOOLEAN";
2073 let (plan, list) = projection(&format!("{predicate} AS p"));
2074 let prepared = Prepared::new(&plan, &list, &schema).expect("the predicate resolves");
2075 let mut scratch = prepared.scratch();
2076 let kept =
2077 prepared.evaluate_filter(&chunk, &mut scratch).expect("the second branch never runs");
2078 assert_eq!(kept, Selection::identity(chunk.len()));
2079 // And the same predicate evaluated as an expression does divide by zero, which is what says
2080 // the test is testing the threading rather than a predicate that happens not to raise.
2081 evaluate(&plan, list[0], &schema, &chunk).expect_err("the tree walk divides by zero");
2082 }
2083
2084 /// The conjunct that rejects the most rows ends up in front of the one that rejects none.
2085 ///
2086 /// The predicate is written the wrong way round on purpose. The plan order costs two passes a
2087 /// chunk where one would do, and after a chunk of watching it the filter runs the selective one
2088 /// first and the other one stops running at all.
2089 #[test]
2090 fn a_filter_learns_which_conjunct_to_run_first() {
2091 let (schema, chunk) = input();
2092 let predicate = "((#0.0::INTEGER > 0::INTEGER)::BOOLEAN AND (#0.0::INTEGER > 9::INTEGER)\
2093 ::BOOLEAN)::BOOLEAN";
2094 let (plan, list) = projection(&format!("{predicate} AS p"));
2095 let prepared = Prepared::new(&plan, &list, &schema).expect("the predicate resolves");
2096 let mut scratch = prepared.scratch();
2097 let root = prepared.roots[0];
2098 assert_eq!(scratch.order(root), None, "nothing has run yet");
2099 let kept = prepared.evaluate_filter(&chunk, &mut scratch).expect("the filter runs");
2100 assert!(kept.is_empty());
2101 assert_eq!(scratch.order(root), Some(&[1, 0][..]), "the second conjunct rejects the most");
2102 // And it stays there, because the conjunct that now runs first empties the selection and
2103 // the one behind it keeps the history it already had rather than losing it.
2104 let kept = prepared.evaluate_filter(&chunk, &mut scratch).expect("the filter runs again");
2105 assert!(kept.is_empty());
2106 assert_eq!(scratch.order(root), Some(&[1, 0][..]));
2107 }
2108
2109 /// Whatever order it settles on, the rows are the rows.
2110 ///
2111 /// Run for longer than the window is wide, because an order that changes halfway through a scan
2112 /// is the shape where a walk that got the subtree bookkeeping wrong would start reading the
2113 /// wrong steps, and the first chunk would not show it.
2114 #[test]
2115 fn reordering_never_changes_which_rows_survive() {
2116 let (schema, chunk) = input();
2117 let predicate = "((#0.0::INTEGER >= 1::INTEGER)::BOOLEAN AND \
2118 (\"+\"(#0.0::INTEGER, 1::INTEGER)::INTEGER < 4::INTEGER)::BOOLEAN AND \
2119 (#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN)::BOOLEAN";
2120 let (plan, list) = projection(&format!("{predicate} AS p"));
2121 let prepared = Prepared::new(&plan, &list, &schema).expect("the predicate resolves");
2122 let mut scratch = prepared.scratch();
2123 let flags = evaluate(&plan, list[0], &schema, &chunk).expect("the tree walk runs");
2124 let expected = Selection::from_predicate(chunk.len(), |row| is_true(&flags.value_at(row)));
2125 for round in 0..40 {
2126 let kept = prepared.evaluate_filter(&chunk, &mut scratch).expect("the filter runs");
2127 assert_eq!(kept, expected, "round {round}");
2128 }
2129 }
2130
2131 /// A nested connective is threaded rather than evaluated into flags.
2132 ///
2133 /// The inner `AND` keeps nothing, so its second conjunct is never reached and the division by
2134 /// zero in it never happens. Evaluating the branch as an expression and narrowing the flags
2135 /// afterwards, which is what an operand that is not a connective still does, would have run it.
2136 #[test]
2137 fn a_nested_connective_stops_where_the_outer_one_would() {
2138 let (schema, chunk) = input();
2139 let predicate = "((#0.0::INTEGER > 9::INTEGER)::BOOLEAN OR ((#0.0::INTEGER > 9::INTEGER)\
2140 ::BOOLEAN AND (\"//\"(#0.0::INTEGER, 0::INTEGER)::INTEGER > 0::INTEGER)\
2141 ::BOOLEAN)::BOOLEAN)::BOOLEAN";
2142 let (plan, list) = projection(&format!("{predicate} AS p"));
2143 let prepared = Prepared::new(&plan, &list, &schema).expect("the predicate resolves");
2144 let mut scratch = prepared.scratch();
2145 let kept =
2146 prepared.evaluate_filter(&chunk, &mut scratch).expect("the division never happens");
2147 assert!(kept.is_empty());
2148 evaluate(&plan, list[0], &schema, &chunk).expect_err("the tree walk divides by zero");
2149 }
2150
2151 /// A branch that is not a comparison, which is the one that goes through the flag kernel.
2152 #[test]
2153 fn an_or_branch_that_is_not_a_comparison_is_threaded_too() {
2154 filters(
2155 "((#0.0::INTEGER > 2::INTEGER)::BOOLEAN OR \
2156 \"~~\"(#0.1::VARCHAR, 'a%'::VARCHAR)::BOOLEAN)::BOOLEAN",
2157 );
2158 filters(
2159 "(\"~~\"(#0.1::VARCHAR, 'c%'::VARCHAR)::BOOLEAN OR (#0.0::INTEGER = 1::INTEGER)\
2160 ::BOOLEAN)::BOOLEAN",
2161 );
2162 }
2163
2164 /// A connective inside a connective, which recurses rather than falling back to flags.
2165 ///
2166 /// Both nestings, because the two carry opposite things: an `AND` under an `OR` starts from the
2167 /// rows no branch has accepted, and an `OR` under an `AND` starts from the rows every conjunct
2168 /// has kept, and getting either one backwards is a wrong set of rows.
2169 #[test]
2170 fn a_connective_inside_a_connective_threads_both_ways() {
2171 filters(
2172 "(((#0.0::INTEGER >= 2::INTEGER)::BOOLEAN AND (#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN)\
2173 ::BOOLEAN OR ((#0.0::INTEGER < 2::INTEGER)::BOOLEAN AND (#0.1::VARCHAR <> 'c'\
2174 ::VARCHAR)::BOOLEAN)::BOOLEAN)::BOOLEAN",
2175 );
2176 filters(
2177 "(((#0.1::VARCHAR = 'c'::VARCHAR)::BOOLEAN OR (#0.0::INTEGER = 3::INTEGER)::BOOLEAN)\
2178 ::BOOLEAN AND ((#0.0::INTEGER <> 1::INTEGER)::BOOLEAN OR (#0.1::VARCHAR = 'a'\
2179 ::VARCHAR)::BOOLEAN)::BOOLEAN)::BOOLEAN",
2180 );
2181 // Three deep, since two levels is where an off by one in the subtree bookkeeping can still
2182 // be hidden by the ranges lining up.
2183 filters(
2184 "((#0.0::INTEGER > 9::INTEGER)::BOOLEAN OR ((#0.0::INTEGER >= 1::INTEGER)::BOOLEAN \
2185 AND ((#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN OR (#0.0::INTEGER = 1::INTEGER)\
2186 ::BOOLEAN)::BOOLEAN)::BOOLEAN)::BOOLEAN",
2187 );
2188 }
2189
2190 /// A predicate where one side is null and the other is true, in both orders. `OR` is true there
2191 /// and a complement taken over the rows a branch rejected rather than the rows it accepted
2192 /// would drop the row, which is the one way this can be wrong and is not a wrong vector but a
2193 /// missing row.
2194 #[test]
2195 fn a_null_branch_beside_a_true_one_keeps_the_row() {
2196 filters(
2197 "((#0.0::INTEGER > 2::INTEGER)::BOOLEAN OR (#0.1::VARCHAR = 'c'::VARCHAR)::BOOLEAN \
2198 OR (#0.0::INTEGER IS NOT DISTINCT FROM NULL::INTEGER)::BOOLEAN)::BOOLEAN",
2199 );
2200 filters(
2201 "((#0.1::VARCHAR > 'b'::VARCHAR)::BOOLEAN OR (#0.0::INTEGER = 1::INTEGER)::BOOLEAN)\
2202 ::BOOLEAN",
2203 );
2204 }
2205
2206 /// A filter over a chunk that has already been narrowed, which is what a second filter in a
2207 /// pipeline sees and is the form pair the threaded kernels have to handle rather than fall
2208 /// through on.
2209 #[test]
2210 fn a_filter_over_a_selected_chunk_keeps_the_same_rows() {
2211 let (schema, chunk) = input();
2212 let predicate = "((#0.0::INTEGER >= 1::INTEGER)::BOOLEAN AND \
2213 (#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN)::BOOLEAN";
2214 let (plan, list) = projection(&format!("{predicate} AS p"));
2215 let prepared = Prepared::new(&plan, &list, &schema).expect("the predicate resolves");
2216 let mut scratch = prepared.scratch();
2217 let narrowed = narrow(&chunk, &[0, 3]).expect("two of the four rows");
2218 let threaded = prepared.evaluate_filter(&narrowed, &mut scratch).expect("the filter runs");
2219 let flags = evaluate(&plan, list[0], &schema, &narrowed).expect("the tree walk runs");
2220 let expected =
2221 Selection::from_predicate(narrowed.len(), |row| is_true(&flags.value_at(row)));
2222 assert_eq!(threaded, expected);
2223 }
2224
2225 /// Preparing is per pipeline and evaluating is per chunk, so the scratch has to survive being
2226 /// used again and give the same answer the second time.
2227 #[test]
2228 fn a_scratch_used_twice_gives_the_same_answer_twice() {
2229 let (schema, chunk) = input();
2230 let (plan, list) = projection("\"+\"(#0.0::INTEGER, 1::INTEGER)::INTEGER AS a");
2231 let prepared = Prepared::new(&plan, &list, &schema).expect("the expressions resolve");
2232 let mut scratch = prepared.scratch();
2233 let mut once = Vec::new();
2234 prepared.evaluate(&chunk, &mut scratch, &mut once).expect("the first chunk runs");
2235 let mut twice = Vec::new();
2236 prepared.evaluate(&chunk, &mut scratch, &mut twice).expect("the second chunk runs");
2237 assert_eq!(once, twice);
2238 }
2239
2240 #[test]
2241 fn a_shared_computed_root_is_compiled_once() {
2242 let (schema, chunk) = input();
2243 let (plan, list) = projection("\"+\"(#0.0::INTEGER, 1::INTEGER)::INTEGER AS a");
2244 let prepared = Prepared::shared(&plan, &[list[0], list[0]], &schema)
2245 .expect("the shared expression resolves");
2246 assert_eq!(prepared.steps.len(), 3);
2247 let mut scratch = prepared.scratch();
2248 let mut answers = Vec::new();
2249 prepared.evaluate(&chunk, &mut scratch, &mut answers).expect("both roots are returned");
2250 assert_eq!(answers[0], answers[1]);
2251 }
2252
2253 /// A chunk shorter than the last one, because a scan's final chunk is that and a constant
2254 /// materialized to the wrong length would be an out of range read rather than a wrong answer.
2255 #[test]
2256 fn a_shorter_chunk_after_a_longer_one_is_evaluated_at_its_own_length() {
2257 let (schema, chunk) = input();
2258 let (plan, list) = projection("7::INTEGER AS a");
2259 let prepared = Prepared::new(&plan, &list, &schema).expect("the expressions resolve");
2260 let mut scratch = prepared.scratch();
2261 let mut full = Vec::new();
2262 prepared.evaluate(&chunk, &mut scratch, &mut full).expect("the full chunk runs");
2263 assert_eq!(full[0].len(), 4);
2264 let short = chunk
2265 .clone()
2266 .select(&{
2267 let mut selection = Selection::with_capacity(2);
2268 selection.push(0);
2269 selection.push(2);
2270 selection
2271 })
2272 .expect("two of the four rows");
2273 let mut cut = Vec::new();
2274 prepared.evaluate(&short, &mut scratch, &mut cut).expect("the short chunk runs");
2275 assert_eq!(cut[0].len(), 2);
2276 }
2277
2278 /// An aggregate is not an expression and saying so when the pipeline is built is better than
2279 /// saying it on the first chunk.
2280 #[test]
2281 fn an_aggregate_is_refused_when_it_is_prepared() {
2282 let (schema, _) = input();
2283 let text = "Aggregate #1 groups=[] aggregates=[sum(#0.0::INTEGER)::HUGEINT]\n \
2284 Get memory.main.t AS t #0 [x::INTEGER, s::VARCHAR]";
2285 let plan = Plan::parse(text).expect("a well formed plan");
2286 let Node::Aggregate { aggregates, .. } = *plan.node(plan.root()) else {
2287 panic!("the root of that text is an aggregate");
2288 };
2289 let list = plan.expr_list(aggregates).to_vec();
2290 let error = Prepared::new(&plan, &list, &schema).expect_err("sum is not a scalar");
2291 assert!(error.message().contains("sum"), "{error}");
2292 }
2293
2294 /// How many of an expression's function steps worked something out when it was prepared, and
2295 /// whether the answer it gives is still the tree walk's answer.
2296 ///
2297 /// The count is the point of the assertion, because an answer that moved would be a bug. The
2298 /// agreement is what says the answer did not move.
2299 fn prepares(expr: &str, lifted: usize) {
2300 let (schema, _) = input();
2301 let projected = format!("{expr} AS a");
2302 let (plan, list) = projection(&projected);
2303 let prepared = Prepared::new(&plan, &list, &schema).expect("the expression resolves");
2304 assert_eq!(prepared.hoisted(), lifted, "`{expr}`");
2305 agrees(&projected);
2306 }
2307
2308 /// A pattern the user wrote is compiled where the plan is, which is once.
2309 #[test]
2310 fn a_literal_pattern_is_compiled_when_the_pipeline_is_built() {
2311 prepares("\"~~\"(#0.1::VARCHAR, 'a%'::VARCHAR)::BOOLEAN", 1);
2312 prepares("\"~~*\"(#0.1::VARCHAR, '%A%'::VARCHAR)::BOOLEAN", 1);
2313 }
2314
2315 /// A regular expression, which is the one where the compiling is worth real time.
2316 ///
2317 /// ClickBench query 29 runs one pattern over a hundred million rows, which is a hundred thousand
2318 /// chunks, and before this each of those hundred thousand compiled the pattern again.
2319 #[test]
2320 fn a_regular_expression_is_compiled_when_the_pipeline_is_built() {
2321 prepares("\"regexp_matches\"(#0.1::VARCHAR, '^a'::VARCHAR)::BOOLEAN", 1);
2322 prepares("\"regexp_replace\"(#0.1::VARCHAR, 'a'::VARCHAR, 'b'::VARCHAR)::VARCHAR", 1);
2323 }
2324
2325 /// A pattern that is not a literal, which is legal SQL and is decided per chunk as it was.
2326 #[test]
2327 fn a_pattern_that_is_not_a_literal_is_left_to_the_chunk() {
2328 prepares("\"~~\"(#0.1::VARCHAR, #0.1::VARCHAR)::BOOLEAN", 0);
2329 }
2330
2331 /// A function with nothing to work out, which is almost all of them.
2332 #[test]
2333 fn a_function_with_no_prepare_step_prepares_nothing() {
2334 prepares("\"upper\"(#0.1::VARCHAR)::VARCHAR", 0);
2335 }
2336
2337 /// How many of an expression's steps are a folded `IN`, and whether the answer still agrees.
2338 fn folds(expr: &str, sets: usize) {
2339 let (schema, _) = input();
2340 let projected = format!("{expr} AS a");
2341 let (plan, list) = projection(&projected);
2342 let prepared = Prepared::new(&plan, &list, &schema).expect("the expression resolves");
2343 assert_eq!(prepared.sets(), sets, "`{expr}`");
2344 agrees(&projected);
2345 }
2346
2347 /// What the binder writes for `x IN (1, 3)`, folded back into one lookup.
2348 ///
2349 /// The test goes through the plan's text, where the three mentions of the column are three
2350 /// expressions rather than one, which is the case `same` exists for. A plan the binder built has
2351 /// one mention and takes the first line of it.
2352 #[test]
2353 fn an_in_list_becomes_one_lookup() {
2354 folds(
2355 "((#0.0::INTEGER = 1::INTEGER)::BOOLEAN OR (#0.0::INTEGER = 3::INTEGER)::BOOLEAN)\
2356 ::BOOLEAN",
2357 1,
2358 );
2359 folds(
2360 "((#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN OR (#0.1::VARCHAR = 'z'::VARCHAR)::BOOLEAN)\
2361 ::BOOLEAN",
2362 1,
2363 );
2364 }
2365
2366 /// `NOT IN`, which the binder writes as an `AND` of inequalities and which reads the same
2367 /// lookup the other way round.
2368 #[test]
2369 fn a_not_in_list_becomes_the_same_lookup() {
2370 folds(
2371 "((#0.0::INTEGER <> 1::INTEGER)::BOOLEAN AND (#0.0::INTEGER <> 3::INTEGER)::BOOLEAN)\
2372 ::BOOLEAN",
2373 1,
2374 );
2375 }
2376
2377 /// A list with a null in it, which is the rule that makes an `IN` not a set lookup.
2378 ///
2379 /// A row that is not in the list is null rather than false, because it might have equalled the
2380 /// value the null stands for. `agrees` is what says the fold kept that, since the `OR` of
2381 /// comparisons it is checked against gets it from three valued logic for free.
2382 #[test]
2383 fn a_list_with_a_null_in_it_folds_and_keeps_the_null_rule() {
2384 folds(
2385 "((#0.0::INTEGER = 1::INTEGER)::BOOLEAN OR (#0.0::INTEGER = NULL::INTEGER)::BOOLEAN \
2386 OR (#0.0::INTEGER = 3::INTEGER)::BOOLEAN)::BOOLEAN",
2387 1,
2388 );
2389 folds(
2390 "((#0.0::INTEGER <> 1::INTEGER)::BOOLEAN AND (#0.0::INTEGER <> NULL::INTEGER)\
2391 ::BOOLEAN AND (#0.0::INTEGER <> 3::INTEGER)::BOOLEAN)::BOOLEAN",
2392 1,
2393 );
2394 }
2395
2396 /// The connectives that are not an `IN`, each for its own reason.
2397 #[test]
2398 fn a_connective_that_is_not_an_in_list_is_left_alone() {
2399 // Two different columns.
2400 folds(
2401 "((#0.0::INTEGER = 1::INTEGER)::BOOLEAN OR (#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN)\
2402 ::BOOLEAN",
2403 0,
2404 );
2405 // One equality and one of something else.
2406 folds(
2407 "((#0.0::INTEGER = 1::INTEGER)::BOOLEAN OR (#0.0::INTEGER > 3::INTEGER)::BOOLEAN)\
2408 ::BOOLEAN",
2409 0,
2410 );
2411 // The right hand side is a column rather than a literal.
2412 folds(
2413 "((#0.0::INTEGER = 1::INTEGER)::BOOLEAN OR (#0.0::INTEGER = #0.0::INTEGER)::BOOLEAN)\
2414 ::BOOLEAN",
2415 0,
2416 );
2417 // An `AND` of equalities is not a `NOT IN`, it is a predicate that is false unless the two
2418 // literals are the same. Folding it as one would answer true where it answers false.
2419 folds(
2420 "((#0.0::INTEGER = 1::INTEGER)::BOOLEAN AND (#0.0::INTEGER = 3::INTEGER)::BOOLEAN)\
2421 ::BOOLEAN",
2422 0,
2423 );
2424 }
2425
2426 /// The same thing in a filter, which is the shape it is written in.
2427 #[test]
2428 fn an_in_list_filters_the_same_rows() {
2429 filters(
2430 "((#0.0::INTEGER = 1::INTEGER)::BOOLEAN OR (#0.0::INTEGER = 3::INTEGER)::BOOLEAN)\
2431 ::BOOLEAN",
2432 );
2433 filters(
2434 "((#0.0::INTEGER <> 1::INTEGER)::BOOLEAN AND (#0.0::INTEGER <> 3::INTEGER)::BOOLEAN)\
2435 ::BOOLEAN",
2436 );
2437 // Inside a larger predicate, where the fold is one operand of the connective above it.
2438 filters(
2439 "(((#0.0::INTEGER = 1::INTEGER)::BOOLEAN OR (#0.0::INTEGER = 3::INTEGER)::BOOLEAN)\
2440 ::BOOLEAN AND (#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN)::BOOLEAN",
2441 );
2442 }
2443
2444 /// The literal side of a comparison is turned into a column when the pipeline is built.
2445 #[test]
2446 fn a_comparison_against_a_literal_builds_it_once() {
2447 let (schema, _) = input();
2448 for (expr, built) in [
2449 ("(#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN AS p", 1),
2450 ("(#0.0::INTEGER > 1::INTEGER)::BOOLEAN AS p", 1),
2451 // The literal on the left, which is the same comparison written the other way round.
2452 ("(1::INTEGER < #0.0::INTEGER)::BOOLEAN AS p", 1),
2453 // Two columns, which has no literal side to build.
2454 ("(#0.0::INTEGER = #0.0::INTEGER)::BOOLEAN AS p", 0),
2455 // Two literals, which the kernel answers once for the whole vector without reading a
2456 // column, so building one would be work that nothing reads.
2457 ("(1::INTEGER = 2::INTEGER)::BOOLEAN AS p", 0),
2458 ] {
2459 let (plan, list) = projection(expr);
2460 let prepared = Prepared::new(&plan, &list, &schema).expect("the expression resolves");
2461 assert_eq!(prepared.literals_built(), built, "`{expr}`");
2462 agrees(expr);
2463 }
2464 }
2465
2466 /// A pattern that does not compile still fails where the query said it does.
2467 ///
2468 /// Preparing is not allowed to move an error earlier. Compiling at build time and reporting
2469 /// there would raise before a row had been read, and under a `CASE` arm it would raise on a
2470 /// query whose rows never reach the call at all.
2471 #[test]
2472 fn a_pattern_that_does_not_compile_fails_on_the_chunk_and_not_before() {
2473 let (schema, chunk) = input();
2474 let (plan, list) =
2475 projection("\"regexp_matches\"(#0.1::VARCHAR, 'a('::VARCHAR)::BOOLEAN AS a");
2476 let prepared = Prepared::new(&plan, &list, &schema).expect("preparing does not compile it");
2477 assert_eq!(prepared.hoisted(), 0);
2478 let mut scratch = prepared.scratch();
2479 let mut out = Vec::new();
2480 prepared.evaluate(&chunk, &mut scratch, &mut out).expect_err("the chunk raises");
2481 }
2482}