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