Skip to main content

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::{Error, LogicalType, PhysicalType, Result, Value};
37use rudb_kernels::{
38    Comparison, Connective, cast, combine, compare, is_true, refine, refine_flags, selection,
39};
40use rudb_plan::{CompareOp, ConjunctionOp, Expr, ExprRef, Plan};
41use rudb_vector::{Chunk, Selection, Vector};
42
43use crate::ordering::Ordering;
44use crate::schema::Schema;
45use crate::written::written;
46
47/// The scheduler's half of the expression contract, imposed now rather than at layer eight.
48///
49/// A prepared expression is the immutable half of a pipeline and layer eight hands one of them to
50/// every thread running that pipeline. That is only sound if it holds nothing thread local, and the
51/// way to find out on the commit that breaks it rather than eight layers later is to ask the
52/// compiler here, exactly as [`Chunk`] does for the data plane.
53const _: () = {
54    const fn assert_shareable<T: Send + Sync>() {}
55    assert_shareable::<Prepared>();
56};
57
58/// One or more bound expressions, flattened and resolved against a schema.
59///
60/// Built once per pipeline with [`Prepared::new`] and evaluated per chunk with
61/// [`Prepared::evaluate`] or [`Prepared::evaluate_one`], each of which wants the [`Scratch`] that
62/// [`Prepared::scratch`] hands out.
63#[derive(Debug)]
64pub struct Prepared {
65    /// The nodes in post order, so every node's operands have already been computed when it runs.
66    steps: Vec<Step>,
67    /// The type each step produces, indexed the same way as `steps`.
68    ///
69    /// A parallel array rather than a field in the variant, for the reason [`Expr`] gives: a
70    /// [`LogicalType`] owns a `Vec` for its nested cases and putting one in every variant would make
71    /// the common variants several times larger for the benefit of the rare ones.
72    types: Vec<LogicalType>,
73    /// The operand lists of the steps that have one, as runs of step indices.
74    operands: Vec<usize>,
75    /// The last step that reads each step's slot, or `usize::MAX` for one nothing reads.
76    ///
77    /// A slot is emptied as soon as the step that was the last to read it has run. Keeping every
78    /// intermediate alive to the end of the array instead is what the first measured version of this
79    /// did, and a chain of eight additions was slower prepared than walked because of it: nine live
80    /// intermediates at eight kilobytes each is seventy two kilobytes of working set where the tree
81    /// walk has two, and two is the pair the allocator hands back and forth and that stays in L1.
82    /// Everything else about the prepared form was faster and this one thing paid all of it back.
83    last_use: Vec<usize>,
84    /// The step index each expression this was built from ends at.
85    roots: Vec<usize>,
86}
87
88/// One node of a flattened expression.
89///
90/// A step refers to its operands by their index in [`Prepared::steps`], which is always smaller than
91/// its own because the array is in post order.
92#[derive(Debug)]
93enum Step {
94    /// A column of the chunk, by resolved position.
95    ///
96    /// This step computes nothing. Its slot stays empty and an operand that names it is read out of
97    /// the chunk, which is the whole of what makes a column reference free rather than a copy.
98    Column(usize),
99    /// A literal, materialized into a constant vector as long as the chunk.
100    Constant(Value),
101    /// A cast to this step's own type.
102    Cast {
103        /// The step being cast.
104        input: usize,
105        /// Whether a failed cast yields null instead of raising.
106        try_cast: bool,
107    },
108    /// A binary comparison.
109    Compare {
110        /// Which comparison.
111        op: Comparison,
112        /// The left operand's step.
113        left: usize,
114        /// The right operand's step.
115        right: usize,
116    },
117    /// An `AND` or `OR` over a run of [`Prepared::operands`].
118    Conjunction {
119        /// Which connective.
120        op: Connective,
121        /// Where the operand list starts.
122        start: usize,
123        /// How many operands it has.
124        len: usize,
125    },
126    /// A scalar function over a run of [`Prepared::operands`].
127    Function {
128        /// The resolved function name, held here so the plan is not consulted per chunk.
129        name: String,
130        /// How the call is written, for the one error message that quotes it.
131        ///
132        /// Rendered when the pipeline is built rather than when a chunk arrives, because the plan
133        /// is here and is not there. It is a short string per function node in the query and it is
134        /// built once, which is a different cost from the tree walk's, where the plan is still to
135        /// hand and the rendering can wait until the row that fails.
136        written: String,
137        /// Where the argument list starts.
138        start: usize,
139        /// How many arguments it has.
140        len: usize,
141    },
142    /// A searched `CASE`, whose branches are prepared expressions of their own.
143    ///
144    /// Nested rather than flattened into the same array because a branch is not evaluated over the
145    /// chunk, it is evaluated over the rows no earlier arm claimed, and a step in the outer array
146    /// would have no way to say that. The selection threaded form in #57 replaces this whole
147    /// variant, and when it does the branches stop being separate arrays.
148    Case {
149        /// The `WHEN`/`THEN` pairs, in order.
150        arms: Vec<PreparedArm>,
151        /// The `ELSE`, if there is one. Absent means null.
152        otherwise: Option<Prepared>,
153    },
154}
155
156/// One `WHEN`/`THEN` pair of a prepared [`Step::Case`].
157#[derive(Debug)]
158struct PreparedArm {
159    /// The condition.
160    when: Prepared,
161    /// The result if the condition is true.
162    then: Prepared,
163}
164
165/// The per chunk working space of one [`Prepared`].
166///
167/// One per pipeline instance and never shared, which is the mutable half of the split the module
168/// documentation describes. It is handed back in rather than made inside [`Prepared::evaluate`] so
169/// that the array of slots survives from one chunk to the next instead of being allocated a hundred
170/// thousand times over a scan.
171#[derive(Debug)]
172pub struct Scratch {
173    /// What each step produced, or `None` for a step that produces nothing and for one that has not
174    /// run yet.
175    slots: Vec<Option<Vector>>,
176    /// What each connective step has learned about its operands, indexed by step.
177    ///
178    /// Empty for every step that is not a connective and for a connective a filter has not reached
179    /// yet, since it is built the first time one runs and the shape it needs is not known before
180    /// then. This is the mutable half of the adaptive ordering and it is here rather than in
181    /// [`Prepared`] because a prepared expression is shared by every thread running the pipeline.
182    orders: Vec<Option<Ordering>>,
183}
184
185impl Scratch {
186    /// The order a connective's operands are run in.
187    ///
188    /// For the tests that say the learning reached the walk. Nothing in the engine asks a scratch
189    /// this, because the walk is the only thing that reads an ordering and it reads its own.
190    #[cfg(test)]
191    fn order(&self, step: usize) -> Option<&[usize]> {
192        self.orders[step].as_ref().map(Ordering::order)
193    }
194}
195
196impl Prepared {
197    /// Prepares `exprs` against `schema`.
198    ///
199    /// # Errors
200    ///
201    /// If a column reference names a binding the schema does not have, or if an aggregate appears
202    /// where an ordinary expression was expected. Both are failures of the plan rather than of the
203    /// data, which is why they are found here, once, rather than on some chunk in the middle of a
204    /// scan.
205    pub fn new(plan: &Plan, exprs: &[ExprRef], schema: &Schema) -> Result<Self> {
206        let mut prepared = Self {
207            steps: Vec::new(),
208            types: Vec::new(),
209            operands: Vec::new(),
210            last_use: Vec::new(),
211            roots: Vec::new(),
212        };
213        for &expr in exprs {
214            let root = prepared.push(plan, expr, schema)?;
215            prepared.roots.push(root);
216        }
217        prepared.last_use = prepared.last_uses();
218        Ok(prepared)
219    }
220
221    /// Which step is the last to read each step, computed once when the expression is prepared.
222    ///
223    /// A root is never freed, because the whole point of running the array was to produce it. A
224    /// step nothing reads and that is not a root cannot happen, since every step is pushed by the
225    /// node that wanted it, but saying `usize::MAX` rather than asserting that keeps this a fact
226    /// about the array rather than a claim about the builder.
227    fn last_uses(&self) -> Vec<usize> {
228        let mut last = vec![usize::MAX; self.steps.len()];
229        for index in 0..self.steps.len() {
230            self.for_each_operand(index, |operand| last[operand] = index);
231        }
232        for &root in &self.roots {
233            last[root] = usize::MAX;
234        }
235        last
236    }
237
238    /// Visits the steps one step reads, whatever shape its operands are held in.
239    fn for_each_operand(&self, index: usize, mut visit: impl FnMut(usize)) {
240        match &self.steps[index] {
241            // A case's branches are arrays of their own and read nothing out of this one.
242            Step::Column(_) | Step::Constant(_) | Step::Case { .. } => {}
243            Step::Cast { input, .. } => visit(*input),
244            Step::Compare { left, right, .. } => {
245                visit(*left);
246                visit(*right);
247            }
248            Step::Conjunction { start, len, .. } | Step::Function { start, len, .. } => {
249                for &operand in &self.operands[*start..*start + *len] {
250                    visit(operand);
251                }
252            }
253        }
254    }
255
256    /// Prepares one expression, which is the common case and saves the caller a slice.
257    ///
258    /// # Errors
259    ///
260    /// Whatever [`Prepared::new`] reports.
261    pub fn one(plan: &Plan, expr: ExprRef, schema: &Schema) -> Result<Self> {
262        Self::new(plan, &[expr], schema)
263    }
264
265    /// Working space sized for this expression.
266    #[must_use]
267    pub fn scratch(&self) -> Scratch {
268        Scratch {
269            slots: (0..self.steps.len()).map(|_| None).collect(),
270            orders: (0..self.steps.len()).map(|_| None).collect(),
271        }
272    }
273
274    /// How many expressions this was built from.
275    #[must_use]
276    pub fn len(&self) -> usize {
277        self.roots.len()
278    }
279
280    /// Whether it was built from no expressions at all.
281    #[must_use]
282    pub fn is_empty(&self) -> bool {
283        self.roots.is_empty()
284    }
285
286    /// Evaluates every expression over `chunk`, appending one vector each to `out`.
287    ///
288    /// Appends rather than returns a `Vec`, so a caller in a loop reuses one buffer.
289    ///
290    /// # Errors
291    ///
292    /// Anything a kernel reports, on the first expression that reports it.
293    pub fn evaluate(
294        &self,
295        chunk: &Chunk,
296        scratch: &mut Scratch,
297        out: &mut Vec<Vector>,
298    ) -> Result<()> {
299        self.run(chunk, scratch)?;
300        for &root in &self.roots {
301            // The one place a column is copied, and it is copied because the caller is taking
302            // ownership of a vector that has to outlive the chunk it came from. `SELECT a` is that
303            // shape and a projection of a bare column is the only expression where it happens.
304            match self.steps[root] {
305                Step::Column(position) => out.push(chunk.column(position)?.clone()),
306                _ => out.push(scratch.slots[root].take().ok_or_else(|| missing(root))?),
307            }
308        }
309        Ok(())
310    }
311
312    /// Evaluates a single expression over `chunk`, handing back a reference to the answer.
313    ///
314    /// A reference rather than a vector, because the caller of this is a filter, which reads the
315    /// flags to build a selection and then drops them. Nothing about that wants ownership, and a
316    /// predicate that is a bare column reference, which `WHERE flag` is, would otherwise copy the
317    /// column to hand it over.
318    ///
319    /// # Errors
320    ///
321    /// Anything a kernel reports, and an internal error if this was not built from exactly one
322    /// expression.
323    pub fn evaluate_one<'s>(
324        &'s self,
325        chunk: &'s Chunk,
326        scratch: &'s mut Scratch,
327    ) -> Result<&'s Vector> {
328        let [root] = self.roots[..] else {
329            return Err(Error::internal(format!(
330                "evaluate_one over a prepared expression of {} roots",
331                self.roots.len()
332            )));
333        };
334        self.run(chunk, scratch)?;
335        self.operand(root, chunk, &scratch.slots)
336    }
337
338    /// Evaluates a single expression as a filter, handing back the rows it keeps.
339    ///
340    /// The difference between this and [`evaluate_one`](Self::evaluate_one) followed by
341    /// [`selection`] is the whole of what a threaded filter is. An `AND` evaluated as an expression
342    /// runs every conjunct over every row and then combines the flag vectors, so a predicate of four
343    /// conjuncts that each pass a fifth of the rows does five times the work of one that stops
344    /// looking at a row as soon as a conjunct rejects it. TPC-H Q6 is exactly that predicate.
345    ///
346    /// So the conjuncts of a top level `AND` are run one at a time, each over the rows the ones
347    /// before it left, and the moment nothing is left the rest of the predicate is not run at all.
348    /// The order they run in starts as the order the plan gives and then moves, because which
349    /// conjunct is worth running first is a question about the data and the scan is the thing
350    /// holding the answer. The `ordering` module has what is measured and how.
351    ///
352    /// A top level `OR` is threaded the same way against the complement. A row the first branch
353    /// accepts is a row the filter keeps whatever the rest of the predicate says about it, so each
354    /// branch is run over the rows no branch before it accepted, and the moment every row has been
355    /// accepted the rest of the predicate is not run either. That is the mirror of the `AND` case
356    /// and not an approximation of it: the answer is the same set of rows, because `OR` over three
357    /// valued logic is true wherever any branch is true and nothing a later branch says can take a
358    /// row back. It is worth less than the `AND` case in practice, since an `OR` of selective
359    /// branches leaves almost every row in play for the branch after, and it is worth having anyway
360    /// because the cost of finding that out is one merge per branch.
361    ///
362    /// What is threaded is the operand's own comparison rather than the whole of its subtree. A
363    /// conjunct of `a + b > 5` still adds over the whole chunk, because the scalar kernels take a
364    /// vector rather than a selection, and it is the comparison and everything downstream of it that
365    /// reads only the rows still in play. An operand that is a bare column or a function produces
366    /// flags over the chunk and is narrowed with [`refine_flags`], which is what keeps one awkward
367    /// operand from putting the others back on the unthreaded path. An operand that is itself a
368    /// connective recurses, so the two conjuncts of each half of `(a AND b) OR (c AND d)` are
369    /// threaded the same way the halves are.
370    ///
371    /// None of this is available to a projection. `SELECT a > 5 AND b LIKE 'x%'` wants a value per
372    /// row and the rows a selection dropped have no value in it, so [`evaluate`](Self::evaluate) and
373    /// [`evaluate_one`](Self::evaluate_one) evaluate the whole tree over the whole chunk and combine
374    /// flags. The two are separate entry points picked when the pipeline is built rather than one
375    /// path with a flag in it, because conflating them is a wrong answer rather than a slow one.
376    ///
377    /// # Errors
378    ///
379    /// Anything a kernel reports, and an internal error if this was not built from exactly one
380    /// expression.
381    pub fn evaluate_filter(&self, chunk: &Chunk, scratch: &mut Scratch) -> Result<Selection> {
382        let [root] = self.roots[..] else {
383            return Err(Error::internal(format!(
384                "evaluate_filter over a prepared expression of {} roots",
385                self.roots.len()
386            )));
387        };
388        scratch.slots.clear();
389        scratch.slots.resize_with(self.steps.len(), || None);
390        // A predicate that is not a connective at all is the same walk over one operand, which is
391        // where [`thread`](Self::thread) starts: it runs the tree and turns the flags into a
392        // selection, with no narrowing to do because nothing has narrowed anything yet.
393        self.thread(root, 0, chunk, scratch, None)
394    }
395
396    /// The operands of one connective, run in order, each over the rows the ones before it left.
397    ///
398    /// `live` is the rows this connective has to decide about and `None` means every row of the
399    /// chunk, which is not the same as a selection of all of them: it lets the first operand take
400    /// the unthreaded kernel rather than a pass over an identity selection. The answer is the rows
401    /// out of `live` the connective is true for.
402    ///
403    /// The walk is the same for both connectives and only the bookkeeping differs. `AND` carries the
404    /// rows every operand so far has kept, so each answer replaces it. `OR` carries the rows no
405    /// operand so far has accepted, so each answer comes out of it and the rows the connective keeps
406    /// are the ones that went missing along the way.
407    ///
408    /// The operand is not `steps[begin..=operand]` evaluated and then narrowed. Its subtree is run
409    /// over the whole chunk and it is the operand itself that reads only the rows in play, except
410    /// where the operand is another connective, which recurses and threads its own operands from
411    /// here rather than falling back to a flag vector. That is what makes `(a AND b) OR (c AND d)`
412    /// four threaded comparisons rather than two threaded ones and two flag passes.
413    fn branches(
414        &self,
415        index: usize,
416        begin: usize,
417        chunk: &Chunk,
418        scratch: &mut Scratch,
419        live: Option<&Selection>,
420    ) -> Result<Selection> {
421        let Step::Conjunction { op, start, len } = self.steps[index] else {
422            return Err(Error::internal("a connective walk over a step that is not a connective"));
423        };
424        let operands = &self.operands[start..start + len];
425        let rows = chunk.len();
426        // Out of the scratch for the length of the walk, because the walk runs steps and running a
427        // step wants the scratch. It goes back at the end, which is also where it learns. A walk
428        // that fails leaves the slot empty and the next chunk starts the connective over, which is
429        // a history lost on a query that is about to stop running anyway.
430        let mut order = scratch.orders[index]
431            .take()
432            .unwrap_or_else(|| Ordering::new(op, self.weights(operands, begin)));
433        let mut carried: Option<Selection> = live.cloned();
434        for slot in 0..len {
435            if carried.as_ref().is_some_and(Selection::is_empty) {
436                break;
437            }
438            let which = order.at(slot);
439            let operand = operands[which];
440            // The array is in post order and an operand's whole subtree sits between the operand
441            // before it and the operand itself, which is a range the run order cannot move. That is
442            // what lets the operands run in any order at all without a second structure to say
443            // where each one starts.
444            let from = if which == 0 { begin } else { operands[which - 1] + 1 };
445            let given = carried.as_ref().map_or(rows, Selection::len);
446            let answered = self.thread(operand, from, chunk, scratch, carried.as_ref())?;
447            order.observed(which, given, answered.len());
448            carried = Some(match (op, carried) {
449                (Connective::And, _) => answered,
450                (Connective::Or, None) => answered.complement(rows),
451                (Connective::Or, Some(carried)) => carried.without(&answered),
452            });
453            // An operand's subtree is its own, because nothing here looks for a common subexpression
454            // and so no step outside the range is reading one inside it.
455            for step in from..=operand {
456                scratch.slots[step] = None;
457            }
458        }
459        order.relearn();
460        scratch.orders[index] = Some(order);
461        Ok(match (op, carried) {
462            // A connective with no operands, which the binder does not build and which is answered
463            // here rather than left to index arithmetic: an empty `AND` is every row and an empty
464            // `OR` is none.
465            (Connective::And, None) => live.cloned().unwrap_or_else(|| Selection::identity(rows)),
466            (Connective::And, Some(kept)) => kept,
467            (Connective::Or, None) => Selection::empty(),
468            (Connective::Or, Some(missed)) => match live {
469                None => missed.complement(rows),
470                Some(live) => live.without(&missed),
471            },
472        })
473    }
474
475    /// What each operand of a connective costs to run over a chunk, for the ordering to divide by.
476    ///
477    /// An operand costs what its whole subtree costs, which is the steps from where the operand
478    /// before it ended up to the operand itself.
479    fn weights(&self, operands: &[usize], begin: usize) -> Vec<f64> {
480        let mut costs = Vec::with_capacity(operands.len());
481        let mut from = begin;
482        for &operand in operands {
483            costs.push((from..=operand).map(|step| self.weight(step)).sum());
484            from = operand + 1;
485        }
486        costs
487    }
488
489    /// Roughly what one step costs to run over a chunk, against a comparison of two fixed width
490    /// columns as the unit.
491    ///
492    /// A ranking rather than a prediction. Nothing downstream reads the number itself, only which
493    /// of two of them is larger, and the differences that decide an order are the big ones: a
494    /// column reference costs nothing because it is read in place, a string function costs many
495    /// times what an integer comparison costs, and a comparison over a variable length type costs
496    /// several times what the same comparison over a fixed width one costs. Everything finer than
497    /// that is below the noise of what the window is measuring anyway.
498    fn weight(&self, index: usize) -> f64 {
499        match &self.steps[index] {
500            // Read straight out of the chunk at the point an operand is wanted, so there is no step
501            // to run and nothing to charge for.
502            Step::Column(_) => 0.0,
503            // One vector built per chunk, however many rows the chunk has.
504            Step::Constant(_) => 0.25,
505            // The operands carry the cost of a connective, and they are steps of their own.
506            Step::Conjunction { .. } => 0.0,
507            Step::Cast { input, .. } => 2.0 * touching(&self.types[*input]),
508            Step::Compare { left, .. } => touching(&self.types[*left]),
509            Step::Function { start, len, .. } => {
510                let widest = self.operands[*start..*start + *len]
511                    .iter()
512                    .map(|&argument| touching(&self.types[argument]))
513                    .fold(1.0, f64::max);
514                4.0 * widest
515            }
516            // A branch per arm, each of which is a prepared expression of its own that this does
517            // not look inside. Charging for the arms alone understates it and says the right thing
518            // about the order, which is that a `CASE` is not what you want in front.
519            Step::Case { arms, .. } => 4.0 * arms.len() as f64,
520        }
521    }
522
523    /// One operand of a connective, over the rows it is still worth asking about.
524    ///
525    /// `begin` is the first step of the operand's subtree, which the caller knows because the steps
526    /// are in post order.
527    fn thread(
528        &self,
529        index: usize,
530        begin: usize,
531        chunk: &Chunk,
532        scratch: &mut Scratch,
533        live: Option<&Selection>,
534    ) -> Result<Selection> {
535        if matches!(self.steps[index], Step::Conjunction { .. }) {
536            return self.branches(index, begin, chunk, scratch, live);
537        }
538        for step in begin..index {
539            self.run_step(step, chunk, scratch)?;
540        }
541        if let Step::Compare { op, left, right } = self.steps[index] {
542            let left = self.operand(left, chunk, &scratch.slots)?;
543            let right = self.operand(right, chunk, &scratch.slots)?;
544            return match live {
545                // The first operand has every row in play, and asking the threaded kernel for that
546                // would be a pass over an identity selection the unthreaded one does not need.
547                None => Ok(selection(&compare(op, left, right)?, chunk.len())),
548                Some(live) => refine(op, left, right, live),
549            };
550        }
551        self.run_step(index, chunk, scratch)?;
552        let flags = self.operand(index, chunk, &scratch.slots)?;
553        match live {
554            None => Ok(selection(flags, chunk.len())),
555            Some(live) => refine_flags(flags, live),
556        }
557    }
558
559    /// Runs every step in order, filling the slots.
560    fn run(&self, chunk: &Chunk, scratch: &mut Scratch) -> Result<()> {
561        scratch.slots.clear();
562        scratch.slots.resize_with(self.steps.len(), || None);
563        for index in 0..self.steps.len() {
564            self.run_step(index, chunk, scratch)?;
565        }
566        Ok(())
567    }
568
569    /// Runs one step and empties the slot of every operand this was the last step to read.
570    fn run_step(&self, index: usize, chunk: &Chunk, scratch: &mut Scratch) -> Result<()> {
571        let produced = self.step(index, chunk, &scratch.slots)?;
572        scratch.slots[index] = produced;
573        let slots = &mut scratch.slots;
574        self.for_each_operand(index, |operand| {
575            if self.last_use[operand] == index {
576                slots[operand] = None;
577            }
578        });
579        Ok(())
580    }
581
582    /// Runs one step, given what the steps before it produced.
583    fn step(
584        &self,
585        index: usize,
586        chunk: &Chunk,
587        slots: &[Option<Vector>],
588    ) -> Result<Option<Vector>> {
589        let ty = &self.types[index];
590        let produced = match &self.steps[index] {
591            Step::Column(_) => None,
592            Step::Constant(value) => Some(Vector::constant(ty.clone(), value.clone(), chunk.len())),
593            Step::Cast { input, try_cast } => {
594                Some(cast(self.operand(*input, chunk, slots)?, ty, *try_cast)?)
595            }
596            Step::Compare { op, left, right } => Some(compare(
597                *op,
598                self.operand(*left, chunk, slots)?,
599                self.operand(*right, chunk, slots)?,
600            )?),
601            Step::Conjunction { op, start, len } => {
602                Some(
603                    self.with_operands(*start, *len, chunk, slots, |children| {
604                        combine(*op, children)
605                    })?,
606                )
607            }
608            Step::Function { name, written, start, len } => {
609                Some(self.with_operands(*start, *len, chunk, slots, |args| {
610                    rudb_kernels::call(name, args, ty, Some(&|| written.clone()))
611                })?)
612            }
613            Step::Case { arms, otherwise } => {
614                Some(self.case(chunk, arms, otherwise.as_ref(), ty)?)
615            }
616        };
617        Ok(produced)
618    }
619
620    /// The vector a step produced, or the chunk's column if the step is a column reference.
621    fn operand<'v>(
622        &self,
623        index: usize,
624        chunk: &'v Chunk,
625        slots: &'v [Option<Vector>],
626    ) -> Result<&'v Vector> {
627        if let Step::Column(position) = self.steps[index] {
628            return chunk.column(position);
629        }
630        slots[index].as_ref().ok_or_else(|| missing(index))
631    }
632
633    /// Hands a kernel the references to an operand list, without allocating for the usual widths.
634    ///
635    /// One, two and three because those are what a bound tree is made of: every scalar function in
636    /// the catalog is unary or binary, a comparison is binary, and a conjunction is two or three
637    /// often enough to be worth a line. A stack array for those means a chain of eight additions
638    /// makes zero allocations for its operand lists over a chunk instead of eight, and eight
639    /// allocations a chunk at the rate a pipeline produces chunks is a real number rather than a
640    /// tidiness argument. Anything wider falls back to [`gather`](Self::gather), which is a `Vec`
641    /// of pointers and still moves no data.
642    fn with_operands<'v, T>(
643        &self,
644        start: usize,
645        len: usize,
646        chunk: &'v Chunk,
647        slots: &'v [Option<Vector>],
648        run: impl FnOnce(&[&'v Vector]) -> Result<T>,
649    ) -> Result<T> {
650        match self.operands[start..start + len] {
651            [a] => run(&[self.operand(a, chunk, slots)?]),
652            [a, b] => run(&[self.operand(a, chunk, slots)?, self.operand(b, chunk, slots)?]),
653            [a, b, c] => run(&[
654                self.operand(a, chunk, slots)?,
655                self.operand(b, chunk, slots)?,
656                self.operand(c, chunk, slots)?,
657            ]),
658            _ => {
659                let gathered = self.gather(start, len, chunk, slots)?;
660                run(&gathered)
661            }
662        }
663    }
664
665    /// References to an operand list, for a kernel that takes a slice of them.
666    ///
667    /// The `Vec` here is the allocation the module documentation names: it holds pointers rather
668    /// than vectors, so it is a dozen bytes an operand and no data moves.
669    fn gather<'v>(
670        &self,
671        start: usize,
672        len: usize,
673        chunk: &'v Chunk,
674        slots: &'v [Option<Vector>],
675    ) -> Result<Vec<&'v Vector>> {
676        let mut gathered = Vec::with_capacity(len);
677        for &operand in &self.operands[start..start + len] {
678            gathered.push(self.operand(operand, chunk, slots)?);
679        }
680        Ok(gathered)
681    }
682
683    /// A searched `CASE` over the rows no earlier arm claimed.
684    ///
685    /// The same shape [`evaluate`](crate::evaluate) has, because the thing that makes it that shape
686    /// is a correctness rule rather than a performance one: `CASE WHEN x <> 0 THEN 1 / x ELSE 0 END`
687    /// divides by zero on the rows the arm excludes if the arm is evaluated for them. What is left
688    /// of it after #57 is the same rule expressed as a selection rather than as a narrowed chunk,
689    /// with the answers scattered back instead of assembled out of a `Vec<Value>`.
690    fn case(
691        &self,
692        chunk: &Chunk,
693        arms: &[PreparedArm],
694        otherwise: Option<&Prepared>,
695        ty: &LogicalType,
696    ) -> Result<Vector> {
697        let mut answers = vec![Value::Null; chunk.len()];
698        let mut pending: Vec<usize> = (0..chunk.len()).collect();
699        for arm in arms {
700            if pending.is_empty() {
701                break;
702            }
703            let narrowed = narrow(chunk, &pending)?;
704            let mut scratch = arm.when.scratch();
705            let flags = arm.when.evaluate_one(&narrowed, &mut scratch)?;
706            let mut taken = Vec::new();
707            let mut still = Vec::new();
708            // row at a time: the scatter that replaces these three loops is #57, and this variant
709            // goes with it.
710            for (at, &row) in pending.iter().enumerate() {
711                if is_true(&flags.value_at(at)) {
712                    taken.push((at, row));
713                } else {
714                    still.push(row);
715                }
716            }
717            if !taken.is_empty() {
718                let positions: Vec<usize> = taken.iter().map(|&(at, _)| at).collect();
719                let matched = narrow(&narrowed, &positions)?;
720                let mut scratch = arm.then.scratch();
721                let results = arm.then.evaluate_one(&matched, &mut scratch)?;
722                // row at a time: the scatter this wants is #57, same as the loop above.
723                for (slot, &(_, row)) in taken.iter().enumerate() {
724                    answers[row] = results.value_at(slot);
725                }
726            }
727            pending = still;
728        }
729        if let Some(otherwise) = otherwise {
730            if !pending.is_empty() {
731                let narrowed = narrow(chunk, &pending)?;
732                let mut scratch = otherwise.scratch();
733                let results = otherwise.evaluate_one(&narrowed, &mut scratch)?;
734                // row at a time: the scatter this wants is #57, same as the two above.
735                for (slot, &row) in pending.iter().enumerate() {
736                    answers[row] = results.value_at(slot);
737                }
738            }
739        }
740        Vector::from_values(ty.clone(), &answers)
741    }
742
743    /// Flattens one expression, appending its steps and returning the index of its last one.
744    fn push(&mut self, plan: &Plan, expr: ExprRef, schema: &Schema) -> Result<usize> {
745        let ty = plan.expr_type(expr).clone();
746        let step = match *plan.expr(expr) {
747            Expr::Column(binding) => {
748                let position = schema.position_of(binding).ok_or_else(|| {
749                    Error::internal(format!(
750                        "column #{}.{} is not in the schema this operator was given",
751                        binding.table, binding.column
752                    ))
753                })?;
754                Step::Column(position)
755            }
756            Expr::Constant(reference) => Step::Constant(plan.value(reference).clone()),
757            Expr::Cast { input, try_cast } => {
758                Step::Cast { input: self.push(plan, input, schema)?, try_cast }
759            }
760            Expr::Compare { op, left, right } => Step::Compare {
761                op: comparison(op),
762                left: self.push(plan, left, schema)?,
763                right: self.push(plan, right, schema)?,
764            },
765            Expr::Conjunction { op, children } => {
766                let (start, len) = self.push_list(plan, plan.expr_list(children), schema)?;
767                Step::Conjunction { op: connective(op), start, len }
768            }
769            Expr::Function { name, args } => {
770                let (start, len) = self.push_list(plan, plan.expr_list(args), schema)?;
771                Step::Function {
772                    name: plan.string(name).to_string(),
773                    written: written(plan, expr, schema),
774                    start,
775                    len,
776                }
777            }
778            Expr::Aggregate { name, .. } => {
779                return Err(Error::internal(format!(
780                    "the {} aggregate was evaluated as an ordinary expression",
781                    plan.string(name)
782                )));
783            }
784            Expr::Case { arms, otherwise } => {
785                let mut prepared = Vec::new();
786                for &arm in plan.arm_list(arms) {
787                    prepared.push(PreparedArm {
788                        when: Self::one(plan, arm.when, schema)?,
789                        then: Self::one(plan, arm.then, schema)?,
790                    });
791                }
792                let otherwise = match otherwise {
793                    Some(otherwise) => Some(Self::one(plan, otherwise, schema)?),
794                    None => None,
795                };
796                Step::Case { arms: prepared, otherwise }
797            }
798        };
799        self.steps.push(step);
800        self.types.push(ty);
801        Ok(self.steps.len() - 1)
802    }
803
804    /// Flattens a list of expressions and records where its operand run starts and how long it is.
805    ///
806    /// The operand run is written after every child has been flattened rather than as they go,
807    /// because a child that is itself a list would otherwise interleave its run with this one.
808    fn push_list(
809        &mut self,
810        plan: &Plan,
811        exprs: &[ExprRef],
812        schema: &Schema,
813    ) -> Result<(usize, usize)> {
814        let mut indices = Vec::with_capacity(exprs.len());
815        for &expr in exprs {
816            indices.push(self.push(plan, expr, schema)?);
817        }
818        let start = self.operands.len();
819        let len = indices.len();
820        self.operands.extend(indices);
821        Ok((start, len))
822    }
823}
824
825/// What touching a value of this type costs, against a fixed width one as the unit.
826///
827/// A variable length value is a pointer to follow and a length that is not the same twice, and a
828/// nested one is that per element. Four is not measured, and what it has to be is large enough that
829/// the ordering puts a fixed width comparison in front of a string one and small enough that it does
830/// not put one in front of a string comparison that rejects every row.
831fn touching(ty: &LogicalType) -> f64 {
832    match ty.physical() {
833        PhysicalType::Varlen => 4.0,
834        PhysicalType::List | PhysicalType::Array | PhysicalType::Struct => 8.0,
835        _ => 1.0,
836    }
837}
838
839/// The error for a slot that should have held something and did not.
840///
841/// This cannot happen while the array is in post order, since every operand's index is smaller than
842/// the index of the step using it and every step runs in order. It is an error rather than a panic
843/// because the property it depends on is a property of [`Prepared::push`], and the day somebody
844/// writes a pass that reorders the array is the day it stops holding.
845fn missing(index: usize) -> Error {
846    Error::internal(format!("step {index} was used as an operand before it produced anything"))
847}
848
849/// The chunk cut down to the given rows.
850///
851/// The reason `CASE` is written with this rather than by evaluating every arm over the whole chunk
852/// and picking afterwards. `CASE WHEN x <> 0 THEN 1 // x ELSE 0 END` divides by zero on the rows the
853/// arm does not apply to if the arm is evaluated for them, and a `CASE` that raises on a row it was
854/// written to exclude is the classic wrong answer this shape prevents.
855pub(crate) fn narrow(chunk: &Chunk, rows: &[usize]) -> Result<Chunk> {
856    let mut selection = Selection::with_capacity(rows.len());
857    for &row in rows {
858        selection.push(row);
859    }
860    chunk.clone().select(&selection)
861}
862
863/// The kernels' comparison for the plan's.
864///
865/// A translation rather than one shared enum, because the kernels are rank 3 and the plan is rank
866/// 9. This function is the whole of what that separation costs.
867pub(crate) fn comparison(op: CompareOp) -> Comparison {
868    match op {
869        CompareOp::Equal => Comparison::Equal,
870        CompareOp::NotEqual => Comparison::NotEqual,
871        CompareOp::Less => Comparison::Less,
872        CompareOp::LessOrEqual => Comparison::LessOrEqual,
873        CompareOp::Greater => Comparison::Greater,
874        CompareOp::GreaterOrEqual => Comparison::GreaterOrEqual,
875        CompareOp::DistinctFrom => Comparison::DistinctFrom,
876        CompareOp::NotDistinctFrom => Comparison::NotDistinctFrom,
877    }
878}
879
880/// The kernels' connective for the plan's.
881pub(crate) fn connective(op: ConjunctionOp) -> Connective {
882    match op {
883        ConjunctionOp::And => Connective::And,
884        ConjunctionOp::Or => Connective::Or,
885    }
886}
887
888#[cfg(test)]
889mod tests {
890    use rudb_common::{Field, LogicalType, Value};
891    use rudb_kernels::is_true;
892    use rudb_plan::{ExprRef, Node, Plan};
893    use rudb_vector::{Chunk, Selection, Vector};
894
895    use super::{Prepared, narrow};
896    use crate::expr::evaluate;
897    use crate::schema::Schema;
898
899    /// Two columns with a null in each, because every disagreement between these two evaluators
900    /// that is worth finding is a disagreement about which rows are null.
901    fn input() -> (Schema, Chunk) {
902        let schema = Schema::numbered(
903            vec![Field::new("x", LogicalType::Integer), Field::new("s", LogicalType::Varchar)],
904            0,
905        );
906        let x = Vector::from_values(
907            LogicalType::Integer,
908            &[Value::Integer(3), Value::Integer(1), Value::Null, Value::Integer(2)],
909        )
910        .expect("four integers");
911        let s = Vector::from_values(
912            LogicalType::Varchar,
913            &[
914                Value::Varchar("a".to_string()),
915                Value::Null,
916                Value::Varchar("c".to_string()),
917                Value::Varchar("a".to_string()),
918            ],
919        )
920        .expect("four strings");
921        (schema, Chunk::new(vec![x, s]).expect("two columns of four rows"))
922    }
923
924    /// The expressions of a projection written in the plan's textual form, over the two columns
925    /// [`input`] produces.
926    ///
927    /// Going through the text rather than the arena builders for the reason the other test module
928    /// gives: a test that says what it evaluates in the notation a plan dump uses is a test whose
929    /// failure can be pasted into a plan and vice versa.
930    fn projection(exprs: &str) -> (Plan, Vec<ExprRef>) {
931        let text =
932            format!("Project #1 [{exprs}]\n  Get memory.main.t AS t #0 [x::INTEGER, s::VARCHAR]");
933        let plan = Plan::parse(&text).expect("a well formed plan");
934        let Node::Project { exprs, .. } = *plan.node(plan.root()) else {
935            panic!("the root of that text is a projection");
936        };
937        let list = plan.expr_list(exprs).to_vec();
938        (plan, list)
939    }
940
941    /// Every expression shape, evaluated both ways over the same chunk.
942    ///
943    /// This is the agreement the module documentation claims and it is the only thing that makes
944    /// the prepared form safe to put in front of the tree walk. The generated well typed trees the
945    /// test gate of #57 asks for are a wider version of this and are worth building once the
946    /// selection threaded shapes exist to disagree about.
947    fn agrees(exprs: &str) {
948        let (schema, chunk) = input();
949        let (plan, list) = projection(exprs);
950        let prepared = Prepared::new(&plan, &list, &schema).expect("the expressions resolve");
951        let mut scratch = prepared.scratch();
952        let mut fast = Vec::new();
953        prepared.evaluate(&chunk, &mut scratch, &mut fast).expect("the prepared form runs");
954        for (at, &expr) in list.iter().enumerate() {
955            let slow = evaluate(&plan, expr, &schema, &chunk).expect("the tree walk runs");
956            for row in 0..chunk.len() {
957                assert_eq!(
958                    fast[at].value_at(row),
959                    slow.value_at(row),
960                    "expression {at} of `{exprs}` at row {row}"
961                );
962            }
963        }
964    }
965
966    #[test]
967    fn a_column_reference_agrees() {
968        agrees("#0.0::INTEGER AS a, #0.1::VARCHAR AS b");
969    }
970
971    #[test]
972    fn a_constant_agrees() {
973        agrees("7::INTEGER AS a, NULL::INTEGER AS b");
974    }
975
976    #[test]
977    fn a_cast_agrees() {
978        agrees("CAST(#0.0::INTEGER)::BIGINT AS a, CAST(#0.0::INTEGER)::VARCHAR AS b");
979    }
980
981    #[test]
982    fn a_comparison_agrees() {
983        agrees("(#0.0::INTEGER > 1::INTEGER)::BOOLEAN AS a");
984    }
985
986    #[test]
987    fn a_conjunction_agrees() {
988        agrees(
989            "((#0.0::INTEGER > 1::INTEGER)::BOOLEAN AND (#0.0::INTEGER < 3::INTEGER)::BOOLEAN)\
990             ::BOOLEAN AS a",
991        );
992    }
993
994    #[test]
995    fn a_function_agrees() {
996        agrees("\"+\"(#0.0::INTEGER, 1::INTEGER)::INTEGER AS a");
997    }
998
999    /// The two evaluators quote the same expression when a divisor is zero. Per #262.
1000    ///
1001    /// This is the one message in the engine that depends on how an expression is written rather
1002    /// than on what it computes, and the two evaluators render it at different times: the prepared
1003    /// form when the pipeline is built, the tree walk on the row that fails. Same renderer, so the
1004    /// same sentence, and this is what says so.
1005    #[test]
1006    fn both_evaluators_quote_the_same_expression_when_a_divisor_is_zero() {
1007        let (schema, chunk) = input();
1008        let (plan, list) = projection("\"//\"(#0.0::INTEGER, 0::INTEGER)::INTEGER AS a");
1009        let prepared = Prepared::new(&plan, &list, &schema).expect("the expression resolves");
1010        let mut scratch = prepared.scratch();
1011        let mut out = Vec::new();
1012        let fast = prepared.evaluate(&chunk, &mut scratch, &mut out).expect_err("divides by zero");
1013        let slow = evaluate(&plan, list[0], &schema, &chunk).expect_err("divides by zero");
1014        assert_eq!(fast.message(), slow.message());
1015        assert!(fast.message().starts_with("Division by zero in expression (x // 0)."), "{fast}");
1016    }
1017
1018    #[test]
1019    fn a_case_agrees() {
1020        agrees(
1021            "CASE WHEN (#0.0::INTEGER > 1::INTEGER)::BOOLEAN THEN 10::INTEGER \
1022             ELSE 20::INTEGER END::INTEGER AS a",
1023        );
1024    }
1025
1026    /// The same expression twice, which is where the tree walk copies the column twice and this
1027    /// does not, and the answers still have to be identical.
1028    #[test]
1029    fn a_column_mentioned_three_times_agrees() {
1030        agrees("\"+\"(\"+\"(#0.0::INTEGER, #0.0::INTEGER)::INTEGER, #0.0::INTEGER)::INTEGER AS a");
1031    }
1032
1033    /// The intermediates of a chain are not all held to the end of it.
1034    ///
1035    /// This is the whole difference between the prepared form being faster than the tree walk on a
1036    /// deep chain and being slower than it, and it is a property of the slot array rather than of
1037    /// any answer, so it is asserted here rather than left to the benchmark to catch.
1038    #[test]
1039    fn a_chain_holds_one_intermediate_at_a_time() {
1040        let (schema, chunk) = input();
1041        let mut expr = "#0.0::INTEGER".to_string();
1042        for _ in 0..8 {
1043            expr = format!("\"+\"({expr}, 1::INTEGER)::INTEGER");
1044        }
1045        let (plan, list) = projection(&format!("{expr} AS a"));
1046        let prepared = Prepared::new(&plan, &list, &schema).expect("the chain resolves");
1047        let mut scratch = prepared.scratch();
1048        prepared.run(&chunk, &mut scratch).expect("the chain runs");
1049        let live = scratch.slots.iter().filter(|slot| slot.is_some()).count();
1050        assert_eq!(live, 1, "a chain that has run should be holding its answer and nothing else");
1051    }
1052
1053    /// The rows a threaded filter keeps are the rows the tree walk says the predicate is true for.
1054    ///
1055    /// Every threaded conjunct is a chance to disagree with the unthreaded answer about a null,
1056    /// about a row an earlier conjunct had already dropped, or about a chunk nothing survives, and
1057    /// the answer is a set of row numbers rather than a vector, so this is checked against the tree
1058    /// walk read a row at a time rather than against the prepared form it is part of.
1059    fn filters(predicate: &str) {
1060        let (schema, chunk) = input();
1061        let (plan, list) = projection(&format!("{predicate} AS p"));
1062        let prepared = Prepared::new(&plan, &list, &schema).expect("the predicate resolves");
1063        let mut scratch = prepared.scratch();
1064        let threaded = prepared.evaluate_filter(&chunk, &mut scratch).expect("the filter runs");
1065        let flags = evaluate(&plan, list[0], &schema, &chunk).expect("the tree walk runs");
1066        let expected = Selection::from_predicate(chunk.len(), |row| is_true(&flags.value_at(row)));
1067        assert_eq!(threaded, expected, "`{predicate}`");
1068        // And running it again over the same scratch is the same answer, because a pipeline calls
1069        // this once a chunk and a slot left behind by the conjunct before would show up here.
1070        let again = prepared.evaluate_filter(&chunk, &mut scratch).expect("the filter runs again");
1071        assert_eq!(again, expected, "`{predicate}` a second time");
1072    }
1073
1074    /// A predicate with no `AND` in it is not threaded and has to keep saying the same thing.
1075    #[test]
1076    fn a_single_comparison_filters_the_same_rows() {
1077        filters("(#0.0::INTEGER > 1::INTEGER)::BOOLEAN");
1078        filters("(#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN");
1079        filters("(#0.0::INTEGER IS NOT DISTINCT FROM NULL::INTEGER)::BOOLEAN");
1080    }
1081
1082    #[test]
1083    fn a_chain_of_conjuncts_keeps_what_all_of_them_keep() {
1084        filters(
1085            "((#0.0::INTEGER > 1::INTEGER)::BOOLEAN AND (#0.0::INTEGER < 3::INTEGER)::BOOLEAN)\
1086             ::BOOLEAN",
1087        );
1088        filters(
1089            "((#0.0::INTEGER >= 1::INTEGER)::BOOLEAN AND (#0.0::INTEGER <= 3::INTEGER)::BOOLEAN \
1090             AND (#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN AND (#0.0::INTEGER <> 2::INTEGER)\
1091             ::BOOLEAN)::BOOLEAN",
1092        );
1093    }
1094
1095    /// A conjunct that rejects every row, in front of one that would have kept some. The rows are
1096    /// the same either way and the point of the shape is that the second conjunct never runs.
1097    #[test]
1098    fn a_conjunct_that_keeps_nothing_ends_the_predicate() {
1099        filters(
1100            "((#0.0::INTEGER > 9::INTEGER)::BOOLEAN AND (#0.0::INTEGER < 9::INTEGER)::BOOLEAN)\
1101             ::BOOLEAN",
1102        );
1103    }
1104
1105    /// A conjunct whose operands are computed rather than read, which is the shape where the
1106    /// comparison is threaded and the arithmetic under it is not.
1107    #[test]
1108    fn a_conjunct_over_a_computed_operand_keeps_the_same_rows() {
1109        filters(
1110            "((#0.0::INTEGER > 1::INTEGER)::BOOLEAN AND \
1111             (\"+\"(#0.0::INTEGER, 1::INTEGER)::INTEGER < 4::INTEGER)::BOOLEAN)::BOOLEAN",
1112        );
1113    }
1114
1115    /// A conjunct that is not a comparison at all, which is the one that goes through the flag
1116    /// kernel rather than the comparison kernel.
1117    #[test]
1118    fn a_conjunct_that_is_not_a_comparison_is_threaded_too() {
1119        filters(
1120            "((#0.0::INTEGER > 1::INTEGER)::BOOLEAN AND ((#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN \
1121             OR (#0.0::INTEGER = 1::INTEGER)::BOOLEAN)::BOOLEAN)::BOOLEAN",
1122        );
1123        filters(
1124            "(((#0.1::VARCHAR = 'c'::VARCHAR)::BOOLEAN OR (#0.0::INTEGER = 3::INTEGER)::BOOLEAN)\
1125             ::BOOLEAN AND (#0.0::INTEGER <> 1::INTEGER)::BOOLEAN)::BOOLEAN",
1126        );
1127    }
1128
1129    /// An `OR` at the top threads the complement: the second branch only sees the rows the first
1130    /// one did not accept, and the rows it accepts are added to them rather than replacing them.
1131    ///
1132    /// The input has a row where the first branch is true, one where the second is, one where both
1133    /// are false and one where the first is null and the second is true, which is the row that says
1134    /// whether the complement was taken over "not true" or over "false".
1135    #[test]
1136    fn an_or_at_the_top_threads_the_complement() {
1137        filters(
1138            "((#0.0::INTEGER > 2::INTEGER)::BOOLEAN OR (#0.1::VARCHAR = 'c'::VARCHAR)::BOOLEAN)\
1139             ::BOOLEAN",
1140        );
1141        filters(
1142            "((#0.0::INTEGER = 1::INTEGER)::BOOLEAN OR (#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN \
1143             OR (#0.0::INTEGER > 2::INTEGER)::BOOLEAN)::BOOLEAN",
1144        );
1145    }
1146
1147    /// A branch that accepts every row, in front of one that would have accepted none. The rows are
1148    /// the same either way and the point of the shape is that the second branch never runs.
1149    #[test]
1150    fn a_branch_that_keeps_everything_ends_the_predicate() {
1151        filters(
1152            "((#0.0::INTEGER IS NOT DISTINCT FROM #0.0::INTEGER)::BOOLEAN OR \
1153             (#0.0::INTEGER > 9::INTEGER)::BOOLEAN)::BOOLEAN",
1154        );
1155    }
1156
1157    /// The branches after one that has accepted every row really are skipped.
1158    ///
1159    /// Every other test here says the threaded answer matches the unthreaded one, which it would
1160    /// even if nothing were threaded at all. This one puts a division by zero behind a branch that
1161    /// accepts everything, so the predicate raises if the second branch runs and does not if the
1162    /// walk stopped where it was supposed to.
1163    #[test]
1164    fn a_branch_behind_one_that_accepted_every_row_does_not_run() {
1165        let (schema, chunk) = input();
1166        let predicate = "((#0.0::INTEGER IS NOT DISTINCT FROM #0.0::INTEGER)::BOOLEAN OR \
1167                         (\"//\"(#0.0::INTEGER, 0::INTEGER)::INTEGER > 0::INTEGER)::BOOLEAN)\
1168                         ::BOOLEAN";
1169        let (plan, list) = projection(&format!("{predicate} AS p"));
1170        let prepared = Prepared::new(&plan, &list, &schema).expect("the predicate resolves");
1171        let mut scratch = prepared.scratch();
1172        let kept =
1173            prepared.evaluate_filter(&chunk, &mut scratch).expect("the second branch never runs");
1174        assert_eq!(kept, Selection::identity(chunk.len()));
1175        // And the same predicate evaluated as an expression does divide by zero, which is what says
1176        // the test is testing the threading rather than a predicate that happens not to raise.
1177        evaluate(&plan, list[0], &schema, &chunk).expect_err("the tree walk divides by zero");
1178    }
1179
1180    /// The conjunct that rejects the most rows ends up in front of the one that rejects none.
1181    ///
1182    /// The predicate is written the wrong way round on purpose. The plan order costs two passes a
1183    /// chunk where one would do, and after a chunk of watching it the filter runs the selective one
1184    /// first and the other one stops running at all.
1185    #[test]
1186    fn a_filter_learns_which_conjunct_to_run_first() {
1187        let (schema, chunk) = input();
1188        let predicate = "((#0.0::INTEGER > 0::INTEGER)::BOOLEAN AND (#0.0::INTEGER > 9::INTEGER)\
1189                         ::BOOLEAN)::BOOLEAN";
1190        let (plan, list) = projection(&format!("{predicate} AS p"));
1191        let prepared = Prepared::new(&plan, &list, &schema).expect("the predicate resolves");
1192        let mut scratch = prepared.scratch();
1193        let root = prepared.roots[0];
1194        assert_eq!(scratch.order(root), None, "nothing has run yet");
1195        let kept = prepared.evaluate_filter(&chunk, &mut scratch).expect("the filter runs");
1196        assert!(kept.is_empty());
1197        assert_eq!(scratch.order(root), Some(&[1, 0][..]), "the second conjunct rejects the most");
1198        // And it stays there, because the conjunct that now runs first empties the selection and
1199        // the one behind it keeps the history it already had rather than losing it.
1200        let kept = prepared.evaluate_filter(&chunk, &mut scratch).expect("the filter runs again");
1201        assert!(kept.is_empty());
1202        assert_eq!(scratch.order(root), Some(&[1, 0][..]));
1203    }
1204
1205    /// Whatever order it settles on, the rows are the rows.
1206    ///
1207    /// Run for longer than the window is wide, because an order that changes halfway through a scan
1208    /// is the shape where a walk that got the subtree bookkeeping wrong would start reading the
1209    /// wrong steps, and the first chunk would not show it.
1210    #[test]
1211    fn reordering_never_changes_which_rows_survive() {
1212        let (schema, chunk) = input();
1213        let predicate = "((#0.0::INTEGER >= 1::INTEGER)::BOOLEAN AND \
1214                         (\"+\"(#0.0::INTEGER, 1::INTEGER)::INTEGER < 4::INTEGER)::BOOLEAN AND \
1215                         (#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN)::BOOLEAN";
1216        let (plan, list) = projection(&format!("{predicate} AS p"));
1217        let prepared = Prepared::new(&plan, &list, &schema).expect("the predicate resolves");
1218        let mut scratch = prepared.scratch();
1219        let flags = evaluate(&plan, list[0], &schema, &chunk).expect("the tree walk runs");
1220        let expected = Selection::from_predicate(chunk.len(), |row| is_true(&flags.value_at(row)));
1221        for round in 0..40 {
1222            let kept = prepared.evaluate_filter(&chunk, &mut scratch).expect("the filter runs");
1223            assert_eq!(kept, expected, "round {round}");
1224        }
1225    }
1226
1227    /// A nested connective is threaded rather than evaluated into flags.
1228    ///
1229    /// The inner `AND` keeps nothing, so its second conjunct is never reached and the division by
1230    /// zero in it never happens. Evaluating the branch as an expression and narrowing the flags
1231    /// afterwards, which is what an operand that is not a connective still does, would have run it.
1232    #[test]
1233    fn a_nested_connective_stops_where_the_outer_one_would() {
1234        let (schema, chunk) = input();
1235        let predicate = "((#0.0::INTEGER > 9::INTEGER)::BOOLEAN OR ((#0.0::INTEGER > 9::INTEGER)\
1236                         ::BOOLEAN AND (\"//\"(#0.0::INTEGER, 0::INTEGER)::INTEGER > 0::INTEGER)\
1237                         ::BOOLEAN)::BOOLEAN)::BOOLEAN";
1238        let (plan, list) = projection(&format!("{predicate} AS p"));
1239        let prepared = Prepared::new(&plan, &list, &schema).expect("the predicate resolves");
1240        let mut scratch = prepared.scratch();
1241        let kept =
1242            prepared.evaluate_filter(&chunk, &mut scratch).expect("the division never happens");
1243        assert!(kept.is_empty());
1244        evaluate(&plan, list[0], &schema, &chunk).expect_err("the tree walk divides by zero");
1245    }
1246
1247    /// A branch that is not a comparison, which is the one that goes through the flag kernel.
1248    #[test]
1249    fn an_or_branch_that_is_not_a_comparison_is_threaded_too() {
1250        filters(
1251            "((#0.0::INTEGER > 2::INTEGER)::BOOLEAN OR \
1252             \"~~\"(#0.1::VARCHAR, 'a%'::VARCHAR)::BOOLEAN)::BOOLEAN",
1253        );
1254        filters(
1255            "(\"~~\"(#0.1::VARCHAR, 'c%'::VARCHAR)::BOOLEAN OR (#0.0::INTEGER = 1::INTEGER)\
1256             ::BOOLEAN)::BOOLEAN",
1257        );
1258    }
1259
1260    /// A connective inside a connective, which recurses rather than falling back to flags.
1261    ///
1262    /// Both nestings, because the two carry opposite things: an `AND` under an `OR` starts from the
1263    /// rows no branch has accepted, and an `OR` under an `AND` starts from the rows every conjunct
1264    /// has kept, and getting either one backwards is a wrong set of rows.
1265    #[test]
1266    fn a_connective_inside_a_connective_threads_both_ways() {
1267        filters(
1268            "(((#0.0::INTEGER >= 2::INTEGER)::BOOLEAN AND (#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN)\
1269             ::BOOLEAN OR ((#0.0::INTEGER < 2::INTEGER)::BOOLEAN AND (#0.1::VARCHAR <> 'c'\
1270             ::VARCHAR)::BOOLEAN)::BOOLEAN)::BOOLEAN",
1271        );
1272        filters(
1273            "(((#0.1::VARCHAR = 'c'::VARCHAR)::BOOLEAN OR (#0.0::INTEGER = 3::INTEGER)::BOOLEAN)\
1274             ::BOOLEAN AND ((#0.0::INTEGER <> 1::INTEGER)::BOOLEAN OR (#0.1::VARCHAR = 'a'\
1275             ::VARCHAR)::BOOLEAN)::BOOLEAN)::BOOLEAN",
1276        );
1277        // Three deep, since two levels is where an off by one in the subtree bookkeeping can still
1278        // be hidden by the ranges lining up.
1279        filters(
1280            "((#0.0::INTEGER > 9::INTEGER)::BOOLEAN OR ((#0.0::INTEGER >= 1::INTEGER)::BOOLEAN \
1281             AND ((#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN OR (#0.0::INTEGER = 1::INTEGER)\
1282             ::BOOLEAN)::BOOLEAN)::BOOLEAN)::BOOLEAN",
1283        );
1284    }
1285
1286    /// A predicate where one side is null and the other is true, in both orders. `OR` is true there
1287    /// and a complement taken over the rows a branch rejected rather than the rows it accepted
1288    /// would drop the row, which is the one way this can be wrong and is not a wrong vector but a
1289    /// missing row.
1290    #[test]
1291    fn a_null_branch_beside_a_true_one_keeps_the_row() {
1292        filters(
1293            "((#0.0::INTEGER > 2::INTEGER)::BOOLEAN OR (#0.1::VARCHAR = 'c'::VARCHAR)::BOOLEAN \
1294             OR (#0.0::INTEGER IS NOT DISTINCT FROM NULL::INTEGER)::BOOLEAN)::BOOLEAN",
1295        );
1296        filters(
1297            "((#0.1::VARCHAR > 'b'::VARCHAR)::BOOLEAN OR (#0.0::INTEGER = 1::INTEGER)::BOOLEAN)\
1298             ::BOOLEAN",
1299        );
1300    }
1301
1302    /// A filter over a chunk that has already been narrowed, which is what a second filter in a
1303    /// pipeline sees and is the form pair the threaded kernels have to handle rather than fall
1304    /// through on.
1305    #[test]
1306    fn a_filter_over_a_selected_chunk_keeps_the_same_rows() {
1307        let (schema, chunk) = input();
1308        let predicate = "((#0.0::INTEGER >= 1::INTEGER)::BOOLEAN AND \
1309                         (#0.1::VARCHAR = 'a'::VARCHAR)::BOOLEAN)::BOOLEAN";
1310        let (plan, list) = projection(&format!("{predicate} AS p"));
1311        let prepared = Prepared::new(&plan, &list, &schema).expect("the predicate resolves");
1312        let mut scratch = prepared.scratch();
1313        let narrowed = narrow(&chunk, &[0, 3]).expect("two of the four rows");
1314        let threaded = prepared.evaluate_filter(&narrowed, &mut scratch).expect("the filter runs");
1315        let flags = evaluate(&plan, list[0], &schema, &narrowed).expect("the tree walk runs");
1316        let expected =
1317            Selection::from_predicate(narrowed.len(), |row| is_true(&flags.value_at(row)));
1318        assert_eq!(threaded, expected);
1319    }
1320
1321    /// Preparing is per pipeline and evaluating is per chunk, so the scratch has to survive being
1322    /// used again and give the same answer the second time.
1323    #[test]
1324    fn a_scratch_used_twice_gives_the_same_answer_twice() {
1325        let (schema, chunk) = input();
1326        let (plan, list) = projection("\"+\"(#0.0::INTEGER, 1::INTEGER)::INTEGER AS a");
1327        let prepared = Prepared::new(&plan, &list, &schema).expect("the expressions resolve");
1328        let mut scratch = prepared.scratch();
1329        let mut once = Vec::new();
1330        prepared.evaluate(&chunk, &mut scratch, &mut once).expect("the first chunk runs");
1331        let mut twice = Vec::new();
1332        prepared.evaluate(&chunk, &mut scratch, &mut twice).expect("the second chunk runs");
1333        assert_eq!(once, twice);
1334    }
1335
1336    /// A chunk shorter than the last one, because a scan's final chunk is that and a constant
1337    /// materialized to the wrong length would be an out of range read rather than a wrong answer.
1338    #[test]
1339    fn a_shorter_chunk_after_a_longer_one_is_evaluated_at_its_own_length() {
1340        let (schema, chunk) = input();
1341        let (plan, list) = projection("7::INTEGER AS a");
1342        let prepared = Prepared::new(&plan, &list, &schema).expect("the expressions resolve");
1343        let mut scratch = prepared.scratch();
1344        let mut full = Vec::new();
1345        prepared.evaluate(&chunk, &mut scratch, &mut full).expect("the full chunk runs");
1346        assert_eq!(full[0].len(), 4);
1347        let short = chunk
1348            .clone()
1349            .select(&{
1350                let mut selection = Selection::with_capacity(2);
1351                selection.push(0);
1352                selection.push(2);
1353                selection
1354            })
1355            .expect("two of the four rows");
1356        let mut cut = Vec::new();
1357        prepared.evaluate(&short, &mut scratch, &mut cut).expect("the short chunk runs");
1358        assert_eq!(cut[0].len(), 2);
1359    }
1360
1361    /// An aggregate is not an expression and saying so when the pipeline is built is better than
1362    /// saying it on the first chunk.
1363    #[test]
1364    fn an_aggregate_is_refused_when_it_is_prepared() {
1365        let (schema, _) = input();
1366        let text = "Aggregate #1 groups=[] aggregates=[sum(#0.0::INTEGER)::HUGEINT]\n  \
1367                    Get memory.main.t AS t #0 [x::INTEGER, s::VARCHAR]";
1368        let plan = Plan::parse(text).expect("a well formed plan");
1369        let Node::Aggregate { aggregates, .. } = *plan.node(plan.root()) else {
1370            panic!("the root of that text is an aggregate");
1371        };
1372        let list = plan.expr_list(aggregates).to_vec();
1373        let error = Prepared::new(&plan, &list, &schema).expect_err("sum is not a scalar");
1374        assert!(error.message().contains("sum"), "{error}");
1375    }
1376}