Skip to main content

spg_engine/eval/
compiled.rs

1//! Compiled expressions — PG's ExprState idea (cut 30, extracted
2//! from `eval.rs`; v7.32 perf knife D / architecture v2 P1).
3//!
4//! Walk the tree ONCE per query, pre-resolve column positions and
5//! collation-fold decisions (both row-invariant), emit a flat
6//! post-order step program; per-row evaluation is a linear loop —
7//! no tree dispatch, no name resolution, no collation lookups.
8//! Anything the compiler doesn't model becomes a `Step::Subtree`
9//! that calls the interpreter for that node, so values AND error
10//! behaviour stay bit-for-bit with `eval_expr` (invariant I3).
11
12use alloc::format;
13use alloc::vec::Vec;
14
15use spg_sql::ast::{BinOp, ColumnName, Expr, Literal, UnOp};
16use spg_storage::{Row, Value};
17
18use super::{
19    EvalContext, EvalError, apply_binary, apply_unary, column_collation, composite_eq, eval_expr,
20    like_match_str, literal_to_value,
21};
22
23pub(crate) enum Step {
24    /// Pre-resolved column read (position into the row).
25    Column(usize),
26    /// Pre-converted literal.
27    Lit(Value<'static>),
28    /// Pops rhs then lhs, pushes the op result. Eager both-sides
29    /// evaluation — same as the interpreter for every op EXCEPT the two
30    /// boolean connectives, which take `Connective` below.
31    Binary(BinOp),
32    /// v7.39 (round 621) — COALESCE and NULLIF as steps on the borrowed
33    /// stack, because they are control flow wearing a function's name.
34    ///
35    /// Through `Step::Function` each had to return `Value<'static>`, which
36    /// forces a clone of a borrowed text argument; and the coalesce arm also
37    /// built a `Vec<DataType>` EVERY row for the numeric widening that
38    /// `COALESCE(1, 2.5)` needs. Measured: `count(coalesce(s,'z'))` at 3.00
39    /// allocations a row, `count(nullif(s,'row1'))` at 2.00, their chain at
40    /// 5.00 — all of it for values that end up borrowed from the row anyway.
41    ///
42    /// On the stack, the chosen argument is handed back AS IS. The widening
43    /// survives by inspection: only when the non-null arguments carry MIXED
44    /// numeric-family types does the step fall to the owned function arm,
45    /// which still does what it always did — same answers, paid only by the
46    /// mixed shapes that need it.
47    Coalesce {
48        n_args: usize,
49    },
50    NullIf,
51    /// v7.39 (round 717) — GREATEST / LEAST. Through `Step::Function`
52    /// every row re-ran `apply_function_lower`'s name dispatch, and
53    /// "least" lives in the crowded five-letter probe chain — measured
54    /// +6 ms over "greatest" on the same 500k scan REGARDLESS of which
55    /// argument wins (the take-always and take-never shapes cost the
56    /// same, so the branch was never the tax; the name was). Uniform
57    /// same-type arguments compare in place off the stack; the mixed /
58    /// coercing / xid / MySQL-NULL shapes fall to the function arm,
59    /// which still does what it always did.
60    Extremum {
61        n_args: usize,
62        max: bool,
63    },
64    /// v7.39 (round 621) — `AND` / `OR`, short-circuiting.
65    ///
66    /// The VM is a stack machine, so both operands were pushed before the
67    /// `Binary` step could look at either: `WHERE x <> 0 AND 1/x > 0` divided
68    /// by zero on exactly the rows the guard exists to exclude. The
69    /// interpreter's arm was fixed first and this path still failed, which is
70    /// the second time a connective has been fixed in one evaluator and not
71    /// the other (round 346's MySQL reading was the first — its comment is
72    /// three screens down).
73    ///
74    /// Rather than turn the hottest loop in the engine into an indexed one
75    /// with jumps, the right operand is its OWN program, run only when the
76    /// left does not decide. Nesting depth is the AND-nesting depth of the
77    /// predicate.
78    Connective {
79        op: BinOp,
80        rhs: Vec<Step>,
81    },
82    /// Comparison whose operands referenced a CaseInsensitive
83    /// column: ASCII-fold Text operands first (decided at compile
84    /// time; the interpreter re-decides per row).
85    /// v7.38.18 — the comparison's collation decides TWO things and the
86    /// step carries both: whether to fold (that is why this variant
87    /// exists) and whether trailing spaces count. `utf8mb4_bin` is
88    /// byte-wise AND `PAD SPACE`; `utf8mb4_0900_ai_ci` folds and does
89    /// NOT pad. One flag cannot say that, so `pads` is resolved once at
90    /// compile time from the collation NAME, next to the fold decision.
91    /// v7.38.19 — the third field is the collator, RESOLVED ONCE at
92    /// compile time.
93    ///
94    /// The run-time arm called `collate::compare(name, a, b)`, and that
95    /// function parses the locale string and builds a whole ICU
96    /// `Collator` on **every call** — so once per row. It is the same
97    /// defect v7.38.18 fixed for sorting, where `Collated` was
98    /// introduced for exactly this reason and the scan-filter path was
99    /// never connected to it.
100    ///
101    /// Measured, 200,000 rows, `WHERE kind < 'd'`, interleaved on one
102    /// box: SPG 62.5 ms against PostgreSQL 18.4's 3.25 ms.
103    BinaryCi(
104        BinOp,
105        crate::collate::TextCompare,
106        Option<alloc::sync::Arc<crate::collate::Collated>>,
107    ),
108    Unary(UnOp),
109    IsNull {
110        negated: bool,
111    },
112    /// v7.39 (round 488) — the verdict of an all-`%` LIKE pattern:
113    /// matches every non-NULL operand, and is NULL for a NULL one.
114    ///
115    /// v7.36 collapsed this shape into `IsNull { negated: !negated }`,
116    /// which answers a three-valued question two-valued. `NULL NOT LIKE
117    /// '%'` came out TRUE where PG18 says NULL, so `WHERE s NOT LIKE '%'`
118    /// SELECTED the NULL row (PG selects nothing), and `SELECT s LIKE '%'`
119    /// printed `false` where PG prints NULL. Same collapse, three-valued.
120    AnyTextMatch {
121        negated: bool,
122    },
123    /// v7.32 (architecture v2, P1) — `needle [NOT] IN (literals…)`.
124    /// The membership SET is a COMPILE PRODUCT, not a runtime cache:
125    /// it lives in the step, so there is no "forgot to pass the
126    /// memo" failure mode (the round-25 18.7 s accident is now
127    /// unconstructable — see v7.32-executor-architecture-design.md
128    /// invariant I2). The needle is the preceding sub-program; this
129    /// step pops it. `fallback` is the whole InList node, used only
130    /// when the runtime needle family doesn't match the set
131    /// (e.g. Float needle vs Int set) — same escape the interpreter
132    /// takes, evaluated cold.
133    InSet {
134        set: crate::memoize::InListSet,
135        has_null: bool,
136        negated: bool,
137        fallback: Expr,
138    },
139    /// v7.32 (P1) — `text [NOT] [I]LIKE '<literal pattern>'`. The
140    /// pattern (and its lowercased form for ILIKE) is compiled once;
141    /// the step pops the text operand.
142    Like {
143        pattern: alloc::vec::Vec<char>,
144        negated: bool,
145        case_insensitive: bool,
146    },
147    /// v7.39 (perf — like_filter tied 1.04×) — unanchored substring
148    /// LIKE: `%[k×_]literal[m×_]%`. Instead of the generic matcher's
149    /// try-every-suffix backtracking (per-position `_`+literal walk),
150    /// scan with `str::find` (two-way, sublinear) over the literal and
151    /// verify the `k` leading / `m` trailing wildcard chars have room.
152    /// v7.39 (round 594) — `text ~ '<literal pattern>'` and its `~*` /
153    /// `regexp_like(...)` spellings. `regexp_like` parsed the pattern into a
154    /// tree for EVERY row: 500k rows cost 350 ms against PG18's 34.5, the
155    /// same 10x whichever way the match was spelled. The pattern is a
156    /// compile product here, exactly as `Step::Like`'s is — PG solves the
157    /// same problem with a cache; a compile product cannot be forgotten.
158    /// v7.39 (round 597) — `<expr> <op> ANY/ALL (<constant array>)`. The
159    /// array is a compile PRODUCT: it used to be rebuilt for every row, and
160    /// `WHERE id = ANY (ARRAY[1..10])` cost 268 ms over 500k rows against
161    /// PG18's 8.3 — 494 at twenty elements — where the equivalent
162    /// `id IN (1..10)` took 2.3. A non-constant right-hand side keeps the
163    /// interpreter, which has to rebuild it: there it really can differ.
164    AnyAll {
165        op: spg_sql::ast::BinOp,
166        is_any: bool,
167        arr: Value<'static>,
168    },
169    /// v7.39 (round 595) — `EXTRACT(<field> FROM <expr>)`. The field is a
170    /// keyword, not a value, so it rides in the step; the source is the
171    /// preceding sub-program and this pops it. `fallback` carries the whole
172    /// node because the extraction's error wording names the source's
173    /// declared type, which only the node knows.
174    Extract {
175        field: spg_sql::ast::ExtractField,
176        fallback: Expr,
177    },
178    Regex {
179        re: crate::eval::CompiledRe,
180        /// The whole call, for an operand that is not text: the interpreter
181        /// owns whatever coercion or error that is, and this step must not
182        /// invent one. Same escape `Step::InSet` takes, evaluated cold.
183        fallback: Expr,
184    },
185    LikeSubstring {
186        needle: alloc::string::String,
187        k_before: usize,
188        m_after: usize,
189        negated: bool,
190        case_insensitive: bool,
191    },
192    /// v7.36 (perf — mailrs Ask 1) — pure scalar function call
193    /// (LENGTH, COALESCE, UPPER, etc.) on already-pushed args.
194    /// Pops `n_args` values, calls `apply_function(name, args, ctx)`,
195    /// pushes the result. Replaces the Subtree fallback for the
196    /// "function over bound columns" shape that aggregate arg paths
197    /// like `SUM(LENGTH(text_body))` and `MAX(COALESCE(col, ''))`
198    /// otherwise force the row-materialise eval path. Only the
199    /// `fully_compilable` whitelist (PURE scalars — no NOW / RANDOM
200    /// / sequence accessors) is emitted; everything else stays on
201    /// `Step::Subtree`.
202    /// `name_lower` is pre-lowercased at compile time so the per-
203    /// row dispatch in `apply_function` skips an allocation on
204    /// every input row.
205    Function {
206        name_lower: alloc::string::String,
207        n_args: usize,
208    },
209    /// v7.36 (perf — mailrs Ask 1 SUM(LENGTH(text_body)) zero-copy)
210    /// — `LENGTH(<column>)` / `CHAR_LENGTH(<column>)` /
211    /// `CHARACTER_LENGTH(<column>)` over a bound column. Reads the
212    /// cell by reference, computes the char length WITHOUT cloning
213    /// the underlying `String` — the 1 KB text bodies in
214    /// `user_storage_usage` otherwise pay 25 k × 1 KB heap allocs
215    /// per query just to push a `Value::Text` onto the stack so the
216    /// next Step pops it and asks `s.len()`.
217    ColumnLength {
218        pos: usize,
219    },
220    /// v7.36 — `OCTET_LENGTH(<column>)` — byte count, regardless of
221    /// encoding. Even simpler than `ColumnLength` (no ASCII probe).
222    ColumnOctetLength {
223        pos: usize,
224    },
225    /// v7.36 — `CAST(<expr> AS <ty>)` over an already-pushed value.
226    /// Pure / context-free conversion goes through the same
227    /// `cast_value` dispatcher the interpreter uses.
228    Cast {
229        target: spg_sql::ast::CastTarget,
230    },
231    /// v7.39 (round 722) — a NAMED cast whose name resolved at COMPILE
232    /// time (`::NUMERIC`, `::REAL`, `numeric(10,2)` — the
233    /// `plain_named_target` table). The blanket Named -> Subtree rule
234    /// sent these to the interpreter — worse, it made the whole
235    /// aggregate argument non-compilable, so `count(id::NUMERIC)` fell
236    /// off the round-716 fused parallel lane entirely. The name rides
237    /// along for error wording only.
238    CastPlain {
239        dt: spg_storage::DataType,
240        name: alloc::string::String,
241    },
242    /// v7.37.5-A2b — `CASE [operand] WHEN x THEN y … ELSE z END`.
243    /// Each `(when, then)` branch and the optional `else` is a
244    /// pre-compiled sub-program; the executor short-circuits on the
245    /// first matching WHEN. Compiles only when **every** sub-program
246    /// is itself `fully_compilable` (so the Case never falls back to
247    /// a Subtree that would force a row materialise — profile-guided
248    /// fix for Track A `COUNT(DISTINCT CASE WHEN ...)` aggregates).
249    /// Searched form has `operand=None` and treats each WHEN as a
250    /// Bool predicate; simple form has `operand=Some(prog)` and
251    /// compares the operand value with each WHEN via `BinOp::Eq`.
252    Case {
253        /// v7.38.14 — may the simple-form operand match fold text?
254        ///
255        /// Decided at COMPILE time, where the operand's and the WHEN
256        /// arms' expressions are still in hand, and never in the row
257        /// loop. The runtime arm below used `apply_binary_by_ref(Eq)`,
258        /// which is dialect-blind, so `CASE s WHEN 'A'` compared bytes
259        /// while the interpreter's `Expr::Case` folded — one semantic
260        /// with two implementations, disagreeing.
261        fold_operand: bool,
262        operand: Option<CompiledExpr>,
263        branches: alloc::vec::Vec<(CompiledExpr, CompiledExpr)>,
264        else_branch: Option<CompiledExpr>,
265    },
266    /// v7.38 (read01) — widen the top-of-stack value to a statically
267    /// resolved PG common type (e.g. a `CASE` whose branches mix integer and
268    /// numeric resolves to numeric). Resolved once at compile time from the
269    /// branch expressions' types, so the per-row cost is a single
270    /// scale-preserving coercion, not a describe. See
271    /// [`crate::eval::widen_value_to`].
272    CoerceCommon(spg_storage::DataType),
273    /// Fallback: interpret this subtree with eval_expr.
274    Subtree(Expr),
275}
276
277pub(crate) struct CompiledExpr {
278    steps: Vec<Step>,
279    /// Which fast predicate shape this program is — settled once, here,
280    /// instead of re-derived per row. See [`PredShape`].
281    pred_shape: PredShape,
282}
283
284/// v7.39 (round 486) — the shape of a compiled predicate, decided at
285/// compile time.
286///
287/// Round 482 added a `<column> <cmp> <literal>` fast path and this round
288/// added `<column> [NOT] IN (<literals>)`. Both were slice pattern-matches
289/// run PER ROW, so a program that is neither paid for every probe in the
290/// list: adding the second one cost `like_filter` — a shape with no `IN`
291/// anywhere in it — 4.5 %, measured against the previous commit on the same
292/// machine minutes apart. A program's shape does not change between its
293/// rows, so it is settled once and the row loop reads one discriminant.
294#[derive(Clone, Copy, PartialEq, Eq, Debug)]
295pub(crate) enum PredShape {
296    Other,
297    /// r1021 — an integer-only arithmetic predicate, run without building a
298    /// single `Value`. See [`CompiledExpr::is_int_arith_pred`].
299    IntArith,
300    ColumnCmpLit,
301    ColumnInSet,
302    ColumnLike,
303}
304
305impl CompiledExpr {
306    /// v7.36 (perf — mailrs Phase 1, user_storage_usage hot loop) —
307    /// shape inspector for the aggregate's tight inner. Returns
308    /// `Some(pos)` iff this compiled expression is exactly the
309    /// single step `ColumnLength { pos }` — i.e. `LENGTH(<column>)`
310    /// on a bound text column with no surrounding work.
311    /// r1021 — the deepest an integer lane will go. A predicate needing
312    /// more stack than this falls back; measured shapes use two or three.
313    const INT_LANE_DEPTH: usize = 8;
314
315    /// r1021 — is this predicate built only from integer columns, integer
316    /// literals and integer arithmetic, ending in one comparison?
317    ///
318    /// Round 482 traced the per-row predicate cost to `Value` churn and
319    /// answered it with ONE hard-coded shape, `<column> <cmp> <literal>`.
320    /// Anything with arithmetic in it — `id % 3 = 0`, the bucketing and
321    /// parity predicates real schemas are full of — still builds and
322    /// destroys a `Value` per step. Profiled (2026-08-14, see
323    /// `docs/PERF_FILTERED_THEN_ORDER_2026-08-14.md`):
324    /// `drop_glue<Value>` is the LARGEST leaf on `WHERE id % 3 = 0`, ahead
325    /// of the modulo it carries, and 22x heavier per rep than on the shape
326    /// that skips the step machine.
327    ///
328    /// So this recognises a CLASS rather than a shape. Structural only —
329    /// no column types are consulted here — because every value the lane
330    /// cannot handle makes it fall back at run time instead of guessing.
331    fn is_int_arith_pred(&self) -> bool {
332        let mut depth = 0usize;
333        let mut comparisons = 0usize;
334        for (i, step) in self.steps.iter().enumerate() {
335            match step {
336                Step::Column(_) => depth += 1,
337                Step::Lit(v) => {
338                    if !matches!(v, Value::Int(_) | Value::BigInt(_)) {
339                        return false;
340                    }
341                    depth += 1;
342                }
343                Step::Binary(op) => {
344                    if depth < 2 {
345                        return false;
346                    }
347                    depth -= 1;
348                    if is_int_comparison(*op) {
349                        comparisons += 1;
350                        // The comparison is the answer, so it ends the
351                        // program; a later step would consume a bool the
352                        // lane does not carry.
353                        if i + 1 != self.steps.len() {
354                            return false;
355                        }
356                    } else if !is_int_arithmetic(*op) {
357                        return false;
358                    }
359                }
360                _ => return false,
361            }
362            if depth > Self::INT_LANE_DEPTH {
363                return false;
364            }
365        }
366        depth == 1 && comparisons == 1
367    }
368
369    /// r1021 — run an [`Self::is_int_arith_pred`] program over `i64`s.
370    ///
371    /// `None` means "this row is not for the lane" and the caller runs the
372    /// ordinary machine. Every case that could answer differently from the
373    /// interpreter takes that exit rather than deciding for itself: a NULL
374    /// or non-integer cell, a division by zero, an overflow, and a result
375    /// that would not fit the width the operands imply. The lane therefore
376    /// cannot change a single answer — it can only reach the same one
377    /// without a heap type in the middle.
378    ///
379    /// Width follows PG: `int4 op int4` stays `int4` and overflowing it is
380    /// an error, so a 32-bit result that leaves 32-bit range hands the row
381    /// back and the interpreter raises exactly as before. Mixed widths
382    /// widen to 64-bit, and `smallint` is simply not admitted.
383    fn eval_int_arith_pred(&self, row: &Row<'static>) -> Option<bool> {
384        let mut vals = [0i64; Self::INT_LANE_DEPTH];
385        let mut narrow = [false; Self::INT_LANE_DEPTH];
386        let mut n = 0usize;
387        for step in &self.steps {
388            match step {
389                Step::Column(pos) => {
390                    let (v, is32) = int_operand(row.values.get(*pos)?)?;
391                    vals[n] = v;
392                    narrow[n] = is32;
393                    n += 1;
394                }
395                Step::Lit(lit) => {
396                    let (v, is32) = int_operand(lit)?;
397                    vals[n] = v;
398                    narrow[n] = is32;
399                    n += 1;
400                }
401                Step::Binary(op) => {
402                    let (rhs, rhs32) = (vals[n - 1], narrow[n - 1]);
403                    let (lhs, lhs32) = (vals[n - 2], narrow[n - 2]);
404                    n -= 2;
405                    if is_int_comparison(*op) {
406                        return Some(match op {
407                            BinOp::Eq => lhs == rhs,
408                            BinOp::NotEq => lhs != rhs,
409                            BinOp::Lt => lhs < rhs,
410                            BinOp::LtEq => lhs <= rhs,
411                            BinOp::Gt => lhs > rhs,
412                            _ => lhs >= rhs,
413                        });
414                    }
415                    let out = match op {
416                        BinOp::Add => lhs.checked_add(rhs)?,
417                        BinOp::Sub => lhs.checked_sub(rhs)?,
418                        BinOp::Mul => lhs.checked_mul(rhs)?,
419                        BinOp::Div => lhs.checked_div(rhs)?,
420                        _ => lhs.checked_rem(rhs)?,
421                    };
422                    let out32 = lhs32 && rhs32;
423                    if out32 && i32::try_from(out).is_err() {
424                        return None;
425                    }
426                    vals[n] = out;
427                    narrow[n] = out32;
428                    n += 1;
429                }
430                _ => return None,
431            }
432        }
433        None
434    }
435
436    /// v7.39 (round 482) — is this exactly `<column> <cmp> <literal>`?
437    ///
438    /// Rounds 478-481 traced the per-row predicate cost to `Value` churn:
439    /// three steps a row (Column, Lit, Binary) means three `Value`s built
440    /// and destroyed, and `drop_glue<Value>` is an out-of-line call that
441    /// switches on the discriminant even when the value carries no heap.
442    /// Round 481's counter ruled out leftovers on the stack — the churn is
443    /// the VM's ordinary operands.
444    ///
445    /// This shape needs none of them: both operands can be read by
446    /// reference. It covers `g = 5` and `s = '…'`; `LIKE` is its own AST
447    /// node rather than a `BinOp`, so it compiles to a different step and
448    /// is NOT covered here — measured, not assumed.
449    ///
450    /// `BinaryCi` is deliberately not matched: it folds its operands
451    /// first, which is a different comparison. Nor is the mirrored
452    /// `<literal> <cmp> <column>` — flipping the operator is a separate
453    /// judgement and this returns None so it takes the general path.
454    pub(crate) fn as_column_cmp_literal(&self) -> Option<(usize, BinOp, &Value<'static>)> {
455        let [Step::Column(pos), Step::Lit(lit), Step::Binary(op)] = &self.steps[..] else {
456            return None;
457        };
458        if !matches!(
459            op,
460            BinOp::Eq | BinOp::NotEq | BinOp::Lt | BinOp::LtEq | BinOp::Gt | BinOp::GtEq
461        ) {
462            return None;
463        }
464        Some((*pos, *op, lit))
465    }
466
467    /// v7.39 (round 486) — the sibling shape `<column> [NOT] IN (<literals>)`.
468    ///
469    /// `big_in` is the read panel's worst shape and compiles to exactly two
470    /// steps, `Column` then `InSet`. The round-482 fast path does not cover
471    /// it (three steps, a `Binary`), so it runs the general VM: a `Value`
472    /// built from the cell, popped, and a `Value::Bool` built and popped
473    /// again. Its profile put `drop_glue<Value>` at 20 % and the VM loop at
474    /// 27 %. The set lookup itself wants nothing but a reference to the
475    /// cell.
476    pub(crate) fn as_column_in_set(
477        &self,
478    ) -> Option<(usize, &crate::memoize::InListSet, bool, bool)> {
479        let [
480            Step::Column(pos),
481            Step::InSet {
482                set,
483                has_null,
484                negated,
485                ..
486            },
487        ] = &self.steps[..]
488        else {
489            return None;
490        };
491        Some((*pos, set, *has_null, *negated))
492    }
493
494    /// v7.39 (round 488) — the third two-step shape: `<column> [NOT]
495    /// [I]LIKE '<literal>'`, in either the general matcher's form or the
496    /// unanchored-substring form round 484 added.
497    ///
498    /// `like_filter` is the read panel's worst shape. Rounds 482 and 486
499    /// covered its two siblings; this one still ran the general VM, which
500    /// pushes the cell as a `Value` and pops it again for a matcher that
501    /// only ever wanted a `&str`.
502    pub(crate) fn as_column_like(&self) -> Option<(usize, &Step)> {
503        let [
504            Step::Column(pos),
505            step @ (Step::Like { .. } | Step::LikeSubstring { .. }),
506        ] = &self.steps[..]
507        else {
508            return None;
509        };
510        Some((*pos, step))
511    }
512
513    pub(crate) fn as_single_column_length(&self) -> Option<usize> {
514        if self.steps.len() == 1
515            && let Step::ColumnLength { pos } = &self.steps[0]
516        {
517            Some(*pos)
518        } else {
519            None
520        }
521    }
522}
523
524/// Column-position resolution at compile time. Mirrors the happy
525/// layers of `resolve_column`; ANY case that would reach an error
526/// path, an ambiguity, or a miss returns None so the node falls
527/// back to the interpreter (identical runtime error / NULL
528/// semantics).
529///
530/// v7.37.16 — pub(crate): the aggregate bind-once fast path
531/// (aggregate.rs `col_pos`) uses this as its resolver so bare-name
532/// group/arg columns bind exactly like compiled-WHERE columns do.
533/// v7.39 (round 693) — does this comparison operand carry a collation the
534/// VM cannot perform?
535///
536/// Deliberately a COMPILE-time question. The answer is the same for every
537/// row of the scan, and the alternative — asking per row inside `compare` —
538/// puts a lookup on the hottest path in the engine.
539/// v7.39 (round 704) — does this comparison pair an unknown string literal
540/// with a numeric-family operand whose type the literal will not parse as?
541/// Compile-time twin of the eval Binary arm's error rewrite; see the bail
542/// site for why the shape cannot stay on the VM.
543fn unparseable_numeric_literal_cmp(lhs: &Expr, rhs: &Expr, ctx: &EvalContext<'_>) -> bool {
544    let check = |lit: &Expr, other: &Expr| -> bool {
545        let Expr::Literal(spg_sql::ast::Literal::String(text)) = lit else {
546            return false;
547        };
548        let Some(desc) = crate::describe::describe_expr(other, ctx.columns) else {
549            return false;
550        };
551        if !matches!(
552            desc.ty,
553            spg_storage::DataType::SmallInt
554                | spg_storage::DataType::Int
555                | spg_storage::DataType::BigInt
556                | spg_storage::DataType::Float
557                | spg_storage::DataType::Real
558                | spg_storage::DataType::Numeric { .. }
559        ) {
560            return false;
561        }
562        crate::conversions::coerce_value(spg_storage::Value::text(text.as_str()), desc.ty, "", 0)
563            .is_err()
564    };
565    check(lhs, rhs) || check(rhs, lhs)
566}
567
568fn operand_declares_a_collation(e: &Expr, ctx: &EvalContext<'_>) -> bool {
569    let derived = crate::collate_derive::derive(e, &|c: &ColumnName| {
570        let pos = crate::eval::find_column_pos(c, ctx)?;
571        ctx.columns.get(pos)?.collation_name.clone()
572    });
573    // A conflict has to leave the VM too — the tree evaluator is where the
574    // error is raised, with PG's own sentence.
575    derived.conflict().is_some()
576        || derived
577            .name()
578            .is_some_and(|n| crate::collate::is_supported(n))
579}
580
581pub(crate) fn compile_column_pos(c: &ColumnName, ctx: &EvalContext<'_>) -> Option<usize> {
582    if let Some(q) = &c.qualifier {
583        if let Some(pos) = ctx
584            .columns
585            .iter()
586            .position(|s| composite_eq(&s.name, q, &c.name))
587        {
588            return Some(pos);
589        }
590        // resolve_column's error layers live behind this point:
591        // composites under the qualifier exist (ColumnNotFound) or
592        // the qualifier is unknown (UnknownQualifier) — interpret.
593        let prefix_exists = ctx.columns.iter().any(|s| {
594            s.name.starts_with(q.as_str()) && s.name.as_bytes().get(q.len()) == Some(&b'.')
595        });
596        if prefix_exists {
597            return None;
598        }
599        match ctx.table_alias {
600            // Alias-accepted single-table reference: fall through
601            // to the bare layers (the inner-subquery hot shape).
602            Some(a) if a == q => {}
603            _ => return None,
604        }
605    }
606    if let Some(pos) = ctx.columns.iter().position(|s| s.name == c.name) {
607        return Some(pos);
608    }
609    let mut matches = ctx.columns.iter().enumerate().filter(|(_, s)| {
610        s.name.len() > c.name.len()
611            && s.name.ends_with(c.name.as_str())
612            && s.name.as_bytes()[s.name.len() - c.name.len() - 1] == b'.'
613    });
614    let first = matches.next();
615    if matches.next().is_some() {
616        return None; // ambiguous — interpreter owns the error text
617    }
618    first.map(|(i, _)| i)
619}
620
621/// v7.39 (round 621) — can evaluating this raise at RUN time?
622///
623/// The errors a short circuit spares are the run-time ones: a division, an
624/// overflow, a cast that will not parse, a function that refuses its input.
625/// A type mismatch is not among them — PG raises those while ANALYSING, so it
626/// raises them whether or not the operand would have been evaluated, and so
627/// does SPG. That is why a predicate built only from columns, literals,
628/// comparisons and the boolean shapes over them needs no short circuit: there
629/// is nothing for it to spare.
630///
631/// Unrecognised shapes answer `true`, so a new kind of expression short
632/// circuits (correct, slightly slower) rather than silently not.
633fn can_raise_at_run_time(e: &Expr) -> bool {
634    match e {
635        Expr::Literal(_) | Expr::Column(_) => false,
636        Expr::Binary { op, lhs, rhs } => {
637            !matches!(
638                op,
639                BinOp::Eq
640                    | BinOp::NotEq
641                    | BinOp::Lt
642                    | BinOp::LtEq
643                    | BinOp::Gt
644                    | BinOp::GtEq
645                    | BinOp::And
646                    | BinOp::Or
647            ) || can_raise_at_run_time(lhs)
648                || can_raise_at_run_time(rhs)
649        }
650        Expr::Unary { op, expr } => !matches!(op, UnOp::Not) || can_raise_at_run_time(expr),
651        Expr::IsNull { expr, .. } | Expr::BoolTest { expr, .. } => can_raise_at_run_time(expr),
652        Expr::Like { expr, pattern, .. } => {
653            can_raise_at_run_time(expr) || can_raise_at_run_time(pattern)
654        }
655        Expr::InList { expr, list, .. } => {
656            can_raise_at_run_time(expr) || list.iter().any(can_raise_at_run_time)
657        }
658        _ => true,
659    }
660}
661
662fn compile_into(e: &Expr, ctx: &EvalContext<'_>, steps: &mut Vec<Step>) {
663    match e {
664        Expr::Literal(l) => steps.push(Step::Lit(literal_to_value(l))),
665        Expr::Column(c) => match compile_column_pos(c, ctx) {
666            // v7.39 (read01 round 56) — a COMPOSITE column must not compile to
667            // a raw `Step::Column`: that loads the stored JSON straight off the
668            // row and skips the rehydration into `Value::Composite` that
669            // `resolve_column` does. `p = ROW(2,'b')::pt` in a WHERE then
670            // compared Json against Composite and errored, while the same
671            // predicate in a projection worked. Route it through eval instead.
672            // The check is COMPILE-time, so the hot column path pays nothing.
673            Some(pos)
674                if ctx
675                    .columns
676                    .get(pos)
677                    .is_some_and(|sc| sc.user_composite_type.is_some()) =>
678            {
679                steps.push(Step::Subtree(e.clone()));
680            }
681            Some(pos) => steps.push(Step::Column(pos)),
682            None => steps.push(Step::Subtree(e.clone())),
683        },
684        Expr::Binary { lhs, op, rhs } => {
685            // v7.39 (round 383) — the MySQL bitwise operators are UNSIGNED
686            // 64-bit (`~ & | ^ << >>`); the VM's Step::Binary calls the
687            // dialect-blind apply_binary, so route them to the interpreter,
688            // which has the dialect (eval.rs `mysql_bitwise`). `<< >>` share
689            // the inet-containment BinOps — the interpreter still keeps the
690            // inet meaning for non-numeric operands.
691            if ctx.mysql_dialect
692                && matches!(
693                    op,
694                    BinOp::BitAnd
695                        | BinOp::BitOr
696                        | BinOp::BitXor
697                        | BinOp::InetContainedBy
698                        | BinOp::InetContains
699                )
700            {
701                steps.push(Step::Subtree(e.clone()));
702                return;
703            }
704            // v7.39 (round 407) — MySQL's logical `XOR` reads both sides as
705            // truth values, which the VM's dialect-blind apply_binary (no
706            // LogicalXor arm) cannot do. Route to the interpreter, whose
707            // eval_expr arm handles the connective (eval.rs
708            // `eval_mysql_connective`).
709            if ctx.mysql_dialect && matches!(op, BinOp::LogicalXor) {
710                steps.push(Step::Subtree(e.clone()));
711                return;
712            }
713            // v7.39 (round 402) — an arithmetic op on a SET / inline-ENUM
714            // column reads the column numerically (bitmask / 1-based
715            // ordinal), which the VM's value-level Add cannot see (it has the
716            // text). Route to the interpreter, which folds it (eval.rs
717            // resolve `collation_fold_for_compare`).
718            if ctx.mysql_dialect
719                && matches!(
720                    op,
721                    BinOp::Add | BinOp::Sub | BinOp::Mul | BinOp::Div | BinOp::Mod
722                )
723                && (crate::eval::expr_set_variants(lhs, ctx.columns).is_some()
724                    || crate::eval::expr_set_variants(rhs, ctx.columns).is_some()
725                    || crate::eval::expr_inline_enum_variants(lhs, ctx.columns).is_some()
726                    || crate::eval::expr_inline_enum_variants(rhs, ctx.columns).is_some())
727            {
728                steps.push(Step::Subtree(e.clone()));
729                return;
730            }
731            // v7.39 (round 621) — the boolean connectives short-circuit, so
732            // the right operand compiles to its own program. The shapes whose
733            // right operand is a literal go to the interpreter instead: those
734            // carry PG's analysis-time half (a non-boolean literal is refused
735            // even when the short circuit would not reach it, and an unknown
736            // string literal is resolved), which is decided there and is not
737            // worth a second implementation for how rare they are in a
738            // compiled predicate.
739            if matches!(op, BinOp::And | BinOp::Or) {
740                if matches!(rhs.as_ref(), Expr::Literal(_)) {
741                    steps.push(Step::Subtree(e.clone()));
742                    return;
743                }
744                // A right operand that cannot fail has nothing to be spared,
745                // so it keeps the eager step and its inline cost. `WHERE g
746                // BETWEEN 10 AND 20` is `g >= 10 AND g <= 20`, the commonest
747                // conjunctive predicate there is, and paying a nested program
748                // per row for it measured +42% to +60% on the panel — a real
749                // regression, reproduced, for a short circuit that can never
750                // change an answer.
751                if !can_raise_at_run_time(rhs) {
752                    compile_into(lhs, ctx, steps);
753                    compile_into(rhs, ctx, steps);
754                    steps.push(Step::Binary(*op));
755                    return;
756                }
757                compile_into(lhs, ctx, steps);
758                let mut rhs_steps = Vec::new();
759                compile_into(rhs, ctx, &mut rhs_steps);
760                steps.push(Step::Connective {
761                    op: *op,
762                    rhs: rhs_steps,
763                });
764                return;
765            }
766            let cmp = matches!(
767                op,
768                BinOp::Eq | BinOp::NotEq | BinOp::Lt | BinOp::LtEq | BinOp::Gt | BinOp::GtEq
769            );
770            // v7.39 (enum order knife) — an enum-witnessed comparison must
771            // order by catalog member order; the VM's value-level compare
772            // cannot. Fall back to the tree evaluator for this subtree
773            // (compile-time check, zero cost when the catalog has no
774            // enum types).
775            if cmp
776                && ctx.catalog.is_some_and(|cat| !cat.enum_types().is_empty())
777                && (crate::eval::expr_enum_labels(lhs, ctx.columns, ctx.catalog).is_some()
778                    || crate::eval::expr_enum_labels(rhs, ctx.columns, ctx.catalog).is_some())
779            {
780                steps.push(Step::Subtree(e.clone()));
781                return;
782            }
783            // v7.39 (round 693) — and the same move for a declared
784            // collation, which is the shape F36 had left: `loc BETWEEN 'a'
785            // AND 'd'` returns a different ROW SET under en_US.utf8 than
786            // under byte order.
787            //
788            // Compile-time, like its enum neighbour, and for the better of
789            // the two reasons. `binop::compare` is the dominant cost of a
790            // scan — its own comment measures 35.6 % of self time on
791            // `g = 5` — so a per-row collation lookup there would have to
792            // earn its place against a bench. Deciding once, while the
793            // predicate compiles, costs the scan nothing at all: a column
794            // that declares nothing never leaves the VM.
795            //
796            // Only the ORDERING operators. Measured on PG18, `=`, `<>`,
797            // LIKE, IN and count(DISTINCT …) all give byte-equality's
798            // answer under a deterministic collation.
799            if matches!(op, BinOp::Lt | BinOp::LtEq | BinOp::Gt | BinOp::GtEq)
800                && operand_declares_a_collation(lhs, ctx) | operand_declares_a_collation(rhs, ctx)
801            {
802                steps.push(Step::Subtree(e.clone()));
803                return;
804            }
805            // v7.39 (round 704) — an UNKNOWN string literal against a
806            // numeric-family operand that will NOT parse as its type. PG's
807            // error for `WHERE i = 'abc'` is the input function's
808            // (`invalid input syntax for type integer: "abc"`); the VM's
809            // value-level compare can only say "operator does not exist",
810            // so this shape leaves for the tree evaluator, whose Binary
811            // arm has the Exprs and rewrites the error. Compile-time and
812            // failure-only: a literal that parses stays on the VM path
813            // and costs nothing.
814            if cmp && unparseable_numeric_literal_cmp(lhs, rhs, ctx) {
815                steps.push(Step::Subtree(e.clone()));
816                return;
817            }
818            compile_into(lhs, ctx, steps);
819            compile_into(rhs, ctx, steps);
820            let ci = cmp
821                && (matches!(
822                    column_collation(lhs, ctx),
823                    Some(spg_storage::Collation::CaseInsensitive)
824                ) || matches!(
825                    column_collation(rhs, ctx),
826                    Some(spg_storage::Collation::CaseInsensitive)
827                ));
828            // v7.39 (round 364, M4 P2) — a MySQL session folds every text
829            // comparison, so it needs the CI step too (the step chooses
830            // the accent-aware fold at run time).
831            let ci = ci || (cmp && super::resolve::mysql_text_fold_applies(lhs, rhs, ctx));
832            // v7.38.18 — the same resolver the interpreted path uses.
833            // A byte-wise collation that PADS still needs the step: it
834            // does not fold case, but trailing spaces do not count.
835            let tc = super::resolve::text_compare_of(lhs, rhs, ctx);
836            // v7.38.19 — `BinaryCi` exists to FOLD: case, and padding.
837            // An ordering collation needs neither, and asking
838            // `is_plain_bytes()` -- which also answers false when an
839            // ORDER is declared -- emitted it for every text comparison
840            // in a database with a locale collation. Its `fold_one` then
841            // copied each side into a fresh String per row and left it
842            // exactly as it found it.
843            //
844            // Two allocations and two copies per row, to change nothing.
845            // Interleaved A/B on one box, 200,000 rows,
846            // `WHERE kind = 'click'`:
847            //
848            //   database collation C              2.177-3.086 ms
849            //   database collation en_US.UTF-8   58.292-76.242 ms
850            //   PostgreSQL 18.4, en_US.utf8       1.487-4.391 ms
851            //
852            // Twenty-six times SPG's own byte path, and thirty times the
853            // engine we stand in for, on the most ordinary predicate
854            // there is. It shipped in v7.38.18 with the database
855            // collation, and the perf gate could not see it: all
856            // sixty-four cells of the sweep run under `C`.
857            //
858            // ORDERING still needs the step: `BinaryCi` does double duty,
859            // folding AND routing to the collator, so `<` under a locale
860            // collation must keep it or it falls back to byte order. The
861            // first version of this change did exactly that, and the
862            // pins written for the database collation caught it in the
863            // same minute: `WHERE x < 'b'` returned `client` where it
864            // owes `Bob Charlie client Zebra`.
865            //
866            // EQUALITY is the case that can skip it, and it can skip it
867            // for a reason rather than for speed: PostgreSQL's locale
868            // collations are DETERMINISTIC, which means two strings that
869            // compare equal are byte-identical. The collator cannot
870            // change the answer, so asking it is pure cost. A
871            // non-deterministic collation is SPG's MySQL
872            // case-insensitive one, and that sets `fold_case`, so it
873            // takes the branch above.
874            let orders = tc.order.is_some();
875            let ordering_op = matches!(op, BinOp::Lt | BinOp::LtEq | BinOp::Gt | BinOp::GtEq);
876            let needs_fold = tc.fold_case || tc.pads || (orders && ordering_op);
877            let collated = tc
878                .order
879                .as_deref()
880                .and_then(crate::collate::Collated::resolve)
881                .map(alloc::sync::Arc::new);
882            steps.push(if cmp && needs_fold {
883                Step::BinaryCi(*op, tc, collated)
884            } else if ci {
885                Step::BinaryCi(*op, tc, collated)
886            } else {
887                Step::Binary(*op)
888            });
889        }
890        Expr::Unary { op, expr } => {
891            // v7.39 (round 383) — MySQL `~x` is the UNSIGNED 64-bit
892            // complement; route to the interpreter (eval.rs `mysql_bit_not`)
893            // since Step::Unary calls the dialect-blind apply_unary.
894            if ctx.mysql_dialect && matches!(op, UnOp::BitNot) {
895                steps.push(Step::Subtree(e.clone()));
896                return;
897            }
898            compile_into(expr, ctx, steps);
899            steps.push(Step::Unary(*op));
900        }
901        Expr::IsNull { expr, negated } => {
902            compile_into(expr, ctx, steps);
903            steps.push(Step::IsNull { negated: *negated });
904        }
905        Expr::InList {
906            expr,
907            list,
908            negated,
909        } => {
910            // v7.39 (round 364, M4 P2) — a MySQL session folds text before
911            // the membership test; the set-based compiled path compares
912            // raw. Route it to the interpreter, which folds (eval.rs
913            // `eval_in_list_arm`). The perf-critical InSet path is PG-only.
914            if ctx.mysql_dialect {
915                steps.push(Step::Subtree(e.clone()));
916                return;
917            }
918            // I2: the set is built at compile time. The gate
919            // (`fully_compilable`) guarantees we only reach here
920            // when the list builds a set and the needle compiles —
921            // but keep the Subtree fallback for defence in depth.
922            match crate::build_in_list_set(list) {
923                Some(entry) if fully_compilable(expr) => {
924                    compile_into(expr, ctx, steps);
925                    steps.push(Step::InSet {
926                        set: entry.set,
927                        has_null: entry.has_null,
928                        negated: *negated,
929                        fallback: e.clone(),
930                    });
931                }
932                _ => steps.push(Step::Subtree(e.clone())),
933            }
934        }
935        Expr::Like {
936            expr,
937            pattern,
938            negated,
939            case_insensitive,
940        } => {
941            // v7.39 (round 364, M4 P2) — LIKE folds accents + case on a
942            // MySQL session (eval.rs `eval_like_arm`); the compiled
943            // pattern walk does not. Route to the interpreter.
944            if ctx.mysql_dialect {
945                steps.push(Step::Subtree(e.clone()));
946                return;
947            }
948            match literal_text_pattern(pattern) {
949                Some(pat) if fully_compilable(expr) => {
950                    // v7.36 (perf — mailrs Phase 1, get_contacts hot
951                    // inner) — trivial all-`%` pattern (`%`, `%%`, …)
952                    // matches every non-NULL text. Collapse the LIKE
953                    // into a `lhs IS NOT NULL` check: emit the operand
954                    // then `IsNull { negated: !*negated }`. For ILIKE
955                    // `%%` on 25 k rows the per-row `like_match_inner`
956                    // → 2-char walk (~30 ns each) becomes a tag check
957                    // (~3 ns); the operand still gets evaluated for the
958                    // NULL semantics that SQL `LIKE` requires.
959                    if !pat.is_empty() && pat.chars().all(|c| c == '%') {
960                        compile_into(expr, ctx, steps);
961                        steps.push(Step::AnyTextMatch { negated: *negated });
962                        return;
963                    }
964                    compile_into(expr, ctx, steps);
965                    let chars: alloc::vec::Vec<char> = if *case_insensitive {
966                        pat.to_lowercase().chars().collect()
967                    } else {
968                        pat.chars().collect()
969                    };
970                    // v7.39 — `%[k×_]lit[m×_]%` runs on the substring fast
971                    // path (see Step::LikeSubstring).
972                    if let Some((k, needle, m)) = like_substring_shape(&chars) {
973                        steps.push(Step::LikeSubstring {
974                            needle,
975                            k_before: k,
976                            m_after: m,
977                            negated: *negated,
978                            case_insensitive: *case_insensitive,
979                        });
980                        return;
981                    }
982                    steps.push(Step::Like {
983                        pattern: chars,
984                        negated: *negated,
985                        case_insensitive: *case_insensitive,
986                    });
987                }
988                _ => steps.push(Step::Subtree(e.clone())),
989            }
990        }
991        // v7.39 (round 594) — a literal-pattern regex compiles here instead
992        // of once per row. `s ~ 'p'` and `s ~* 'p'` both lower to
993        // `regexp_like`, so this one shape covers the operators too. A
994        // pattern that is not a literal (or flags that are not) stays on the
995        // interpreter, which still has to compile per row: the pattern can
996        // differ row to row.
997        Expr::FunctionCall { name, args }
998            if name.eq_ignore_ascii_case("regexp_like")
999                && matches!(args.len(), 2 | 3)
1000                && regex_literal_parts(args.as_slice()).is_some()
1001                && fully_compilable(&args[0]) =>
1002        {
1003            let (pat, ci) = regex_literal_parts(args.as_slice()).expect("checked above");
1004            match crate::eval::compile_re(pat, ci) {
1005                Ok(re) => {
1006                    compile_into(&args[0], ctx, steps);
1007                    steps.push(Step::Regex {
1008                        re,
1009                        fallback: e.clone(),
1010                    });
1011                }
1012                // An invalid pattern is an error the interpreter words; let
1013                // it keep raising it, in its own wording.
1014                Err(_) => steps.push(Step::Subtree(e.clone())),
1015            }
1016        }
1017        // v7.36 — PURE scalar function call: emit args then a
1018        // single Function step that pops them. `fully_compilable`
1019        // gates the whitelist + recurses into args, so this branch
1020        // only fires when the entire subtree is compilable.
1021        Expr::FunctionCall { name, args } if is_pure_scalar_function(name) => {
1022            // v7.36 — specialise `LENGTH(<column>)` /
1023            // `OCTET_LENGTH(<column>)` so the column's `Value::Text`
1024            // isn't cloned just to read its length. The general
1025            // `Step::Function` path goes through `apply_function`,
1026            // which can't borrow off the stack — it copies.
1027            let lower = name.to_ascii_lowercase();
1028            if args.len() == 1 {
1029                if let Expr::Column(c) = &args[0]
1030                    && let Some(pos) = compile_column_pos(c, ctx)
1031                {
1032                    match lower.as_str() {
1033                        "length" | "char_length" | "character_length" => {
1034                            steps.push(Step::ColumnLength { pos });
1035                            return;
1036                        }
1037                        "octet_length" => {
1038                            steps.push(Step::ColumnOctetLength { pos });
1039                            return;
1040                        }
1041                        _ => {}
1042                    }
1043                }
1044            }
1045            for a in args {
1046                compile_into(a, ctx, steps);
1047            }
1048            // v7.39 (round 621) — COALESCE / NULLIF compile to their own
1049            // steps (see the variants) so the chosen argument stays borrowed.
1050            // The arguments are already on the stack from the loop above — a
1051            // first cut recompiled them here and doubled them.
1052            if name.eq_ignore_ascii_case("coalesce") && !args.is_empty() {
1053                steps.push(Step::Coalesce { n_args: args.len() });
1054                return;
1055            }
1056            if name.eq_ignore_ascii_case("nullif") && args.len() == 2 {
1057                steps.push(Step::NullIf);
1058                return;
1059            }
1060            // v7.39 (round 717) — GREATEST / LEAST get their own step;
1061            // see the variant.
1062            if (lower == "greatest" || lower == "least") && !args.is_empty() {
1063                steps.push(Step::Extremum {
1064                    n_args: args.len(),
1065                    max: lower == "greatest",
1066                });
1067                return;
1068            }
1069            steps.push(Step::Function {
1070                name_lower: lower,
1071                n_args: args.len(),
1072            });
1073        }
1074        // v7.39 (round 605) — a CONSTANT subexpression is evaluated once here
1075        // rather than for every row. `WHERE id < ('500')::INT` cost two
1076        // allocations a row against none for `WHERE id < 500`, and the same
1077        // gap is much wider in a projection. A literal is already a `Lit`
1078        // step, so this is only about the shapes built OUT of literals.
1079        //
1080        // An error stays where it was: if the fold does not evaluate, the
1081        // expression compiles as before and raises per row, in the
1082        // interpreter's own wording.
1083        e if !matches!(e, Expr::Literal(_)) && constant_expr(e) => {
1084            match eval_expr(e, &Row::new(alloc::vec::Vec::new()), ctx) {
1085                Ok(v) => steps.push(Step::Lit(v)),
1086                Err(_) => steps.push(Step::Subtree(e.clone())),
1087            }
1088        }
1089        // v7.39 (round 597) — `x = ANY (ARRAY[literals])` is `x IN (…)` and
1090        // `x <> ALL (…)` is `x NOT IN (…)`, down to the three-valued
1091        // treatment of a NULL element, so they take the membership set the
1092        // IN list already builds at compile time: 40.9 ms for a ten-element
1093        // array against 2.1 for the IN spelling of the same question. Folding
1094        // the array (below) alone left the per-row cost growing with the
1095        // array's length; a set does not.
1096        Expr::AnyAll {
1097            expr,
1098            op,
1099            array,
1100            is_any,
1101        } if !ctx.mysql_dialect
1102            && ((matches!(op, spg_sql::ast::BinOp::Eq) && *is_any)
1103                || (matches!(op, spg_sql::ast::BinOp::NotEq) && !*is_any))
1104            && array_literal_items(array)
1105                .is_some_and(|it| !it.is_empty() && crate::build_in_list_set(it).is_some())
1106            && fully_compilable(expr) =>
1107        {
1108            let items = array_literal_items(array).expect("checked above");
1109            let entry = crate::build_in_list_set(items).expect("checked above");
1110            compile_into(expr, ctx, steps);
1111            steps.push(Step::InSet {
1112                set: entry.set,
1113                has_null: entry.has_null,
1114                negated: !*is_any,
1115                fallback: e.clone(),
1116            });
1117        }
1118        // v7.39 (round 597) — any other ANY/ALL whose right-hand array is
1119        // constant: build it once here rather than per row.
1120        Expr::AnyAll {
1121            expr,
1122            op,
1123            array,
1124            is_any,
1125        } if constant_expr(array) => {
1126            match eval_expr(array, &Row::new(alloc::vec::Vec::new()), ctx) {
1127                Ok(arr) => {
1128                    compile_into(expr, ctx, steps);
1129                    // v7.39 (round 604) — with the array in hand, an equality
1130                    // ANY / inequality ALL is a membership test whatever the
1131                    // spelling: `'{1,2,3}'::int[]` keeps its elements inside a
1132                    // string, so round 597's literal-list route could not see
1133                    // them, but they are values now.
1134                    if !ctx.mysql_dialect
1135                        && ((matches!(op, spg_sql::ast::BinOp::Eq) && *is_any)
1136                            || (matches!(op, spg_sql::ast::BinOp::NotEq) && !*is_any))
1137                        && let Some(entry) = value_array_in_list_set(&arr)
1138                    {
1139                        steps.push(Step::InSet {
1140                            set: entry.set,
1141                            has_null: entry.has_null,
1142                            negated: !*is_any,
1143                            fallback: e.clone(),
1144                        });
1145                        return;
1146                    }
1147                    steps.push(Step::AnyAll {
1148                        op: *op,
1149                        is_any: *is_any,
1150                        arr,
1151                    });
1152                }
1153                // A constant that does not evaluate is the interpreter's
1154                // error to raise, per row, in its own wording.
1155                Err(_) => steps.push(Step::Subtree(e.clone())),
1156            }
1157        }
1158        // v7.39 (round 595) — EXTRACT over a compilable source. One
1159        // non-compilable node used to disqualify the WHOLE predicate, so
1160        // `WHERE extract(year FROM t) = 2020` interpreted the column read
1161        // and the comparison as well: 81.7 ms on 500k rows against PG18's
1162        // 14.5, where a compiled comparison on the same column is 13.1.
1163        Expr::Extract { field, source } => {
1164            compile_into(source, ctx, steps);
1165            steps.push(Step::Extract {
1166                field: field.clone(),
1167                fallback: e.clone(),
1168            });
1169        }
1170        Expr::Cast { expr, target } => {
1171            // v7.39 (read01 ruleutils.c) — catalog-dependent casts run
1172            // through eval's pre-hook (regclass dual-shape, domain/enum/
1173            // composite named types).
1174            // v7.39 (round 621) — the varchar/char FAMILY is catalog-free, so
1175            // it stays on the compiled path; the blanket Named -> Subtree rule
1176            // sent `s::VARCHAR(20)` to the interpreter, which pays two
1177            // allocations a row. Everything else Named (domains, enums,
1178            // composites, regtypes) still needs the interpreter's catalog.
1179            let named_text_family = match target {
1180                spg_sql::ast::CastTarget::Named(n) => named_varchar_family(n),
1181                _ => false,
1182            };
1183            // v7.39 (round 722) — a plain scalar spelling resolves NOW, not
1184            // per row; see `Step::CastPlain`. The text family keeps its
1185            // dedicated route (the timestamptz::text Subtree guard below
1186            // must still see it).
1187            if let spg_sql::ast::CastTarget::Named(n) = target
1188                && !named_text_family
1189                && let Some(dt) = super::cast::plain_named_target(n)
1190            {
1191                compile_into(expr, ctx, steps);
1192                steps.push(Step::CastPlain {
1193                    dt,
1194                    name: n.clone(),
1195                });
1196                return;
1197            }
1198            if matches!(target, spg_sql::ast::CastTarget::RegClass)
1199                || (matches!(target, spg_sql::ast::CastTarget::Named(_)) && !named_text_family)
1200            {
1201                steps.push(Step::Subtree(e.clone()));
1202                return;
1203            }
1204            // v7.39 (read01 round 76) — `<timestamptz>::text` renders the
1205            // `+00` offset, and tz-ness lives in the *static* type, not in
1206            // the runtime `Value::Timestamp`. `Step::Cast` calls the pure
1207            // `cast_value(value, target)`, which cannot see the expression
1208            // it came from — so a cast the interpreter renders with an
1209            // offset came out without one whenever the compiled VM drove
1210            // it (every cast inside an aggregate argument, and every cast
1211            // over an aggregate result: `string_agg(x::text, ',')`,
1212            // `min(x)::text`). Keep this one shape on Subtree.
1213            if (matches!(target, spg_sql::ast::CastTarget::Text) || named_text_family)
1214                && crate::describe::describe_expr(expr, ctx.columns)
1215                    .is_some_and(|s| matches!(s.ty, spg_storage::DataType::Timestamptz))
1216            {
1217                steps.push(Step::Subtree(e.clone()));
1218                return;
1219            }
1220            compile_into(expr, ctx, steps);
1221            steps.push(Step::Cast {
1222                target: target.clone(),
1223            });
1224        }
1225        Expr::Case {
1226            operand,
1227            branches,
1228            else_branch,
1229        } => {
1230            // Gate by `fully_compilable` at the leaf: if any sub-expr
1231            // can't compile natively, the whole Case stays Subtree so
1232            // a single Case never escapes to a row-materialise eval.
1233            let all_ok = operand.as_deref().is_none_or(fully_compilable)
1234                && branches
1235                    .iter()
1236                    .all(|(w, t)| fully_compilable(w) && fully_compilable(t))
1237                && else_branch.as_deref().is_none_or(fully_compilable);
1238            if !all_ok {
1239                steps.push(Step::Subtree(e.clone()));
1240                return;
1241            }
1242            let op_c = operand.as_deref().map(|o| compile_expr(o, ctx));
1243            let branches_c: alloc::vec::Vec<(CompiledExpr, CompiledExpr)> = branches
1244                .iter()
1245                .map(|(w, t)| (compile_expr(w, ctx), compile_expr(t, ctx)))
1246                .collect();
1247            let else_c = else_branch.as_deref().map(|el| compile_expr(el, ctx));
1248            // Either side byte-wise keeps the whole operand match byte-wise,
1249            // the rule `mysql_text_fold_applies` states for a comparison's
1250            // two operands — a CASE arm is that comparison.
1251            let fold_operand = operand.as_deref().is_some_and(|o| {
1252                branches
1253                    .iter()
1254                    .all(|(w, _)| super::resolve::mysql_text_fold_applies(o, w, ctx))
1255            });
1256            steps.push(Step::Case {
1257                fold_operand,
1258                operand: op_c,
1259                branches: branches_c,
1260                else_branch: else_c,
1261            });
1262            // v7.38 (read01) — resolve the CASE result to PG's common type of
1263            // every THEN/ELSE branch once, here, and append a scale-preserving
1264            // coercion so a taken integer branch is widened to numeric (and
1265            // `pg_typeof` / downstream division match PG). Costs nothing when
1266            // the branches already share a type (common_type → None).
1267            let branch_types: Vec<spg_storage::DataType> = branches
1268                .iter()
1269                .map(|(_, t)| t)
1270                .chain(else_branch.iter().map(|b| b.as_ref()))
1271                .filter_map(|e| crate::describe::describe_expr(e, ctx.columns).map(|s| s.ty))
1272                .collect();
1273            if let Some(common) = crate::describe::common_type(&branch_types) {
1274                steps.push(Step::CoerceCommon(common));
1275            }
1276        }
1277        other => steps.push(Step::Subtree(other.clone())),
1278    }
1279}
1280
1281/// Literal text pattern behind a LIKE/ILIKE, if any.
1282/// v7.39 — recognise `%[k×_]literal[m×_]%` (any number of leading /
1283/// trailing `%`; literal free of `%` / `_` / `\`). Returns
1284/// `(k, literal, m)` when the pattern fits the substring fast path.
1285fn like_substring_shape(pat: &[char]) -> Option<(usize, alloc::string::String, usize)> {
1286    let mut lo = 0;
1287    while lo < pat.len() && pat[lo] == '%' {
1288        lo += 1;
1289    }
1290    if lo == 0 {
1291        return None; // not %-anchored at the front
1292    }
1293    let mut hi = pat.len();
1294    while hi > lo && pat[hi - 1] == '%' {
1295        hi -= 1;
1296    }
1297    if hi == pat.len() {
1298        return None; // not %-anchored at the back
1299    }
1300    let inner = &pat[lo..hi];
1301    let mut i = 0;
1302    while i < inner.len() && inner[i] == '_' {
1303        i += 1;
1304    }
1305    let mut j = inner.len();
1306    while j > i && inner[j - 1] == '_' {
1307        j -= 1;
1308    }
1309    let lit = &inner[i..j];
1310    if lit.is_empty() || lit.iter().any(|&c| c == '%' || c == '_' || c == '\\') {
1311        return None;
1312    }
1313    Some((i, lit.iter().collect(), inner.len() - j))
1314}
1315
1316/// v7.39 — `%[k×_]needle[m×_]%` matcher: walk `str::find` hits of the
1317/// literal and accept one with ≥k chars before it and ≥m chars after.
1318/// v7.39 (round 484) — find `needle` in `hay` at or after `start`.
1319///
1320/// `str::find(&str)` runs the two-way algorithm, and its SETUP is the cost:
1321/// round 484's profile of `s LIKE '%_05%'` put `StrSearcher::new` at 14.6 %
1322/// of self time — rebuilt for every row against a needle that is a compile
1323/// -time constant, and only two bytes long here.
1324///
1325/// An ASCII needle can be scanned as bytes instead: a UTF-8 continuation
1326/// byte is always >= 0x80, so an ASCII byte match can never land inside a
1327/// multi-byte character and every hit is on a char boundary. A non-ASCII
1328/// needle keeps `find`, where that reasoning does not hold.
1329fn like_find_from(hay: &str, needle: &str, start: usize) -> Option<usize> {
1330    if needle.is_empty() {
1331        return Some(start);
1332    }
1333    if !needle.is_ascii() {
1334        return hay[start..].find(needle).map(|rel| start + rel);
1335    }
1336    let h = hay.as_bytes();
1337    let n = needle.as_bytes();
1338    if h.len() < n.len() {
1339        return None;
1340    }
1341    let last = h.len() - n.len();
1342    let mut i = start;
1343    while i <= last {
1344        let off = h[i..=last].iter().position(|&b| b == n[0])?;
1345        let at = i + off;
1346        if &h[at..at + n.len()] == n {
1347            return Some(at);
1348        }
1349        i = at + 1;
1350    }
1351    None
1352}
1353
1354fn like_substring_match(hay: &str, needle: &str, k: usize, m: usize) -> bool {
1355    let mut start = 0;
1356    while let Some(off) = like_find_from(hay, needle, start) {
1357        let before_ok = k == 0 || hay[..off].chars().take(k).count() == k;
1358        let after_ok = m == 0 || hay[off + needle.len()..].chars().take(m).count() == m;
1359        if before_ok && after_ok {
1360            return true;
1361        }
1362        // Advance one char past this hit's start and retry.
1363        match hay[off..].chars().next() {
1364            Some(c) => start = off + c.len_utf8(),
1365            None => return false,
1366        }
1367    }
1368    false
1369}
1370
1371fn literal_text_pattern(pattern: &Expr) -> Option<&str> {
1372    match pattern {
1373        Expr::Literal(Literal::String(s)) => Some(s.as_str()),
1374        _ => None,
1375    }
1376}
1377
1378/// True when the whole tree consists of nodes the compiler models
1379/// natively. Mixed trees stay on the interpreted path: a Subtree
1380/// fallback would run WITHOUT the per-query MemoizeCache, and
1381/// memo-dependent nodes (InList set fast path — round-25) rebuild
1382/// per row there. Measured: compiling a search WHERE with an
1383/// InList subtree regressed 634 ms → 18.7 s.
1384/// v7.39 (round 621) — is this cast an identity on THIS value?
1385///
1386/// `s::TEXT` over a text cell changes nothing, and neither does an unbounded
1387/// `::VARCHAR`; the compiled path used to clone the cell anyway. Only the
1388/// pairs that provably change nothing are listed — a bounded VARCHAR(n) must
1389/// still check its length, numerics their range — so an unlisted pair merely
1390/// keeps the owned path, never a wrong answer.
1391/// The catalog-free varchar/char family, in the canonical `name(p)` spelling
1392/// the parser produces. Only these Named targets stay on the compiled path.
1393fn named_varchar_family(n: &str) -> bool {
1394    let base = n.split('(').next().unwrap_or(n);
1395    base.eq_ignore_ascii_case("varchar")
1396        || base.eq_ignore_ascii_case("text")
1397        || base.eq_ignore_ascii_case("char")
1398        || base.eq_ignore_ascii_case("bpchar")
1399        || base.eq_ignore_ascii_case("character")
1400}
1401
1402/// `varchar(k)`'s k, when the name carries one.
1403fn varchar_limit(n: &str) -> Option<usize> {
1404    let base = n.split('(').next().unwrap_or(n);
1405    if !base.eq_ignore_ascii_case("varchar") {
1406        return None;
1407    }
1408    let inner = n.split('(').nth(1)?.strip_suffix(')')?;
1409    inner.trim().parse().ok()
1410}
1411
1412fn cast_is_identity_for(v: &Value<'_>, target: &spg_sql::ast::CastTarget) -> bool {
1413    match (v, target) {
1414        (Value::Text(_), spg_sql::ast::CastTarget::Text) => true,
1415        (Value::Text(t), spg_sql::ast::CastTarget::Named(n)) => {
1416            // Unbounded text and varchar change nothing. A BOUNDED varchar is
1417            // an identity exactly when the text is within its limit — VARCHAR
1418            // truncates and never pads. CHAR(n) pads, so it is never one.
1419            n.eq_ignore_ascii_case("text")
1420                || n.eq_ignore_ascii_case("varchar")
1421                || varchar_limit(n).is_some_and(|k| t.chars().take(k + 1).count() <= k)
1422        }
1423        (Value::Int(_), spg_sql::ast::CastTarget::Int) => true,
1424        (Value::BigInt(_), spg_sql::ast::CastTarget::BigInt) => true,
1425        (Value::Float(_), spg_sql::ast::CastTarget::Float) => true,
1426        (Value::Bool(_), spg_sql::ast::CastTarget::Bool) => true,
1427        _ => false,
1428    }
1429}
1430
1431pub(crate) fn fully_compilable(e: &Expr) -> bool {
1432    match e {
1433        Expr::Literal(_) | Expr::Column(_) => true,
1434        // v7.38.19 — an all-literal array folds to one `Step::Lit` at
1435        // compile time (the constant arm in `compile_expr`), so it costs
1436        // nothing per row. Leaving it off this list kept the WHOLE
1437        // predicate containing it off the compiled path, and the
1438        // interpreter rebuilt the array -- a `Vec` and a `String` per
1439        // element -- for every row.
1440        //
1441        // Measured on 200,000 rows of sentori's `events`:
1442        //
1443        //     traits ?  'plan'                 10.103 ms
1444        //     traits ?| 'plan'                 10.388     (bare TEXT, same)
1445        //     traits ?| ARRAY['plan']          26.227     (one element)
1446        //     traits ?| ARRAY['x','plan']      39.626
1447        //     traits ?| ARRAY['x','y','plan']  45.403
1448        //
1449        // 35-65 ns per ELEMENT per row, and the second line is what says
1450        // the operator was never the cost.
1451        Expr::Array(items) if items.iter().all(constant_expr) => true,
1452        Expr::Binary { lhs, rhs, .. } => fully_compilable(lhs) && fully_compilable(rhs),
1453        Expr::Unary { expr, .. } | Expr::IsNull { expr, .. } => fully_compilable(expr),
1454        // I2: an InList is compilable ONLY when it becomes a real
1455        // InSet (all-literal list + compilable needle). A
1456        // non-set-able InList must keep the whole tree off the
1457        // compiled path so it never degrades to a memo-less,
1458        // O(list) per-row Subtree (the round-25 18.7 s trap).
1459        Expr::InList { expr, list, .. } => {
1460            fully_compilable(expr) && crate::build_in_list_set(list).is_some()
1461        }
1462        Expr::Like { expr, pattern, .. } => {
1463            fully_compilable(expr) && literal_text_pattern(pattern).is_some()
1464        }
1465        // v7.36 (perf — mailrs Ask 1) — PURE scalar functions over
1466        // compilable args go to `Step::Function`. The whitelist
1467        // covers the high-traffic / non-volatile cases; anything
1468        // outside (NOW, RANDOM, sequence accessors, EXTRACT-with-
1469        // context-dependent fields, etc.) stays on Subtree where
1470        // the interpreter has the full ctx.
1471        // v7.39 (round 594) — a `regexp_like` with a LITERAL pattern is
1472        // compilable even though the function is not on the pure list: the
1473        // pattern becomes a compile product (`Step::Regex`) rather than an
1474        // argument the step would have to re-parse per row. A non-literal
1475        // pattern stays off, because then it really can differ row to row.
1476        Expr::FunctionCall { name, args }
1477            if name.eq_ignore_ascii_case("regexp_like")
1478                && matches!(args.len(), 2 | 3)
1479                && regex_literal_parts(args.as_slice()).is_some() =>
1480        {
1481            fully_compilable(&args[0])
1482        }
1483        Expr::FunctionCall { name, args } => {
1484            is_pure_scalar_function(name) && args.iter().all(fully_compilable)
1485        }
1486        // v7.36 — CAST over a compilable expression. `cast_value`
1487        // is pure / context-free for the scalar targets we care
1488        // about (text, ints, floats, bool, dates).
1489        // v7.39 (read01 ruleutils.c) — regclass / user-named casts
1490        // need the catalog (dual-shape resolve, domain/enum/composite
1491        // hooks); they stay Subtree so eval's pre-hook runs.
1492        Expr::AnyAll { expr, array, .. } if constant_expr(array) => fully_compilable(expr),
1493        Expr::Extract { source, .. } => fully_compilable(source),
1494        Expr::Cast { expr, target } => {
1495            // v7.39 (round 621) — the varchar/char family is catalog-free and
1496            // compiles (the compile arm gates it the same way); other Named
1497            // targets still need eval's catalog pre-hooks.
1498            let target_ok = match target {
1499                spg_sql::ast::CastTarget::RegClass => false,
1500                // v7.39 (round 722) — a compile-time-resolvable plain name
1501                // is as compilable as the dedicated variants; see
1502                // `Step::CastPlain`.
1503                spg_sql::ast::CastTarget::Named(n) => {
1504                    named_varchar_family(n) || super::cast::plain_named_target(n).is_some()
1505                }
1506                _ => true,
1507            };
1508            target_ok && fully_compilable(expr)
1509        }
1510        // v7.37.5-A2b — `CASE [operand] WHEN x THEN y … ELSE z END`
1511        // when every sub-expression is itself fully-compilable. Hot
1512        // shape: Track A's 14 aggregates over
1513        // `COUNT(DISTINCT CASE WHEN m.message_id != '' THEN
1514        //                          m.message_id
1515        //                     ELSE CAST(m.id AS TEXT) END)` — without
1516        // this, every Case fell to `arg_compiled = None`, forced
1517        // `needs_mat = true` per-row, and triggered a full combined-
1518        // row `Vec<Value>` clone for the eval path.
1519        Expr::Case {
1520            operand,
1521            branches,
1522            else_branch,
1523        } => {
1524            operand.as_deref().is_none_or(fully_compilable)
1525                && branches
1526                    .iter()
1527                    .all(|(w, t)| fully_compilable(w) && fully_compilable(t))
1528                && else_branch.as_deref().is_none_or(fully_compilable)
1529        }
1530        _ => false,
1531    }
1532}
1533
1534/// v7.39 (round 595) — functions that are NOT context-free but ARE fixed for
1535/// the whole statement: they read the session's time zone, DateStyle or
1536/// lc_time out of the `EvalContext`, and `Step::Function` hands that context
1537/// to `apply_function_lower` exactly as the interpreter would.
1538///
1539/// Keeping them off the compiled path cost the whole predicate, not just the
1540/// call: one non-compilable node disqualifies the entire WHERE, so
1541/// `WHERE date_trunc('day', t) = TIMESTAMP '…'` interpreted the column read
1542/// and the comparison too — 153.8 ms over 500k rows against PG18's 9.7,
1543/// where a compiled comparison on the same column is 13.1.
1544///
1545/// `now` / `random` / sequence accessors stay off: they are not fixed for
1546/// the statement in the way these are.
1547fn is_session_deterministic_function(name: &str) -> bool {
1548    matches!(
1549        name.to_ascii_lowercase().as_str(),
1550        // v7.39 (round 717) — `format` belongs here, not on the pure
1551        // list: it renders arguments through the SESSION's RenderStyle
1552        // (datestyle / extra_float_digits / bytea_output), exactly the
1553        // dependency class to_char carries. Its absence from BOTH lists
1554        // was the round-716 panel's 4.89× cell — the only remaining
1555        // text-shape loss that was pure fallback tax.
1556        "date_trunc" | "date_part" | "to_char" | "age" | "format"
1557    )
1558}
1559
1560/// v7.36 — PURE scalar function whitelist for `Step::Function`.
1561/// "Pure" means: deterministic, context-independent, no side
1562/// effects. Aggregate names (sum / count / max / …) are filtered
1563/// upstream by the caller — they never reach the compiler. NOW /
1564/// RANDOM / sequence accessors are excluded because they need the
1565/// `EvalContext`'s clock / sequence resolver and aren't
1566/// deterministic. EXTRACT is excluded because the field kind is
1567/// parsed off the Expr tree, not an arg.
1568fn is_pure_scalar_function(name: &str) -> bool {
1569    is_session_deterministic_function(name)
1570        || matches!(
1571            name.to_ascii_lowercase().as_str(),
1572            // string length + slicing
1573            "length"
1574                | "char_length"
1575                | "character_length"
1576                | "octet_length"
1577                | "upper"
1578                | "lower"
1579                | "trim"
1580                | "ltrim"
1581                | "rtrim"
1582                | "btrim"
1583                | "left"
1584                | "right"
1585                | "substring"
1586                | "substr"
1587                | "replace"
1588                | "position"
1589                | "strpos"
1590                | "concat"
1591                | "concat_ws"
1592                | "reverse"
1593                | "repeat"
1594                | "lpad"
1595                | "rpad"
1596                | "split_part"
1597                // v7.39 (round 728) — the JSON constructors: pure over
1598                // their arguments (JSON's number/text rendering is fixed
1599                // by the format, not the session's RenderStyle — probed
1600                // against the ::JSONB cast lane, already whitelisted).
1601                // v7.39 (round 730) — the digest family: pure bytes-in,
1602                // hex/bytea-out. count(md5(s)) was the panel's last
1603                // serial-lane text cell (2.37×): the hash itself is
1604                // ~40% faster than PG's per call here, and ALL of the
1605                // loss was the missing parallel lane.
1606                | "md5"
1607                | "sha224"
1608                | "sha256"
1609                | "sha384"
1610                | "sha512"
1611                | "to_json"
1612                | "to_jsonb"
1613                | "jsonb_build_object"
1614                | "json_build_object"
1615                | "jsonb_build_array"
1616                | "json_build_array"
1617                // null/conditional
1618                | "coalesce"
1619                | "nullif"
1620                | "greatest"
1621                | "least"
1622                | "ifnull"
1623                | "isnull"
1624                | "nvl"
1625                // numeric
1626                | "abs"
1627                | "ceil"
1628                | "ceiling"
1629                | "floor"
1630                | "round"
1631                | "trunc"
1632                | "sqrt"
1633                | "power"
1634                | "pow"
1635                | "mod"
1636                | "sign"
1637                | "log"
1638                | "log10"
1639                | "exp"
1640                | "ln"
1641                // boolean / cast helpers
1642                | "cast"
1643        )
1644}
1645
1646/// r1021 — the arithmetic the integer lane runs. `Div` and `Mod` are in
1647/// because their zero divisor is handled by falling back, not by guessing.
1648const fn is_int_arithmetic(op: BinOp) -> bool {
1649    matches!(
1650        op,
1651        BinOp::Add | BinOp::Sub | BinOp::Mul | BinOp::Div | BinOp::Mod
1652    )
1653}
1654
1655/// r1021 — the comparison that ends an integer-lane program.
1656const fn is_int_comparison(op: BinOp) -> bool {
1657    matches!(
1658        op,
1659        BinOp::Eq | BinOp::NotEq | BinOp::Lt | BinOp::LtEq | BinOp::Gt | BinOp::GtEq
1660    )
1661}
1662
1663/// r1021 — `(value, is_32_bit)` for the two widths the lane admits. NULL,
1664/// `smallint` and every non-integer answer `None`, which sends the row to
1665/// the ordinary machine.
1666const fn int_operand(v: &Value<'_>) -> Option<(i64, bool)> {
1667    match v {
1668        Value::Int(i) => Some((*i as i64, true)),
1669        Value::BigInt(i) => Some((*i, false)),
1670        _ => None,
1671    }
1672}
1673
1674pub(crate) fn compile_expr(e: &Expr, ctx: &EvalContext<'_>) -> CompiledExpr {
1675    let mut steps = Vec::new();
1676    compile_into(e, ctx, &mut steps);
1677    let mut c = CompiledExpr {
1678        steps,
1679        pred_shape: PredShape::Other,
1680    };
1681    // Classified through the very matchers the row loop will use, so the
1682    // label and the destructuring cannot disagree.
1683    c.pred_shape = if c.as_column_cmp_literal().is_some() {
1684        PredShape::ColumnCmpLit
1685    } else if c.as_column_in_set().is_some() {
1686        PredShape::ColumnInSet
1687    } else if c.as_column_like().is_some() {
1688        PredShape::ColumnLike
1689    } else if c.is_int_arith_pred() {
1690        PredShape::IntArith
1691    } else {
1692        PredShape::Other
1693    };
1694    c
1695}
1696
1697/// Run a compiled program. `stack` is caller-owned scratch
1698/// (cleared here) so tight row loops never touch the allocator
1699/// for the machine itself.
1700pub(crate) fn eval_compiled(
1701    c: &CompiledExpr,
1702    row: &Row<'static>,
1703    ctx: &EvalContext<'_>,
1704    stack: &mut Vec<Value<'static>>,
1705) -> Result<Value<'static>, EvalError> {
1706    // v7.37.16 — reuse the caller's stack allocation across rows.
1707    // v7.37.9 T3 S2 had severed this: `eval_compiled_ref` pushes
1708    // `Value<'val>` where `'val` is the per-call RowRef borrow, and
1709    // `Vec<Value<'val>>` is invariant in `'val`, so the caller's
1710    // `Vec<Value<'static>>` could not be lent in-place and every call
1711    // allocated a fresh local Vec. That was sized for the ~50×/query
1712    // post-group projection path, but the aggregate/scan WHERE filter
1713    // loops (select.rs) call this once PER ROW — 50 k allocs/query on
1714    // a 50 k-row filter (the heavy.rs filter_agg 1.5×-vs-PG18 loss).
1715    // Instead: MOVE the caller's Vec in (covariant shrink 'static →
1716    // 'val, safe), run, then hand the emptied allocation back via
1717    // `recycle_stack`. Zero per-row alloc; the borrowed-push (S2/S3)
1718    // zero-clone Text path is untouched.
1719    let rowref = crate::join::RowRef::Owned(row);
1720    let mut local_stack: Vec<Value<'_>> = core::mem::take(stack);
1721    let result = eval_compiled_ref(c, rowref, ctx, &mut local_stack);
1722    let owned = result.map(Value::into_owned);
1723    *stack = recycle_stack(local_stack);
1724    owned
1725}
1726
1727/// v7.39 (round 479) — evaluate a compiled WHERE and answer the bool,
1728/// without ever materialising an owned `Value`.
1729///
1730/// `eval_compiled` ends in `result.map(Value::into_owned)` because its
1731/// contract is to hand back a `Value<'static>`. A predicate does not want
1732/// a value at all — it wants one bool — and round 478's profile put
1733/// `Value::into_owned` at 5.8 % of self time and `drop_glue<Value>` at
1734/// 15.1 %, against 5.5 % for the comparison the predicate exists to
1735/// perform. The `into_owned` and the owned value's drop are both pure
1736/// overhead on this path.
1737///
1738/// Everything else is `eval_compiled`'s bridge unchanged: the caller's
1739/// stack is moved in (covariant shrink), run, and handed back emptied.
1740pub(crate) fn eval_compiled_pred(
1741    c: &CompiledExpr,
1742    row: &Row<'static>,
1743    ctx: &EvalContext<'_>,
1744    stack: &mut Vec<Value<'static>>,
1745    mysql: bool,
1746) -> Result<bool, EvalError> {
1747    // The shape was settled at compile time; the row loop reads one
1748    // discriminant instead of re-matching the step list per row.
1749    match c.pred_shape {
1750        // r1021 — integer arithmetic runs on an i64 stack, building no
1751        // `Value` at all. `None` means the row carried something the lane
1752        // does not decide (NULL, a non-integer, a zero divisor, an
1753        // overflow) and the ordinary machine below answers it instead.
1754        PredShape::IntArith => {
1755            if let Some(verdict) = c.eval_int_arith_pred(row) {
1756                crate::bump_counter!(STEP_VM_INTLANE_FIRE);
1757                return Ok(verdict);
1758            }
1759            crate::bump_counter!(STEP_VM_INTLANE_FALLBACK);
1760        }
1761        // v7.39 (round 482) — `<column> <cmp> <literal>` compares in place.
1762        //
1763        // The general path builds three `Value`s a row and drops them;
1764        // this one reads both operands by reference and builds only the
1765        // comparison result. `apply_binary_by_ref` is the SAME function
1766        // `Step::Binary` reaches for first, so the answer is identical by
1767        // construction rather than by a second reading of the semantics.
1768        PredShape::ColumnCmpLit => {
1769            if let Some((pos, op, lit)) = c.as_column_cmp_literal() {
1770                crate::bump_counter!(STEP_VM_FASTPRED_FIRE);
1771                let cell = row.values.get(pos).unwrap_or(&Value::Null);
1772                if let Some(res) = super::apply_binary_by_ref(op, cell, lit)? {
1773                    return crate::eval::predicate_is_true(&res, "WHERE", mysql);
1774                }
1775                // The by-ref form declined (an op that builds an owned
1776                // result); fall through rather than answer differently
1777                // from the VM.
1778            }
1779        }
1780        // v7.39 (round 486) — `<column> [NOT] IN (<literals>)` looks the
1781        // cell up in place. Same `in_set_verdict` the `InSet` step calls,
1782        // so the answer is identical by construction; a family mismatch
1783        // returns None and falls through to the general path, which takes
1784        // the step's interpreter fallback.
1785        PredShape::ColumnInSet => {
1786            if let Some((pos, set, has_null, negated)) = c.as_column_in_set() {
1787                let cell = row.values.get(pos).unwrap_or(&Value::Null);
1788                if let Some(v) = in_set_verdict(cell, set, has_null, negated) {
1789                    crate::bump_counter!(STEP_VM_FASTPRED_FIRE);
1790                    return crate::eval::predicate_is_true(&v, "WHERE", mysql);
1791                }
1792            }
1793        }
1794        // v7.39 (round 488) — `<column> [NOT] [I]LIKE '<literal>'` matches
1795        // straight off the cell. The matcher wanted a `&str` all along;
1796        // the VM was pushing a `Value` and popping it for no other reason.
1797        PredShape::ColumnLike => {
1798            if let Some((pos, step)) = c.as_column_like() {
1799                let cell = row.values.get(pos).unwrap_or(&Value::Null);
1800                if let Some(v) = like_verdict(cell, step) {
1801                    crate::bump_counter!(STEP_VM_FASTPRED_FIRE);
1802                    return crate::eval::predicate_is_true(&v?, "WHERE", mysql);
1803                }
1804            }
1805        }
1806        PredShape::Other => {}
1807    }
1808    let rowref = crate::join::RowRef::Owned(row);
1809    let mut local_stack: Vec<Value<'_>> = core::mem::take(stack);
1810    let verdict = eval_compiled_ref(c, rowref, ctx, &mut local_stack)
1811        .and_then(|v| crate::eval::predicate_is_true(&v, "WHERE", mysql));
1812    *stack = recycle_stack(local_stack);
1813    verdict
1814}
1815
1816/// v7.39 (round 486) — the membership decision, shared by `Step::InSet`
1817/// and by the fast predicate below so the two cannot drift. `None` means
1818/// the needle's family does not match the set's, which is the caller's
1819/// cue to take the interpreter's coercion path on the whole node.
1820///
1821/// v7.39 (round 489) — `#[inline(always)]` is load-bearing, and the
1822/// measurement behind it is worth stating because round 486 got it wrong.
1823/// Round 486 saw the shared-helper form cost `like_filter` 4.5 % and
1824/// concluded "editing this loop is expensive"; it then duplicated the
1825/// body into the arm to avoid touching it. Re-measured with the shape
1826/// ISOLATED (round 488 found the panel's shapes contaminate each other),
1827/// `like_filter` shows no such cost — that reading was its neighbours.
1828/// What IS real is `big_in`: +4.6 % with a plain call, separated spreads,
1829/// on a shape that takes the fast path and never executes this arm.
1830/// `#[inline(always)]` returns it to parity (-0.1 %, overlapping), so the
1831/// duplicate bought nothing and is gone.
1832///
1833/// `e2e_in_set_fast_path_round486` still runs every needle × set ×
1834/// negated × has-NULL combination down BOTH entry points.
1835#[allow(clippy::inline_always)] // measured: see the note above
1836#[inline(always)]
1837fn in_set_verdict(
1838    needle: &Value<'_>,
1839    set: &crate::memoize::InListSet,
1840    has_null: bool,
1841    negated: bool,
1842) -> Option<Value<'static>> {
1843    let contained = match (needle, set) {
1844        // Non-empty list + NULL needle → NULL (NOT NULL is still NULL) —
1845        // matches the interpreter and eval_with_in_sets.
1846        (Value::Null, _) => return Some(Value::Null),
1847        (Value::SmallInt(n), crate::memoize::InListSet::Int(s)) => s.contains(&i64::from(*n)),
1848        (Value::Int(n), crate::memoize::InListSet::Int(s)) => s.contains(&i64::from(*n)),
1849        (Value::BigInt(n), crate::memoize::InListSet::Int(s)) => s.contains(n),
1850        (Value::Text(t), crate::memoize::InListSet::Text(s)) => s.contains(t.as_ref()),
1851        _ => return None,
1852    };
1853    let inner = if contained {
1854        Value::Bool(true)
1855    } else if has_null {
1856        Value::Null
1857    } else {
1858        Value::Bool(false)
1859    };
1860    Some(match (negated, inner) {
1861        (true, Value::Bool(b)) => Value::Bool(!b),
1862        (_, v) => v,
1863    })
1864}
1865
1866/// v7.39 (round 604) — the membership set of an ALREADY-EVALUATED constant
1867/// array.
1868///
1869/// Round 597 gave `x = ANY (ARRAY[1,2,3])` the same set an IN list builds,
1870/// which took it from 268 ms over 500k rows to 1.93. It could not do the
1871/// same for `x = ANY ('{1,2,3}'::int[])`, because it built the set from AST
1872/// literals and that spelling keeps its elements inside a string: the array
1873/// was folded once but every row still walked it, and the shape stayed at
1874/// 43.49 ms against PG18's 9.37. The array has been evaluated by the time
1875/// this is asked, so the elements are right there.
1876///
1877/// The families are the ones `build_in_list_set` accepts, for the same
1878/// reason: an integer set answers `Int = BigInt` correctly across widths,
1879/// and a text set compares verbatim. Anything else — a mixed array, floats,
1880/// NUMERIC, dates — returns `None` and keeps the folded-array walk.
1881fn value_array_in_list_set(arr: &Value<'_>) -> Option<crate::memoize::InListSetEntry> {
1882    let len = crate::eval::values::array_len(arr)?;
1883    if len == 0 {
1884        return None;
1885    }
1886    let mut ints: hashbrown::HashSet<i64> = hashbrown::HashSet::with_capacity(len);
1887    let mut texts: hashbrown::HashSet<alloc::string::String> =
1888        hashbrown::HashSet::with_capacity(len);
1889    let mut has_null = false;
1890    for i in 0..len {
1891        match crate::eval::values::array_element_at(arr, i) {
1892            None | Some(Value::Null) => has_null = true,
1893            Some(Value::SmallInt(n)) => {
1894                ints.insert(i64::from(n));
1895            }
1896            Some(Value::Int(n)) => {
1897                ints.insert(i64::from(n));
1898            }
1899            Some(Value::BigInt(n)) => {
1900                ints.insert(n);
1901            }
1902            Some(Value::Text(s) | Value::BpChar(s)) => {
1903                texts.insert(s.into_owned());
1904            }
1905            _ => return None,
1906        }
1907        if !ints.is_empty() && !texts.is_empty() {
1908            return None;
1909        }
1910    }
1911    let set = if !ints.is_empty() {
1912        crate::memoize::InListSet::Int(ints)
1913    } else if !texts.is_empty() {
1914        crate::memoize::InListSet::Text(texts)
1915    } else {
1916        return None;
1917    };
1918    Some(crate::memoize::InListSetEntry { set, has_null })
1919}
1920
1921/// v7.39 (round 597) — the literal elements of an `ARRAY[…]` constructor.
1922/// `None` for any other right-hand side, including the `'{1,2}'::int[]`
1923/// spelling, whose elements live inside a string rather than the tree.
1924fn array_literal_items(e: &Expr) -> Option<&[Expr]> {
1925    match e {
1926        Expr::Array(items) if items.iter().all(constant_expr) => Some(items.as_slice()),
1927        _ => None,
1928    }
1929}
1930
1931/// v7.39 (round 605) — the value of a projection item that cannot depend on
1932/// the row, evaluated once. `None` for anything that depends on a row, or
1933/// that fails to evaluate — the latter so its error still comes from the row
1934/// loop, in the interpreter's own wording, rather than from planning.
1935pub(crate) fn constant_projection_value(e: &Expr, ctx: &EvalContext<'_>) -> Option<Value<'static>> {
1936    if matches!(e, Expr::Literal(_)) || !constant_expr(e) {
1937        return None;
1938    }
1939    eval_expr(e, &Row::new(alloc::vec::Vec::new()), ctx).ok()
1940}
1941
1942/// v7.39 (round 597) — an expression whose value cannot depend on the row.
1943/// An allowlist of node kinds, for the reason rounds 590 and 596 recorded:
1944/// asking "does it mention a column" would admit a node the walk did not
1945/// know about, and a function whose volatility SPG cannot look up.
1946///
1947/// r1043 — there is a lookup now, so a call to a function on
1948/// [`crate::immutable_fn`]'s positive list joins the list. Round 605's
1949/// own note recorded `upper('abc')` as the one it could not fold; the
1950/// gate it was missing was never the node kind, it was the volatility.
1951/// A name that has not been checked against `pg_proc` is still refused,
1952/// so the failure mode stays "not folded" rather than "folded wrongly".
1953pub(crate) fn constant_expr(e: &Expr) -> bool {
1954    match e {
1955        Expr::Literal(_) => true,
1956        Expr::Array(items) => items.iter().all(constant_expr),
1957        Expr::Unary { expr, .. } | Expr::Cast { expr, .. } => constant_expr(expr),
1958        Expr::Binary { lhs, rhs, .. } => constant_expr(lhs) && constant_expr(rhs),
1959        Expr::FunctionCall { name, args } => {
1960            crate::immutable_fn::is_immutable_builtin(name) && args.iter().all(constant_expr)
1961        }
1962        _ => false,
1963    }
1964}
1965
1966/// v7.39 (round 595) — the source sub-expression of the EXTRACT node a
1967/// `Step::Extract` was compiled from. Only its declared TYPE is read, for
1968/// the error wording; the value came off the stack.
1969fn source_of_extract(node: &Expr) -> &Expr {
1970    match node {
1971        Expr::Extract { source, .. } => source,
1972        other => other,
1973    }
1974}
1975
1976/// v7.39 (round 594) — the literal pattern and case flag of a `regexp_like`
1977/// call, when both are literals. `None` keeps the call on the interpreter.
1978fn regex_literal_parts(args: &[Expr]) -> Option<(&str, bool)> {
1979    let Expr::Literal(spg_sql::ast::Literal::String(pat)) = &args[1] else {
1980        return None;
1981    };
1982    let ci = match args.get(2) {
1983        None => false,
1984        Some(Expr::Literal(spg_sql::ast::Literal::String(f))) => f.contains('i'),
1985        Some(_) => return None,
1986    };
1987    Some((pat.as_str(), ci))
1988}
1989
1990/// The verdict `Step::Regex` produces. `None` means the operand is not text,
1991/// which is the caller's cue to fall through to the interpreter for its own
1992/// coercion and wording.
1993fn regex_verdict(
1994    cell: &Value<'_>,
1995    re: &crate::eval::CompiledRe,
1996) -> Option<Result<Value<'static>, EvalError>> {
1997    let text = match cell {
1998        Value::Null => return Some(Ok(Value::Null)),
1999        Value::Text(t) | Value::BpChar(t) => t.as_ref(),
2000        _ => return None,
2001    };
2002    Some(crate::eval::compiled_is_match(re, text).map(Value::Bool))
2003}
2004
2005/// v7.39 (round 488) — the verdict `Step::Like` / `Step::LikeSubstring`
2006/// produce, restated for the fast predicate. `None` means the operand is
2007/// not text, which is the caller's cue to fall through to the VM and let
2008/// it raise the type error in its own wording.
2009///
2010/// v7.39 (round 489) — the VM arm calls this too, so there is one body
2011/// rather than two that can drift. Round 488 kept them separate on round
2012/// 486's belief that editing that loop costs unrelated shapes; round 489
2013/// re-measured that belief with the shapes isolated and force-inlined the
2014/// helper, and the cost is gone (see `in_set_verdict`).
2015/// `e2e_like_fast_path_round488` runs both entry points over the same
2016/// matrix.
2017#[allow(clippy::inline_always)] // measured: see `in_set_verdict`
2018#[inline(always)]
2019fn like_verdict(cell: &Value<'_>, step: &Step) -> Option<Result<Value<'static>, EvalError>> {
2020    let (text, negated) = match (cell, step) {
2021        (Value::Null, _) => return Some(Ok(Value::Null)),
2022        (
2023            Value::Text(t) | Value::BpChar(t),
2024            Step::Like { negated, .. } | Step::LikeSubstring { negated, .. },
2025        ) => (t.as_ref(), *negated),
2026        _ => return None,
2027    };
2028    let matched = match step {
2029        Step::Like {
2030            pattern,
2031            case_insensitive,
2032            ..
2033        } => {
2034            let r = if *case_insensitive {
2035                like_match_str(&text.to_lowercase(), pattern, 0)
2036            } else {
2037                like_match_str(text, pattern, 0)
2038            };
2039            match r {
2040                Ok(m) => m,
2041                Err(e) => return Some(Err(e)),
2042            }
2043        }
2044        Step::LikeSubstring {
2045            needle,
2046            k_before,
2047            m_after,
2048            case_insensitive,
2049            ..
2050        } => {
2051            if *case_insensitive {
2052                like_substring_match(&text.to_lowercase(), needle, *k_before, *m_after)
2053            } else {
2054                like_substring_match(text, needle, *k_before, *m_after)
2055            }
2056        }
2057        _ => return None,
2058    };
2059    Some(Ok(Value::Bool(if negated { !matched } else { matched })))
2060}
2061
2062/// Return an emptied stack's allocation with its value lifetime reset.
2063/// This is the standard "recycle" pattern (cf. the `recycle_vec` crate):
2064/// an EMPTY `Vec<Value<'a>>` holds no values, only a raw allocation, so
2065/// re-labelling its element lifetime cannot dangle.
2066#[allow(unsafe_code)] // empty-Vec lifetime relabel; isolated (see SAFETY).
2067fn recycle_stack(mut v: Vec<Value<'_>>) -> Vec<Value<'static>> {
2068    // v7.39 (round 481) — read before the clear: this is exactly the set of
2069    // values the clear is about to drop.
2070    crate::bump_counter!(STEP_VM_STACK_LEFTOVER, v.len() as u64);
2071    #[cfg(feature = "perf-counters")]
2072    {
2073        let heap = v
2074            .iter()
2075            .filter(|x| {
2076                matches!(
2077                    x,
2078                    Value::Text(_) | Value::Bytes(_) | Value::Json(_) | Value::Vector(_)
2079                )
2080            })
2081            .count();
2082        crate::bump_counter!(STEP_VM_STACK_LEFTOVER_HEAP, heap as u64);
2083    }
2084    v.clear();
2085    debug_assert!(v.is_empty());
2086    // SAFETY: `v` is empty (cleared above) — there are no `Value<'_>`s
2087    // whose lifetime could be unsoundly extended; `Vec<Value<'a>>` and
2088    // `Vec<Value<'static>>` are the same type constructor differing only
2089    // in a lifetime parameter, so they have identical size/align/layout
2090    // (lifetimes are erased before layout is computed).
2091    unsafe { core::mem::transmute::<Vec<Value<'_>>, Vec<Value<'static>>>(v) }
2092}
2093
2094/// v7.32 (P4 borrow channel, increment 2) — the RowRef-borrowing form of
2095/// `eval_compiled`. `Step::Column` borrows its cell straight from the
2096/// RowRef (a join tuple resolves it via `tuple_value`, never
2097/// materialising a combined Row); only the rare Subtree / InSet
2098/// cross-family fallback materialises the row once. Bit-for-bit
2099/// equivalent to the Owned path — `eval_compiled` above is now a thin
2100/// `RowRef::Owned` wrapper, so there is a single interpreter (invariant
2101/// I3); a differential test pins the equivalence.
2102// v7.37.9 T3 S1 — row-lifetime stack plumbing. Two lifetimes:
2103// `'row` = the RowRef's data lifetime; `'val` = stack value lifetime
2104// (must outlive function return). Constraint `'row: 'val` allows the
2105// step body to push `Value::Text(Cow::Borrowed(row_cell))` (S2+) while
2106// the caller's stack stays at whatever lifetime it declared (often
2107// `'static` for Vec<Value<'static>>). S1 keeps every step body forcing
2108// `.into_owned()` so behaviour is bit-identical; later stages
2109// (S2 Column, S3 Lit, S4 Binary, S6 Function, S7 Case) progressively
2110// switch to borrowed push to eliminate per-row String allocs.
2111pub(crate) fn eval_compiled_ref<'row, 'val>(
2112    c: &'val CompiledExpr,
2113    // v7.39 (round 656) — BY VALUE. `RowRef` is `Copy` and its `get`
2114    // borrows the row data, not the wrapper, so taking a reference here
2115    // only served to tie the result's lifetime to a caller local — which
2116    // is what stopped the aggregate loop from holding its `RowRef` by
2117    // value and forced a materialised `Vec<RowRef>` per scan.
2118    row: crate::join::RowRef<'row>,
2119    ctx: &EvalContext<'_>,
2120    stack: &mut Vec<Value<'val>>,
2121) -> Result<Value<'val>, EvalError>
2122where
2123    'row: 'val,
2124{
2125    stack.clear();
2126    run_compiled_steps(&c.steps, row, ctx, stack)?;
2127    Ok(stack.pop().unwrap_or(Value::Null))
2128}
2129
2130/// v7.37.5-A2b — append-mode entry point for nested sub-programs (the
2131/// `Step::Case` executor's per-branch evaluations). Does NOT clear the
2132/// stack; pushes the program's result on top of whatever was already
2133/// there. Caller uses the `mark` to know where to truncate / pop. Kept
2134/// out of public surface — only the Case opcode reaches for it.
2135fn eval_compiled_ref_into<'row, 'val>(
2136    c: &'val CompiledExpr,
2137    // v7.39 (round 656) — BY VALUE. `RowRef` is `Copy` and its `get`
2138    // borrows the row data, not the wrapper, so taking a reference here
2139    // only served to tie the result's lifetime to a caller local — which
2140    // is what stopped the aggregate loop from holding its `RowRef` by
2141    // value and forced a materialised `Vec<RowRef>` per scan.
2142    row: crate::join::RowRef<'row>,
2143    ctx: &EvalContext<'_>,
2144    stack: &mut Vec<Value<'val>>,
2145    _mark: usize,
2146) -> Result<(), EvalError>
2147where
2148    'row: 'val,
2149{
2150    run_compiled_steps(&c.steps, row, ctx, stack)
2151}
2152
2153#[inline]
2154fn run_compiled_steps<'row, 'val>(
2155    steps: &'val [Step],
2156    // v7.39 (round 656) — BY VALUE. `RowRef` is `Copy` and its `get`
2157    // borrows the row data, not the wrapper, so taking a reference here
2158    // only served to tie the result's lifetime to a caller local — which
2159    // is what stopped the aggregate loop from holding its `RowRef` by
2160    // value and forced a materialised `Vec<RowRef>` per scan.
2161    row: crate::join::RowRef<'row>,
2162    ctx: &EvalContext<'_>,
2163    stack: &mut Vec<Value<'val>>,
2164) -> Result<(), EvalError>
2165where
2166    'row: 'val,
2167{
2168    // v7.37.9 Phase 1A-ext-2 T1 — counter per call into the Step VM
2169    // interpreter. Tells us "how many steps does the average compiled
2170    // arg run per row" → narrows the attack target (subtree CSE vs
2171    // column-ref-push vs multi-spec combine). Read-only.
2172    crate::bump_counter!(STEP_VM_CALL_COUNT);
2173    crate::bump_counter!(STEP_VM_STEPS_TOTAL, steps.len() as u64);
2174    for step in steps {
2175        match step {
2176            Step::Column(pos) => {
2177                crate::bump_counter!(STEP_VM_COLUMN_FIRE);
2178                // v7.37.9 T3 S2 — catalog rows hold `Cow::Owned(String)`
2179                // for Text-class variants (per `spg-storage/src/lib.rs:539`
2180                // — "Persistent / catalog Values use Value<'static> with
2181                // Cow::Owned(...)"). Plain `.clone()` would therefore
2182                // still trigger `String::clone()` per cell read. Instead
2183                // manually wrap the existing storage into a borrowed Cow
2184                // pointing at the same bytes — zero-alloc push.
2185                let cell: Value<'val> = match row.get(*pos) {
2186                    Some(spg_storage::Value::Text(s)) => {
2187                        spg_storage::Value::Text(alloc::borrow::Cow::Borrowed(s.as_ref()))
2188                    }
2189                    Some(spg_storage::Value::Bytes(b)) => {
2190                        spg_storage::Value::Bytes(alloc::borrow::Cow::Borrowed(b.as_ref()))
2191                    }
2192                    Some(spg_storage::Value::Json(s)) => {
2193                        spg_storage::Value::Json(alloc::borrow::Cow::Borrowed(s.as_ref()))
2194                    }
2195                    Some(spg_storage::Value::Vector(v)) => {
2196                        spg_storage::Value::Vector(alloc::borrow::Cow::Borrowed(v.as_ref()))
2197                    }
2198                    // Copy-light variants: clone is free (just enum copy).
2199                    Some(v) => v.clone(),
2200                    None => Value::Null,
2201                };
2202                // Classification counter unchanged (still counts cells
2203                // that WERE heap-bearing in the baseline).
2204                if matches!(
2205                    &cell,
2206                    spg_storage::Value::Text(_)
2207                        | spg_storage::Value::Bytes(_)
2208                        | spg_storage::Value::Json(_)
2209                        | spg_storage::Value::Vector(_)
2210                ) {
2211                    crate::bump_counter!(STEP_VM_COLUMN_HEAP_ALLOC);
2212                }
2213                stack.push(cell);
2214            }
2215            Step::Lit(v) => {
2216                crate::bump_counter!(STEP_VM_LIT_FIRE);
2217                if matches!(
2218                    v,
2219                    spg_storage::Value::Text(_)
2220                        | spg_storage::Value::Bytes(_)
2221                        | spg_storage::Value::Json(_)
2222                        | spg_storage::Value::Vector(_)
2223                ) {
2224                    crate::bump_counter!(STEP_VM_LIT_HEAP_ALLOC);
2225                }
2226                // v7.37.9 T3 S3 — borrow literal storage instead of
2227                // String::clone'ing it. Step variants own their
2228                // literal (`Value<'static>` enum payload), so we can
2229                // safely construct a `Cow::Borrowed(&'static …)` view.
2230                // Same pattern as S2's Column path.
2231                let pushed: Value<'val> = match v {
2232                    spg_storage::Value::Text(s) => {
2233                        spg_storage::Value::Text(alloc::borrow::Cow::Borrowed(s.as_ref()))
2234                    }
2235                    spg_storage::Value::Bytes(b) => {
2236                        spg_storage::Value::Bytes(alloc::borrow::Cow::Borrowed(b.as_ref()))
2237                    }
2238                    spg_storage::Value::Json(s) => {
2239                        spg_storage::Value::Json(alloc::borrow::Cow::Borrowed(s.as_ref()))
2240                    }
2241                    spg_storage::Value::Vector(vec) => {
2242                        spg_storage::Value::Vector(alloc::borrow::Cow::Borrowed(vec.as_ref()))
2243                    }
2244                    other => other.clone(),
2245                };
2246                stack.push(pushed);
2247            }
2248            Step::Binary(op) => {
2249                crate::bump_counter!(STEP_VM_BINARY_FIRE);
2250                // v7.37.9 T3 S4 — try the by-ref fast path first
2251                // (comparison + 3VL ops). For those, operand bytes are
2252                // read but never stored in the result; we avoid the
2253                // .into_owned() that would clone every Cow::Borrowed
2254                // Text/Bytes/Json/Vector pushed by S2/S3. For ops that
2255                // build owned results (arithmetic, concat, json get,
2256                // etc.) apply_binary_by_ref returns None and we fall
2257                // through to the owning path.
2258                // v7.39 (round 346, M1) — the MySQL reading of AND / OR
2259                // has to be here TOO: a compiled predicate never passes
2260                // through `eval_expr`'s arm, so `WHERE a AND 1` still
2261                // errored on a MySQL session while the interpreted form
2262                // answered. (The pin found this, not the reading.)
2263                if ctx.mysql_dialect && matches!(op, BinOp::And | BinOp::Or) {
2264                    let r = super::as_mysql_truth(stack.pop().unwrap_or(Value::Null).into_owned())?;
2265                    let l = super::as_mysql_truth(stack.pop().unwrap_or(Value::Null).into_owned())?;
2266                    stack.push(apply_binary(*op, l, r)?);
2267                    continue;
2268                }
2269                let n = stack.len();
2270                if n >= 2 {
2271                    if let Some(result) =
2272                        super::apply_binary_by_ref(*op, &stack[n - 2], &stack[n - 1])?
2273                    {
2274                        stack.truncate(n - 2);
2275                        stack.push(result);
2276                        continue;
2277                    }
2278                }
2279                let r = stack.pop().unwrap_or(Value::Null).into_owned();
2280                let l = stack.pop().unwrap_or(Value::Null).into_owned();
2281                stack.push(apply_binary(*op, l, r)?);
2282            }
2283            Step::Connective { op, rhs } => {
2284                crate::bump_counter!(STEP_VM_BINARY_FIRE);
2285                let l = stack.pop().unwrap_or(Value::Null).into_owned();
2286                // The left decides, or it does not. A NULL decides nothing:
2287                // NULL AND false is false, so the right side is still needed.
2288                match (op, &l) {
2289                    (BinOp::And, Value::Bool(false)) => {
2290                        stack.push(Value::Bool(false));
2291                        continue;
2292                    }
2293                    (BinOp::Or, Value::Bool(true)) => {
2294                        stack.push(Value::Bool(true));
2295                        continue;
2296                    }
2297                    _ => {}
2298                }
2299                run_compiled_steps(rhs, row, ctx, stack)?;
2300                let r = stack.pop().unwrap_or(Value::Null).into_owned();
2301                stack.push(apply_binary(*op, l, r)?);
2302            }
2303            Step::BinaryCi(op, tc, collated) => {
2304                // v7.38.18 — fold and pad are two bits, resolved once at
2305                // compile time by `text_compare_of` and carried here.
2306                let fold_one = |v: Value<'static>| -> Value<'static> {
2307                    let (text, pad) = match v {
2308                        Value::BpChar(s) => (s.into_owned(), true),
2309                        Value::Text(s) => (s.into_owned(), tc.pads),
2310                        other => return other,
2311                    };
2312                    let base = if pad {
2313                        text.trim_end_matches(' ')
2314                    } else {
2315                        text.as_str()
2316                    };
2317                    Value::text(if tc.fold_case {
2318                        if ctx.mysql_dialect {
2319                            spg_storage::mysql_ci_fold(base)
2320                        } else {
2321                            base.to_ascii_lowercase()
2322                        }
2323                    } else {
2324                        alloc::string::ToString::to_string(base)
2325                    })
2326                };
2327                let r = fold_one(stack.pop().unwrap_or(Value::Null).into_owned());
2328                let l = fold_one(stack.pop().unwrap_or(Value::Null).into_owned());
2329                // v7.38.18 (S2) — order by the LOCALE when the
2330                // comparison has one. `apply_binary` compares text by
2331                // bytes, which is right for `C` and wrong for every
2332                // other collation; without this a scan filter answered
2333                // `x < 'b'` by bytes while `ORDER BY x` over the same
2334                // column answered by the locale.
2335                if let (Value::Text(a), Value::Text(b)) = (&l, &r)
2336                    && let Some(ord) = collated
2337                        .as_ref()
2338                        .map(|c| c.compare(a.as_ref(), b.as_ref()))
2339                        .or_else(|| tc.compare(a.as_ref(), b.as_ref()))
2340                {
2341                    let verdict = match op {
2342                        BinOp::Lt => ord == core::cmp::Ordering::Less,
2343                        BinOp::LtEq => ord != core::cmp::Ordering::Greater,
2344                        BinOp::Gt => ord == core::cmp::Ordering::Greater,
2345                        BinOp::GtEq => ord != core::cmp::Ordering::Less,
2346                        BinOp::Eq => ord == core::cmp::Ordering::Equal,
2347                        BinOp::NotEq => ord != core::cmp::Ordering::Equal,
2348                        _ => {
2349                            stack.push(apply_binary(*op, l, r)?);
2350                            continue;
2351                        }
2352                    };
2353                    stack.push(Value::Bool(verdict));
2354                    continue;
2355                }
2356                stack.push(apply_binary(*op, l, r)?);
2357            }
2358            Step::Unary(op) => {
2359                let v = stack.pop().unwrap_or(Value::Null).into_owned();
2360                if ctx.mysql_dialect
2361                    && matches!(op, UnOp::Not)
2362                    && !matches!(v, Value::Bool(_) | Value::Null)
2363                {
2364                    stack.push(Value::Bool(!super::predicate_is_true(&v, "NOT", true)?));
2365                    continue;
2366                }
2367                stack.push(apply_unary(*op, v)?);
2368            }
2369            Step::IsNull { negated } => {
2370                let v = stack.pop().unwrap_or(Value::Null);
2371                let is_null = matches!(v, Value::Null);
2372                stack.push(Value::Bool(if *negated { !is_null } else { is_null }));
2373            }
2374            Step::AnyTextMatch { negated } => {
2375                let v = stack.pop().unwrap_or(Value::Null);
2376                stack.push(match v {
2377                    Value::Null => Value::Null,
2378                    _ => Value::Bool(!*negated),
2379                });
2380            }
2381            Step::InSet {
2382                set,
2383                has_null,
2384                negated,
2385                fallback,
2386            } => {
2387                let needle = stack.pop().unwrap_or(Value::Null);
2388                match in_set_verdict(&needle, set, *has_null, *negated) {
2389                    Some(v) => stack.push(v),
2390                    // Cross-family needle: take the interpreter's
2391                    // exact coercion / error path on the whole node.
2392                    None => stack.push(eval_expr(fallback, &row.as_row(), ctx)?),
2393                }
2394            }
2395            Step::AnyAll { op, is_any, arr } => {
2396                let lhs = stack.pop().unwrap_or(Value::Null).into_owned();
2397                stack.push(crate::eval::any_all_over(lhs, arr.clone(), op, *is_any)?);
2398            }
2399            Step::Extract { field, fallback } => {
2400                let v = stack.pop().unwrap_or(Value::Null).into_owned();
2401                stack.push(crate::eval::extract_from_value(
2402                    field,
2403                    v,
2404                    source_of_extract(fallback),
2405                    ctx,
2406                )?);
2407            }
2408            Step::Regex { re, fallback } => {
2409                let v = stack.pop().unwrap_or(Value::Null);
2410                match regex_verdict(&v, re) {
2411                    Some(r) => stack.push(r?),
2412                    // Not text: the interpreter's coercion and wording, on
2413                    // the whole node, exactly as `Step::InSet` does.
2414                    None => stack.push(eval_expr(fallback, &row.as_row(), ctx)?),
2415                }
2416            }
2417            step @ (Step::Like { .. } | Step::LikeSubstring { .. }) => {
2418                // v7.39 (round 489) — one arm for both pattern steps,
2419                // sharing `like_verdict` with the fast predicate.
2420                //
2421                // The matching itself was already out of line: v7.37.16
2422                // borrowed the operand instead of paying `.into_owned()`
2423                // plus a per-row `Vec<char>` collect (~90 ns/row of
2424                // allocator traffic on a LIKE table scan), and round 484
2425                // replaced `str::find`'s two-way searcher — whose SETUP
2426                // was 14.6 % of self time, rebuilt every row for a
2427                // two-byte constant needle — with an ASCII byte scan.
2428                // ILIKE still lowercases; plain LIKE allocates nothing.
2429                let v = stack.pop().unwrap_or(Value::Null);
2430                match like_verdict(&v, step) {
2431                    Some(r) => stack.push(r?),
2432                    None => {
2433                        return Err(EvalError::TypeMismatch {
2434                            detail: format!(
2435                                "LIKE requires text operands, got {}",
2436                                crate::conversions::pg_type_name_for_error_opt(v.data_type())
2437                            ),
2438                        });
2439                    }
2440                }
2441            }
2442            Step::ColumnLength { pos } => {
2443                // v7.36 — zero-copy LENGTH on a column. Read the
2444                // cell by reference; compute char count without
2445                // cloning the underlying `String`. Saves 25 k ×
2446                // ~1 KB heap clones on the user_storage_usage shape.
2447                let v = row.get(*pos).unwrap_or(&Value::Null);
2448                let pushed = match v {
2449                    Value::Null => Value::Null,
2450                    Value::Text(s) => {
2451                        let n = if s.is_ascii() {
2452                            i32::try_from(s.len()).unwrap_or(i32::MAX)
2453                        } else {
2454                            i32::try_from(s.chars().count()).unwrap_or(i32::MAX)
2455                        };
2456                        Value::Int(n)
2457                    }
2458                    // v7.39 (bpchar epic) — length(bpchar) counts with the
2459                    // trailing blanks stripped (length('ab'::char(5)) = 2).
2460                    Value::BpChar(s) => {
2461                        let t = s.trim_end_matches(' ');
2462                        let n = if t.is_ascii() {
2463                            i32::try_from(t.len()).unwrap_or(i32::MAX)
2464                        } else {
2465                            i32::try_from(t.chars().count()).unwrap_or(i32::MAX)
2466                        };
2467                        Value::Int(n)
2468                    }
2469                    Value::Bytes(b) => Value::Int(i32::try_from(b.len()).unwrap_or(i32::MAX)),
2470                    other => {
2471                        return Err(EvalError::TypeMismatch {
2472                            detail: format!(
2473                                "length() needs text or bytea, got {}",
2474                                crate::conversions::pg_type_name_for_error_opt(other.data_type())
2475                            ),
2476                        });
2477                    }
2478                };
2479                stack.push(pushed);
2480            }
2481            Step::ColumnOctetLength { pos } => {
2482                let v = row.get(*pos).unwrap_or(&Value::Null);
2483                let pushed = match v {
2484                    Value::Null => Value::Null,
2485                    // v7.39 (bpchar epic) — octet_length(bpchar) counts the
2486                    // PADDED stored form.
2487                    Value::Text(s) | Value::BpChar(s) => {
2488                        Value::Int(i32::try_from(s.len()).unwrap_or(i32::MAX))
2489                    }
2490                    Value::Bytes(b) => Value::Int(i32::try_from(b.len()).unwrap_or(i32::MAX)),
2491                    other => {
2492                        return Err(EvalError::TypeMismatch {
2493                            detail: format!(
2494                                "octet_length() needs text or bytea, got {}",
2495                                crate::conversions::pg_type_name_for_error_opt(other.data_type())
2496                            ),
2497                        });
2498                    }
2499                };
2500                stack.push(pushed);
2501            }
2502            Step::Function { name_lower, n_args } => {
2503                crate::bump_counter!(STEP_VM_FUNCTION_FIRE);
2504                let start = stack.len().saturating_sub(*n_args);
2505                // `apply_function` borrows the trailing `n_args`
2506                // values off the stack; we then truncate + push the
2507                // result. `name_lower` is pre-lowercased at compile
2508                // time, so dispatch skips the per-row
2509                // `to_ascii_lowercase()` allocation.
2510                // v7.37.9 T3 S6 — apply_function_lower signature relaxed
2511                // to `&[Value<'_>]`; pass the borrowed stack slice
2512                // directly. Eliminates the Vec materialise + per-arg
2513                // String::clone that S1 introduced as a placeholder.
2514                let result =
2515                    super::functions::apply_function_lower(name_lower, &stack[start..], ctx)?;
2516                stack.truncate(start);
2517                stack.push(result);
2518            }
2519            Step::Coalesce { n_args } => {
2520                let start = stack.len().saturating_sub(*n_args);
2521                // The widening `COALESCE(1, 2.5)` needs only exists when the
2522                // non-null arguments carry MIXED types; inspected by ref, and
2523                // the mixed shapes fall to the owned arm that always did it.
2524                let mut mixed = false;
2525                let mut seen: Option<spg_storage::DataType> = None;
2526                for v in &stack[start..] {
2527                    if let Some(t) = v.data_type() {
2528                        match seen {
2529                            None => seen = Some(t),
2530                            Some(prev) if prev != t => {
2531                                mixed = true;
2532                                break;
2533                            }
2534                            Some(_) => {}
2535                        }
2536                    }
2537                }
2538                if mixed {
2539                    let result =
2540                        super::functions::apply_function_lower("coalesce", &stack[start..], ctx)?;
2541                    stack.truncate(start);
2542                    stack.push(result);
2543                } else {
2544                    let chosen = stack[start..]
2545                        .iter()
2546                        .position(|v| !matches!(v, Value::Null));
2547                    match chosen {
2548                        Some(k) => {
2549                            let v = stack.swap_remove(start + k);
2550                            stack.truncate(start);
2551                            stack.push(v);
2552                        }
2553                        None => {
2554                            stack.truncate(start);
2555                            stack.push(Value::Null);
2556                        }
2557                    }
2558                }
2559            }
2560            Step::Extremum { n_args, max } => {
2561                let start = stack.len().saturating_sub(*n_args);
2562                // Fast path: every non-NULL argument carries the SAME
2563                // concrete type — the comparison is the type's own and
2564                // the widen-to-common finish is the identity. Everything
2565                // else (mixed types, unknown-type text beside a typed
2566                // sibling, xid's refusal, MySQL's NULL-poisoning) falls
2567                // to the function arm unchanged.
2568                let mut uniform: Option<spg_storage::DataType> = None;
2569                let mut any_null = false;
2570                let mut fall_back = false;
2571                for v in &stack[start..] {
2572                    if matches!(v, Value::Null) {
2573                        any_null = true;
2574                        continue;
2575                    }
2576                    if matches!(v, Value::Xid(_)) {
2577                        fall_back = true;
2578                        break;
2579                    }
2580                    match (v.data_type(), uniform) {
2581                        (Some(t), None) => uniform = Some(t),
2582                        (Some(t), Some(prev)) if t != prev => {
2583                            fall_back = true;
2584                            break;
2585                        }
2586                        (Some(_), Some(_)) => {}
2587                        (None, _) => {
2588                            fall_back = true;
2589                            break;
2590                        }
2591                    }
2592                }
2593                if fall_back || (ctx.mysql_dialect && any_null) {
2594                    let name = if *max { "greatest" } else { "least" };
2595                    let result =
2596                        super::functions::apply_function_lower(name, &stack[start..], ctx)?;
2597                    stack.truncate(start);
2598                    stack.push(result);
2599                } else {
2600                    let mut best: Option<usize> = None;
2601                    for k in start..stack.len() {
2602                        if matches!(&stack[k], Value::Null) {
2603                            continue;
2604                        }
2605                        match best {
2606                            None => best = Some(k),
2607                            Some(b) => {
2608                                let ord = super::values::value_cmp_for_min_max(
2609                                    &stack[b],
2610                                    &stack[k],
2611                                    ctx.mysql_dialect,
2612                                );
2613                                let take = if *max {
2614                                    ord == core::cmp::Ordering::Less
2615                                } else {
2616                                    ord == core::cmp::Ordering::Greater
2617                                };
2618                                if take {
2619                                    best = Some(k);
2620                                }
2621                            }
2622                        }
2623                    }
2624                    match best {
2625                        Some(k) => {
2626                            let v = stack.swap_remove(k);
2627                            stack.truncate(start);
2628                            stack.push(v);
2629                        }
2630                        None => {
2631                            stack.truncate(start);
2632                            stack.push(Value::Null);
2633                        }
2634                    }
2635                }
2636            }
2637            Step::NullIf => {
2638                let n = stack.len();
2639                // NULLIF is `=` under the hood and keeps round 238's refusal
2640                // of incomparable operands; both reads are by reference.
2641                let verdict = match (&stack[n - 2], &stack[n - 1]) {
2642                    (Value::Null, _) => Some(true),
2643                    (_, Value::Null) => Some(false),
2644                    (a, b) => {
2645                        super::binop::require_comparable(spg_sql::ast::BinOp::Eq, a, b)?;
2646                        match super::apply_binary_by_ref(spg_sql::ast::BinOp::Eq, a, b)? {
2647                            Some(Value::Bool(eq)) => Some(eq),
2648                            _ => None,
2649                        }
2650                    }
2651                };
2652                match verdict {
2653                    Some(true) => {
2654                        stack.truncate(n - 2);
2655                        stack.push(Value::Null);
2656                    }
2657                    Some(false) => {
2658                        let a = stack.swap_remove(n - 2);
2659                        stack.truncate(n - 2);
2660                        stack.push(a);
2661                    }
2662                    // The by-ref compare could not decide — the owned arm can.
2663                    None => {
2664                        let result =
2665                            super::functions::apply_function_lower("nullif", &stack[n - 2..], ctx)?;
2666                        stack.truncate(n - 2);
2667                        stack.push(result);
2668                    }
2669                }
2670            }
2671            Step::Cast { target } => {
2672                crate::bump_counter!(STEP_VM_CAST_FIRE);
2673                // v7.39 (round 621) — two allocations a row lived on this one
2674                // line: `into_owned()` cloned a borrowed text cell just to
2675                // hand it to the cast, and `target.clone()` re-built the
2676                // target (a String, for the Named form) EVERY row even though
2677                // it is a compile product. `count(s::TEXT)` — a cast that
2678                // changes nothing — measured 2.00 allocs/row and 12 ms where
2679                // `count(s)` measures 0.00 and 2.8 ms.
2680                //
2681                // A cast that is an identity on the value it was given hands
2682                // the borrowed value straight back; everything else takes the
2683                // owned path, with the target passed by reference.
2684                let v = stack.pop().unwrap_or(Value::Null);
2685                if cast_is_identity_for(&v, target) {
2686                    stack.push(v);
2687                } else {
2688                    stack.push(super::cast::cast_value_ref_in(
2689                        v.into_owned(),
2690                        target,
2691                        ctx.mysql_dialect,
2692                    )?);
2693                }
2694            }
2695            Step::CastPlain { dt, name } => {
2696                let v = stack.pop().unwrap_or(Value::Null);
2697                // The name is pre-validated (it came off the plain table),
2698                // so NULL keeps its short-circuit; a same-type value passes
2699                // through untouched, exactly the identity the Cast step
2700                // recognises.
2701                let identity = matches!(
2702                    (&v, dt),
2703                    (Value::Null, _)
2704                        | (Value::Int(_), spg_storage::DataType::Int)
2705                        | (Value::BigInt(_), spg_storage::DataType::BigInt)
2706                        | (Value::SmallInt(_), spg_storage::DataType::SmallInt)
2707                        | (Value::Real(_), spg_storage::DataType::Real)
2708                        | (Value::Float(_), spg_storage::DataType::Float)
2709                        | (Value::Bool(_), spg_storage::DataType::Bool)
2710                        | (Value::Date(_), spg_storage::DataType::Date)
2711                        | (Value::Uuid(_), spg_storage::DataType::Uuid)
2712                );
2713                if identity {
2714                    stack.push(v);
2715                } else {
2716                    stack.push(super::cast::finish_named_cast_plain(
2717                        v.into_owned(),
2718                        *dt,
2719                        name,
2720                        ctx.mysql_dialect,
2721                    )?);
2722                }
2723            }
2724            Step::Case {
2725                fold_operand,
2726                operand,
2727                branches,
2728                else_branch,
2729            } => {
2730                crate::bump_counter!(STEP_VM_CASE_FIRE);
2731                // v7.37.5-A2b — short-circuit Case executor. Mirrors
2732                // `Expr::Case` interpreter semantics bit-for-bit (each
2733                // WHEN evaluates with its own scratch stack; first
2734                // match wins; ELSE = NULL when absent). The outer
2735                // `stack` is reused (truncated back to its pre-Case
2736                // mark after each sub-program); allocator-free per
2737                // branch — the prior version allocated a fresh
2738                // `Vec<Value>` per sub-program which showed up as
2739                // ~3 % `drop_in_place<Vec<Value>>` self time.
2740                let mark = stack.len();
2741                // v7.37.9 T3 S7 — Case sub-program lifetime threads
2742                // through naturally via S1's `'row: 'val`. Operand /
2743                // when / matched / else results are pushed by sub-progs
2744                // into our same stack; we pop them as `Value<'val>` and
2745                // keep them at that lifetime instead of forcing
2746                // into_owned. The simple-form operand match (Eq) uses
2747                // apply_binary_by_ref to avoid the operand clone +
2748                // pop-side into_owned the S1 placeholder was paying.
2749                let operand_value: Option<Value<'val>> = if let Some(op) = operand {
2750                    eval_compiled_ref_into(op, row, ctx, stack, mark)?;
2751                    Some(stack.pop().unwrap_or(Value::Null))
2752                } else {
2753                    None
2754                };
2755                stack.truncate(mark);
2756                let mut matched_value: Option<Value<'val>> = None;
2757                for (when_c, then_c) in branches {
2758                    eval_compiled_ref_into(when_c, row, ctx, stack, mark)?;
2759                    let when_v = stack.pop().unwrap_or(Value::Null);
2760                    stack.truncate(mark);
2761                    let matched = match &operand_value {
2762                        None => matches!(when_v, Value::Bool(true)),
2763                        Some(op_v) => {
2764                            // Try the by-ref comparison fast path; fall
2765                            // back to owning apply_binary only if the
2766                            // by-ref path returns None (non-comparison
2767                            // op, which Eq never is).
2768                            // v7.38.14 — fold first when the compile-time
2769                            // decision says this arm's collation folds.
2770                            // `apply_binary_by_ref` compares bytes and knows
2771                            // nothing about a dialect, which is why
2772                            // `CASE s WHEN 'A'` missed 'a' on a
2773                            // case-insensitive column.
2774                            let folded = if *fold_operand {
2775                                match (op_v, &when_v) {
2776                                    // v7.38.18 — each side on its OWN type;
2777                                    // see `mysql_fold_value`. The pair match
2778                                    // missed a CHAR against a literal.
2779                                    (x, y)
2780                                        if spg_storage::mysql_fold_value(x).is_some()
2781                                            && spg_storage::mysql_fold_value(y).is_some() =>
2782                                    {
2783                                        Some((
2784                                            Value::text(spg_storage::mysql_fold_value(x).unwrap()),
2785                                            Value::text(spg_storage::mysql_fold_value(y).unwrap()),
2786                                        ))
2787                                    }
2788                                    _ => None,
2789                                }
2790                            } else {
2791                                None
2792                            };
2793                            let eq_result = match folded {
2794                                Some((l, r)) => apply_binary(BinOp::Eq, l, r)?,
2795                                None => match super::apply_binary_by_ref(BinOp::Eq, op_v, &when_v)?
2796                                {
2797                                    Some(v) => v,
2798                                    None => apply_binary(
2799                                        BinOp::Eq,
2800                                        op_v.clone().into_owned(),
2801                                        when_v.clone().into_owned(),
2802                                    )?,
2803                                },
2804                            };
2805                            matches!(eq_result, Value::Bool(true))
2806                        }
2807                    };
2808                    if matched {
2809                        eval_compiled_ref_into(then_c, row, ctx, stack, mark)?;
2810                        matched_value = Some(stack.pop().unwrap_or(Value::Null));
2811                        stack.truncate(mark);
2812                        break;
2813                    }
2814                }
2815                let v: Value<'val> = match matched_value {
2816                    Some(v) => v,
2817                    None => match else_branch {
2818                        Some(el) => {
2819                            eval_compiled_ref_into(el, row, ctx, stack, mark)?;
2820                            let v = stack.pop().unwrap_or(Value::Null);
2821                            stack.truncate(mark);
2822                            v
2823                        }
2824                        None => Value::Null,
2825                    },
2826                };
2827                stack.push(v);
2828            }
2829            Step::CoerceCommon(target) => {
2830                let v = stack.pop().unwrap_or(Value::Null).into_owned();
2831                stack.push(super::widen_value_to(v, *target));
2832            }
2833            Step::Subtree(e) => stack.push(eval_expr(e, &row.as_row(), ctx)?),
2834        }
2835    }
2836    Ok(())
2837}
2838
2839/// v7.37.9 Phase 1A-ext-2 T1 — Step VM internal step-type counters.
2840/// Read-only diagnostic; gates no behaviour. Used by counter_dump.rs
2841/// to ground-truth subtree CSE / column-ref-push / multi-spec-combine
2842/// attack ROI estimates.
2843pub static STEP_VM_CALL_COUNT: core::sync::atomic::AtomicU64 =
2844    core::sync::atomic::AtomicU64::new(0);
2845pub static STEP_VM_STEPS_TOTAL: core::sync::atomic::AtomicU64 =
2846    core::sync::atomic::AtomicU64::new(0);
2847pub static STEP_VM_COLUMN_FIRE: core::sync::atomic::AtomicU64 =
2848    core::sync::atomic::AtomicU64::new(0);
2849pub static STEP_VM_LIT_FIRE: core::sync::atomic::AtomicU64 = core::sync::atomic::AtomicU64::new(0);
2850pub static STEP_VM_BINARY_FIRE: core::sync::atomic::AtomicU64 =
2851    core::sync::atomic::AtomicU64::new(0);
2852pub static STEP_VM_FUNCTION_FIRE: core::sync::atomic::AtomicU64 =
2853    core::sync::atomic::AtomicU64::new(0);
2854pub static STEP_VM_CAST_FIRE: core::sync::atomic::AtomicU64 = core::sync::atomic::AtomicU64::new(0);
2855pub static STEP_VM_CASE_FIRE: core::sync::atomic::AtomicU64 = core::sync::atomic::AtomicU64::new(0);
2856
2857/// v7.37.9 Round 3 — heap-alloc counters specifically for the T3
2858/// structural attack's ROI estimate. Step::Column / Step::Lit hits
2859/// pay a String alloc when the cell variant is heap-bearing
2860/// (Text/Bytes/Json/Vector). T3 stack-lifetime push-by-borrow
2861/// would eliminate these for the bulk of per-row work.
2862pub static STEP_VM_COLUMN_HEAP_ALLOC: core::sync::atomic::AtomicU64 =
2863    core::sync::atomic::AtomicU64::new(0);
2864pub static STEP_VM_LIT_HEAP_ALLOC: core::sync::atomic::AtomicU64 =
2865    core::sync::atomic::AtomicU64::new(0);
2866
2867/// v7.39 (round 481) — how many values the stack still holds when a call
2868/// finishes, and how many are heap-bearing.
2869///
2870/// Round 480 left `drop_glue<Value>` at 16 % of self time with the drops
2871/// attributed to the predicate closure, i.e. to the stack rather than to
2872/// the returned value (round 479 removed that one). Whether the ops leave
2873/// operands behind for the next call's `clear()` to drop is a question
2874/// with a number, so this counts it rather than reasoning about it — the
2875/// previous round was spent acting on an inference that turned out to name
2876/// an unreachable branch.
2877/// v7.39 (round 482) — how often the `<column> <cmp> <literal>` fast
2878/// predicate fires, so "is it even reached" is a number and not a guess
2879/// (round 480 was spent on a branch that turned out to be unreachable).
2880pub static STEP_VM_FASTPRED_FIRE: core::sync::atomic::AtomicU64 =
2881    core::sync::atomic::AtomicU64::new(0);
2882
2883/// r1021 — how often the integer lane answers, and how often a row makes it
2884/// hand back. Round 480 was spent on a branch that turned out unreachable,
2885/// so "is it even reached" stays a number here too — and the fallback
2886/// counter is the one that matters for correctness review: it is every row
2887/// the lane declined to decide.
2888pub static STEP_VM_INTLANE_FIRE: core::sync::atomic::AtomicU64 =
2889    core::sync::atomic::AtomicU64::new(0);
2890pub static STEP_VM_INTLANE_FALLBACK: core::sync::atomic::AtomicU64 =
2891    core::sync::atomic::AtomicU64::new(0);
2892
2893pub static STEP_VM_STACK_LEFTOVER: core::sync::atomic::AtomicU64 =
2894    core::sync::atomic::AtomicU64::new(0);
2895pub static STEP_VM_STACK_LEFTOVER_HEAP: core::sync::atomic::AtomicU64 =
2896    core::sync::atomic::AtomicU64::new(0);
2897
2898#[cfg(test)]
2899mod like_substring_tests {
2900    use super::{like_substring_match, like_substring_shape};
2901
2902    fn shape(p: &str) -> Option<(usize, alloc::string::String, usize)> {
2903        let chars: alloc::vec::Vec<char> = p.chars().collect();
2904        like_substring_shape(&chars)
2905    }
2906
2907    #[test]
2908    fn shape_recognition() {
2909        assert_eq!(shape("%_05%"), Some((1, "05".into(), 0)));
2910        assert_eq!(shape("%abc%"), Some((0, "abc".into(), 0)));
2911        assert_eq!(shape("%ab_%"), Some((0, "ab".into(), 1)));
2912        assert_eq!(shape("%%x%%"), Some((0, "x".into(), 0)));
2913        assert_eq!(shape("%__a__%"), Some((2, "a".into(), 2)));
2914        // Not eligible: missing anchors, inner %, escapes, empty literal.
2915        assert_eq!(shape("ab%"), None);
2916        assert_eq!(shape("%ab"), None);
2917        assert_eq!(shape("%a%b%"), None);
2918        assert_eq!(shape("%___%"), None);
2919        assert_eq!(shape("%a\\%b%"), None);
2920        assert_eq!(shape("%"), None);
2921    }
2922
2923    #[test]
2924    fn matcher_semantics() {
2925        // %_05% — needs one char before "05".
2926        assert!(like_substring_match("x05", "05", 1, 0));
2927        assert!(!like_substring_match("05", "05", 1, 0));
2928        assert!(like_substring_match("ab05cd", "05", 1, 0));
2929        // Overlapping / repeated hits: first hit fails the k-check,
2930        // a later one passes.
2931        assert!(like_substring_match("05x05", "05", 1, 0));
2932        // Trailing underscore needs one char after.
2933        assert!(like_substring_match("abz", "ab", 0, 1));
2934        assert!(!like_substring_match("ab", "ab", 0, 1));
2935        // Plain substring.
2936        assert!(like_substring_match("hello", "ell", 0, 0));
2937        assert!(!like_substring_match("hello", "xyz", 0, 0));
2938        // Multi-byte chars count as single wildcard chars.
2939        assert!(like_substring_match("é05", "05", 1, 0));
2940        assert!(!like_substring_match("é5", "05", 1, 0));
2941    }
2942}