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