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