Skip to main content

spg_engine/
join.rs

1//! Join execution — the deferred-join row sources (JoinSrc), the
2//! row-index-tuple view handed to the aggregate engine (RowRef), the
3//! per-stage peer descriptor (JoinedPeer), the deferred output
4//! (DeferredJoin), the bounded top-N sink (TopNEntry), the tuple<->Row
5//! helpers, and the `Engine` join planner methods that build them
6//! (`build_joined_filtered_rows`, the LATERAL probe/materialise pair,
7//! and the streamed inner-join top-N path). Split out of `lib.rs`
8//! (v7.32 engine modularisation).
9
10use alloc::borrow::Cow;
11use alloc::string::{String, ToString};
12use alloc::vec::Vec;
13
14use spg_sql::ast::{Expr, FromClause, JoinKind, SelectItem, SelectStatement, TableRef};
15use spg_storage::{ColumnSchema, DataType, Row, Table, Value};
16
17use crate::eval::EvalContext;
18use crate::{
19    ByteBudget, CancelToken, Engine, EngineError, OrderKey, QueryResult, aggregate,
20    apply_offset_and_limit, approx_row_bytes, approx_rows_bytes, approx_value_bytes,
21    build_order_keys, build_projection, cmp_multi_key, collect_column_qualifiers,
22    collect_qualified_refs, eval, expr_has_subquery, memoize, reorder, value_cmp,
23    value_to_literal_expr,
24};
25
26/// v7.17.0 Phase 3.P0-41 — LATERAL peer descriptor. Either eagerly
27/// materialised (every regular table / unnest / generate_series) or
28/// lateral (subquery re-evaluated per outer row).
29pub(crate) struct JoinedPeer<'a> {
30    pub(crate) eager_rows: Option<Vec<Row<'static>>>,
31    pub(crate) cols: Vec<ColumnSchema>,
32    pub(crate) alias: String,
33    pub(crate) kind: JoinKind,
34    pub(crate) on: Option<&'a Expr>,
35    pub(crate) lateral: Option<&'a SelectStatement>,
36    /// v7.28 (round-22) — plain-table name for the index-nested-loop
37    /// path. None for unnest/lateral.
38    pub(crate) join_table: Option<String>,
39    /// v7.33 (mailrs 7.33.0) — WHERE conjuncts pushed onto this (INNER)
40    /// peer that were NOT applied by eager materialisation. A deferred
41    /// plain peer carries them here so the join stages apply them as a
42    /// residual filter on matched (left,right) pairs — keeping the
43    /// index-nested-loop path (seek driver + look up only matched peer
44    /// rows) instead of eagerly scanning the whole peer table to filter
45    /// it. Empty for eager peers (already filtered) and LEFT peers
46    /// (analyze_join_pushdown only pushes onto INNER peers).
47    pub(crate) where_preds: Vec<Expr>,
48}
49
50/// v7.31 (perf campaign) — deferred-join row source: one per join
51/// stage. The working set advances as row-index tuples instead of
52/// cloned combined rows; each tuple slot indexes into one of these.
53pub(crate) enum JoinSrc<'a> {
54    /// Owned by the join: the primary scan, a lazily-materialised
55    /// peer, or the arena of per-outer-row LATERAL results.
56    Owned(Vec<Row<'static>>),
57    /// Peer rows materialised up front and still owned by `JoinedPeer`.
58    Eager(&'a [Row<'static>]),
59    /// Index-nested-loop peer reading the stored table in place.
60    Stored(&'a spg_storage::persistent::PersistentVec<Row<'static>>),
61    /// v7.36 — hot tier borrowed in place + cold tier owned. INL
62    /// probe consults `cold_locator_map` to translate a Cold
63    /// `RowLocator` (which only carries `(segment_id, page_offset)`
64    /// — that pair identifies the PAGE, not the row, so multiple
65    /// rows on one page collide) into a per-row offset via the
66    /// PK key (`IndexKey::Int(i64)`) instead. The cold-tier
67    /// architecture already requires an integer PK, so this is
68    /// the unique-per-row identifier the segment lookup already
69    /// uses internally. Indices `0..hot.len()` map to hot rows;
70    /// `hot.len()..` map to `cold[i - hot.len()]`.
71    Mixed {
72        hot: &'a spg_storage::persistent::PersistentVec<Row<'static>>,
73        cold: Vec<Row<'static>>,
74        cold_locator_map: hashbrown::HashMap<i64, usize>,
75    },
76}
77
78/// v7.39 (round 576) — a hash-join bucket that does not allocate for the
79/// row it usually holds.
80///
81/// The build side of an FK-to-PK join is unique, so nearly every bucket
82/// holds exactly one row — and each one was its own `Vec`. A counting
83/// allocator put the cost at ONE allocation per peer row: a 200k-row
84/// self-join made 200,127 allocations where a single-table scan of the
85/// same table makes 47, and it made them whether the query wanted
86/// 200,000 rows or 100. That is the allocator's 28% round 575 could not
87/// name, and the reason the join does not get cheaper when a predicate
88/// narrows it.
89///
90/// The first row lives inline; a second one promotes to a `Vec`.
91#[derive(Debug)]
92enum Bucket {
93    One(usize),
94    Many(Vec<usize>),
95}
96
97impl Bucket {
98    fn push(&mut self, ri: usize) {
99        match self {
100            Self::One(first) => *self = Self::Many(alloc::vec![*first, ri]),
101            Self::Many(v) => v.push(ri),
102        }
103    }
104
105    fn as_slice(&self) -> &[usize] {
106        match self {
107            Self::One(x) => core::slice::from_ref(x),
108            Self::Many(v) => v.as_slice(),
109        }
110    }
111}
112
113impl JoinSrc<'_> {
114    pub(crate) fn get(&self, i: usize) -> Option<&Row<'static>> {
115        match self {
116            Self::Owned(v) => v.get(i),
117            Self::Eager(s) => s.get(i),
118            Self::Stored(p) => p.get(i),
119            Self::Mixed { hot, cold, .. } => {
120                if i < hot.len() {
121                    hot.get(i)
122                } else {
123                    cold.get(i - hot.len())
124                }
125            }
126        }
127    }
128
129    pub(crate) fn len(&self) -> usize {
130        match self {
131            Self::Owned(v) => v.len(),
132            Self::Eager(s) => s.len(),
133            Self::Stored(p) => p.len(),
134            Self::Mixed { hot, cold, .. } => hot.len() + cold.len(),
135        }
136    }
137
138    /// v7.36 — translate a PK key (`i64` — the cold tier's
139    /// integer-only PK contract) into the corresponding row index
140    /// inside this `Mixed` source. Returns `None` for non-Mixed
141    /// sources or when the key has no cold-tier row registered.
142    pub(crate) fn cold_pk_offset(&self, pk_key: i64) -> Option<usize> {
143        match self {
144            Self::Mixed {
145                hot,
146                cold_locator_map,
147                ..
148            } => cold_locator_map
149                .get(&pk_key)
150                .copied()
151                .map(|off| hot.len() + off),
152            _ => None,
153        }
154    }
155}
156
157/// Resolve one combined-schema position against a row-index tuple.
158/// `offsets` holds the prefix column offsets of the consumed sources
159/// (`offsets.len() == tuple.len() + 1`). `None` means SQL NULL: a
160/// LEFT-extended slot (`usize::MAX`), or a position past the row's
161/// width.
162///
163/// v7.37.43 (DISTA A-2) — slow-path fallback used only when the caller
164/// has no `pos_to_src` table. Most RowRef::Tuple uses now go through
165/// `tuple_value_indexed` (direct index lookup, no partition_point).
166pub(crate) fn tuple_value<'s>(
167    sources: &'s [JoinSrc<'_>],
168    offsets: &[usize],
169    tuple: &[usize],
170    pos: usize,
171) -> Option<&'s Value<'static>> {
172    let k = offsets.partition_point(|&o| o <= pos).checked_sub(1)?;
173    let ri = *tuple.get(k)?;
174    if ri == usize::MAX {
175        return None;
176    }
177    sources.get(k)?.get(ri)?.values.get(pos - offsets[k])
178}
179
180/// v7.37.43 (DISTA A-2) — direct-index variant: `pos_to_src[pos]` =
181/// source index `k` for combined position `pos`. Built once per
182/// JoinPipeline / DeferredJoin (linear in combined width); per-row
183/// `RowRef::get` becomes a single array read instead of a binary
184/// search over `offsets` per call.
185///
186/// For DISTA (~100k joined rows × ~5 cell reads/row in the aggregate
187/// loop) this strips ~50 ns × 500k = ~25 ms of partition_point ops down
188/// to direct indexing.
189#[inline]
190pub(crate) fn tuple_value_indexed<'s>(
191    sources: &'s [JoinSrc<'_>],
192    offsets: &[usize],
193    pos_to_src: &[u16],
194    tuple: &[usize],
195    pos: usize,
196) -> Option<&'s Value<'static>> {
197    let k = *pos_to_src.get(pos)? as usize;
198    let ri = *tuple.get(k)?;
199    if ri == usize::MAX {
200        return None;
201    }
202    sources.get(k)?.get(ri)?.values.get(pos - offsets[k])
203}
204
205/// v7.37.43 (DISTA A-2) — build the position → source-index table for
206/// a combined schema. `offsets.len() == sources + 1`, last entry is
207/// the total combined width.
208pub(crate) fn build_pos_to_src(offsets: &[usize]) -> Vec<u16> {
209    let width = offsets.last().copied().unwrap_or(0);
210    let mut tab: Vec<u16> = Vec::with_capacity(width);
211    for k in 0..offsets.len().saturating_sub(1) {
212        let span = offsets[k + 1] - offsets[k];
213        for _ in 0..span {
214            // 2^16 sources is comfortably beyond the planner cap;
215            // `as u16` truncation here is a non-issue in practice.
216            tab.push(k as u16);
217        }
218    }
219    tab
220}
221
222/// v7.39 (round 656) — what the aggregate engine reads its input through.
223///
224/// The scan path used to hand `aggregate::run` a `Vec<RowRef>` built by
225/// `filtered.iter().map(RowRef::Owned).collect()` — one 64-byte enum per
226/// row to wrap an 8-byte reference. Measured, that is the whole of a
227/// scalar aggregate's working memory and it is O(rows): 7.0 MB at 100k
228/// rows, 19.8 at 250k, 40.4 at 500k, 79.6 at 1M — ~81 bytes a row for a
229/// query that returns one number. At 50M rows it is 3.2 GB, and what the
230/// customer meets is not slowness, it is OOM.
231///
232/// `RowRef` is 64 bytes because its `Tuple` variant carries four slice
233/// references for the join path; a single-table scan only ever uses
234/// `Owned`. So the scan now passes its `&[Row]` straight through and the
235/// `RowRef` is built per row on the stack, where it costs nothing. The
236/// join path keeps handing over its `&[RowRef]` exactly as before.
237#[derive(Clone, Copy)]
238pub(crate) enum AggRows<'a> {
239    /// A single-table scan's rows, borrowed. No per-row allocation.
240    Owned(&'a [Row<'static>]),
241    /// The join path's deferred tuples, already built.
242    Refs(&'a [RowRef<'a>]),
243    /// A single-table scan whose survivors are already a list of row
244    /// POINTERS (`Vec<&Row>` after WHERE). This is the shape the plain
245    /// relational scan has, and it is where the measured cost lived: it
246    /// used to `collect()` those pointers into a second vector of
247    /// 64-byte `RowRef`s, 8 bytes of data wrapped in 64.
248    Ptrs(&'a [&'a Row<'static>]),
249}
250
251impl<'a> AggRows<'a> {
252    #[inline]
253    pub(crate) fn len(&self) -> usize {
254        match self {
255            Self::Owned(r) => r.len(),
256            Self::Refs(r) => r.len(),
257            Self::Ptrs(r) => r.len(),
258        }
259    }
260
261    #[inline]
262    pub(crate) fn is_empty(&self) -> bool {
263        self.len() == 0
264    }
265
266    #[inline]
267    pub(crate) fn get(&self, i: usize) -> Option<RowRef<'a>> {
268        match self {
269            Self::Owned(r) => r.get(i).map(RowRef::Owned),
270            Self::Refs(r) => r.get(i).copied(),
271            Self::Ptrs(r) => r.get(i).map(|p| RowRef::Owned(p)),
272        }
273    }
274
275    /// The sub-range the parallel shards walk. Slicing is free on both
276    /// arms — no copy, no allocation.
277    #[inline]
278    pub(crate) fn range(&self, lo: usize, hi: usize) -> Self {
279        match self {
280            Self::Owned(r) => Self::Owned(&r[lo..hi]),
281            Self::Refs(r) => Self::Refs(&r[lo..hi]),
282            Self::Ptrs(r) => Self::Ptrs(&r[lo..hi]),
283        }
284    }
285
286    #[inline]
287    pub(crate) fn first(&self) -> Option<RowRef<'a>> {
288        self.get(0)
289    }
290
291    #[inline]
292    pub(crate) fn iter(&self) -> impl Iterator<Item = RowRef<'a>> + '_ {
293        (0..self.len()).filter_map(move |i| self.get(i))
294    }
295}
296
297/// v7.32 (P4 borrow channel, increment 2) — a row handed to the
298/// aggregate engine. Either a borrowed materialised `Row` (single-table
299/// and legacy paths) or a deferred row-index tuple over join sources
300/// (the join+aggregate path) that resolves cells *by reference* via
301/// `tuple_value`, so the join+aggregate path never materialises a
302/// combined `Row` for the bound-column fast path.
303#[derive(Clone, Copy)]
304pub(crate) enum RowRef<'a> {
305    Owned(&'a Row<'static>),
306    Tuple {
307        sources: &'a [JoinSrc<'a>],
308        offsets: &'a [usize],
309        /// v7.37.43 (DISTA A-2) — precomputed combined-position →
310        /// source-index map (built once per JoinPipeline / DeferredJoin
311        /// in `build_pos_to_src`). `RowRef::get` uses this for direct
312        /// indexing instead of binary search over `offsets` per call.
313        pos_to_src: &'a [u16],
314        tuple: &'a [usize],
315    },
316}
317
318impl<'a> RowRef<'a> {
319    /// Borrow the cell at a combined-schema position. The bound-column
320    /// fast path in `aggregate::run` reads cells this way — zero clone.
321    ///
322    /// v7.39 (round 656) — the returned reference borrows the ROW DATA
323    /// (`'a`), not `&self`. Both variants only ever hand back something
324    /// that lives in the `'a` slices, and saying so is what lets the
325    /// aggregate loop hold a `RowRef` by value: a per-iteration local can
326    /// then still yield references that outlive it, which is exactly what
327    /// the group-key `Vec<&Value>` needs.
328    #[inline]
329    pub(crate) fn get(&self, pos: usize) -> Option<&'a Value<'a>> {
330        match self {
331            RowRef::Owned(r) => r.values.get(pos),
332            RowRef::Tuple {
333                sources,
334                offsets,
335                pos_to_src,
336                tuple,
337            } => tuple_value_indexed(sources, offsets, pos_to_src, tuple, pos),
338        }
339    }
340
341    /// Present the row as a `&Row<'static>` for the eval path. `Owned` borrows
342    /// directly (zero cost); `Tuple` materialises once into owned values
343    /// — the only allocation, paid solely on the eval (non-bound) path,
344    /// never for the bound fast path. The materialised width is the full
345    /// combined schema (`offsets.last()`); a LEFT-NULL slot or an out-of-
346    /// range position becomes `Value::Null` (same as `tuple_value`).
347    pub(crate) fn as_row(&self) -> Cow<'_, Row<'static>> {
348        match self {
349            RowRef::Owned(r) => Cow::Borrowed(r),
350            RowRef::Tuple {
351                sources,
352                offsets,
353                pos_to_src,
354                tuple,
355            } => {
356                let width = offsets.last().copied().unwrap_or(0);
357                let mut vals: Vec<Value<'static>> = Vec::with_capacity(width);
358                for pos in 0..width {
359                    vals.push(
360                        tuple_value_indexed(sources, offsets, pos_to_src, tuple, pos)
361                            .cloned()
362                            .unwrap_or(Value::Null),
363                    );
364                }
365                Cow::Owned(Row::new(vals))
366            }
367        }
368    }
369
370    /// v7.37.5-A2b (profile-guided Track A) — same as `as_row` but
371    /// writes into a caller-owned buffer that survives the row loop.
372    /// Profile showed `as_row` allocating + freeing a fresh
373    /// `Vec<Value>` of full combined width per outer row in
374    /// `accumulate_groups`'s `needs_mat` path (~15 % self time in
375    /// `to_vec`/`Value::clone`/`drop` combined, ~10-20 MB per-query
376    /// allocator churn at 24 k × 30-cell width). Reusing the buffer
377    /// keeps the Vec backing across iterations — Value clones still
378    /// fire (they're semantically owned) but the Vec allocation +
379    /// free goes away. `Owned` rows clone into the buffer too so the
380    /// caller can pass a single buffer through both branches; the
381    /// Vec stays warm in the allocator across all calls.
382    pub(crate) fn as_row_into(&self, buf: &mut Vec<Value<'static>>) {
383        buf.clear();
384        match self {
385            RowRef::Owned(r) => {
386                buf.reserve(r.values.len());
387                for v in &r.values {
388                    buf.push(v.clone());
389                }
390            }
391            RowRef::Tuple {
392                sources,
393                offsets,
394                pos_to_src,
395                tuple,
396            } => {
397                let width = offsets.last().copied().unwrap_or(0);
398                buf.reserve(width);
399                for pos in 0..width {
400                    buf.push(
401                        tuple_value_indexed(sources, offsets, pos_to_src, tuple, pos)
402                            .cloned()
403                            .unwrap_or(Value::Null),
404                    );
405                }
406            }
407        }
408    }
409}
410
411/// Clone a source row's values into a combined-row buffer. A mask
412/// (per-column "is referenced anywhere in the statement") NULLs the
413/// unreferenced columns instead of cloning them — the in-place
414/// equivalent of `null_out_unreferenced` for sources that were never
415/// pre-cloned.
416pub(crate) fn extend_masked(
417    vals: &mut Vec<Value<'static>>,
418    row: &Row<'static>,
419    mask: Option<&[bool]>,
420) {
421    match mask {
422        Some(keep) => {
423            for (i, v) in row.values.iter().enumerate() {
424                if keep.get(i).copied().unwrap_or(false) {
425                    vals.push(v.clone());
426                } else {
427                    vals.push(Value::Null);
428                }
429            }
430        }
431        None => vals.extend(row.values.iter().cloned()),
432    }
433}
434
435/// Materialise a row-index tuple into owned values, NULL-padding
436/// LEFT-extended slots to the source's schema width.
437pub(crate) fn materialise_tuple_vals(
438    sources: &[JoinSrc<'_>],
439    widths: &[usize],
440    masks: &[Option<Vec<bool>>],
441    tuple: &[usize],
442    cap: usize,
443) -> Vec<Value<'static>> {
444    let mut vals: Vec<Value<'static>> = Vec::with_capacity(cap);
445    for (k, &ri) in tuple.iter().enumerate() {
446        let row = if ri == usize::MAX {
447            None
448        } else {
449            sources[k].get(ri)
450        };
451        match row {
452            Some(r) => extend_masked(&mut vals, r, masks[k].as_deref()),
453            None => {
454                for _ in 0..widths[k] {
455                    vals.push(Value::Null);
456                }
457            }
458        }
459    }
460    vals
461}
462
463/// v7.32 (P4 borrow channel, increment 2) — the deferred output of
464/// `build_joined_filtered_rows`: WHERE-surviving rows held as row-index
465/// tuples over the join sources, NOT materialised into combined Rows.
466/// The aggregate path borrows each survivor as a `RowRef::Tuple` (the
467/// bound fast path reads source cells by reference — zero clone); the
468/// projection / window paths call `materialise()` for an owned
469/// `Vec<Row<'static>>` identical to the pre-increment-2 output.
470pub(crate) struct DeferredJoin<'a> {
471    pub(crate) sources: Vec<JoinSrc<'a>>,
472    pub(crate) offsets: Vec<usize>,
473    /// v7.37.43 (DISTA A-2) — combined-position → source-index map; built
474    /// once via `build_pos_to_src(&offsets)` at construction time, so
475    /// per-row `RowRef::get` is a direct index instead of a partition_point
476    /// over `offsets`.
477    pub(crate) pos_to_src: Vec<u16>,
478    pub(crate) widths: Vec<usize>,
479    pub(crate) masks: Vec<Option<Vec<bool>>>,
480    /// Flat row-index tuples — one stride-long group per surviving row.
481    pub(crate) survivors: Vec<usize>,
482    pub(crate) stride: usize,
483    pub(crate) combined_schema: Vec<ColumnSchema>,
484}
485
486impl DeferredJoin<'_> {
487    pub(crate) fn len(&self) -> usize {
488        if self.stride == 0 {
489            0
490        } else {
491            self.survivors.len() / self.stride
492        }
493    }
494
495    /// Borrow each surviving tuple as a `RowRef::Tuple` for the
496    /// aggregate engine — no combined Row is materialised.
497    pub(crate) fn row_refs(&self) -> Vec<RowRef<'_>> {
498        if self.stride == 0 {
499            return Vec::new();
500        }
501        self.survivors
502            .chunks(self.stride)
503            .map(|tuple| RowRef::Tuple {
504                sources: &self.sources,
505                offsets: &self.offsets,
506                pos_to_src: &self.pos_to_src,
507                tuple,
508            })
509            .collect()
510    }
511
512    /// Materialise the survivors into owned combined Rows (projection /
513    /// window paths). Byte-identical to the pre-deferral output.
514    pub(crate) fn materialise(&self) -> Vec<Row<'static>> {
515        if self.stride == 0 {
516            return Vec::new();
517        }
518        let cap = self.offsets.last().copied().unwrap_or(0);
519        self.survivors
520            .chunks(self.stride)
521            .map(|tuple| {
522                Row::new(materialise_tuple_vals(
523                    &self.sources,
524                    &self.widths,
525                    &self.masks,
526                    tuple,
527                    cap,
528                ))
529            })
530            .collect()
531    }
532}
533
534/// v7.32 (P4 borrow channel, increment 2) — byte estimate of a
535/// row-index tuple WITHOUT materialising it: walk each referenced source
536/// cell by reference and sum, applying the same per-column mask
537/// `materialise_tuple_vals` would (unreferenced columns count as NULL).
538/// Mirrors `approx_row_bytes(materialised)` so the v7.30.3 byte budget
539/// meters identical live bytes on the deferred path.
540pub(crate) fn approx_tuple_bytes(
541    sources: &[JoinSrc<'_>],
542    offsets: &[usize],
543    masks: &[Option<Vec<bool>>],
544    tuple: &[usize],
545) -> usize {
546    let width = offsets.last().copied().unwrap_or(0);
547    let mut bytes = width * core::mem::size_of::<Value>();
548    for (k, &ri) in tuple.iter().enumerate() {
549        if ri == usize::MAX {
550            continue;
551        }
552        let Some(row) = sources.get(k).and_then(|s| s.get(ri)) else {
553            continue;
554        };
555        let mask = masks.get(k).and_then(|m| m.as_deref());
556        for (i, v) in row.values.iter().enumerate() {
557            let kept = mask.map_or(true, |m| m.get(i).copied().unwrap_or(false));
558            if kept {
559                bytes += approx_value_bytes(v);
560            }
561        }
562    }
563    bytes
564}
565
566/// v7.30.3 (mailrs round-26) — bounded top-N sink entry for the
567/// streamed single-join path. `keys` are the `OrderKey`s
568/// `build_order_keys` emits; `descs` (shared across all entries via
569/// `Rc`) drives the per-key reverse so ordering matches the general
570/// path's `cmp_multi_key` exactly (including the ±INF NULL placements
571/// and full-precision text keys). `seq` is production order: ties keep
572/// the earliest-produced rows, matching what the general path's stable
573/// in-budget sort yields. The `BinaryHeap` is a max-heap, so `peek()`
574/// is the worst kept row.
575///
576/// v7.37.16 — `keys` moved from `Vec<f64>` (DESC pre-encoded by
577/// negation) to `Vec<OrderKey>`: text keys can't be negated, so DESC
578/// is now applied by `cmp_multi_key` via the carried `descs`.
579struct TopNEntry {
580    keys: Vec<OrderKey>,
581    descs: alloc::rc::Rc<[bool]>,
582    seq: u64,
583    row: Row<'static>,
584}
585
586impl PartialEq for TopNEntry {
587    fn eq(&self, other: &Self) -> bool {
588        self.cmp(other) == core::cmp::Ordering::Equal
589    }
590}
591impl Eq for TopNEntry {}
592impl PartialOrd for TopNEntry {
593    fn partial_cmp(&self, other: &Self) -> Option<core::cmp::Ordering> {
594        Some(self.cmp(other))
595    }
596}
597impl Ord for TopNEntry {
598    fn cmp(&self, other: &Self) -> core::cmp::Ordering {
599        cmp_multi_key(&self.keys, &other.keys, &self.descs).then(self.seq.cmp(&other.seq))
600    }
601}
602
603// v7.28 (round-22) - intermediate-row ceiling: a join whose working set
604// explodes errors instead of eating the host (mailrs watched RSS climb
605// to 7 GiB of 15 before a manual restart). The ceiling is per join
606// STAGE, not per query.
607const MAX_JOIN_INTERMEDIATE_ROWS: usize = 4_000_000;
608
609/// v7.32 — the accumulating state of the deferred-join pipeline: one
610/// `JoinSrc` / mask / width per source joined so far, the prefix column
611/// `offsets`, and the flat row-index tuple `working` set (`stride` =
612/// sources joined, `usize::MAX` = a LEFT-join NULL slot). Each join
613/// stage reads the prior state to probe the next peer and `advance`s the
614/// pipeline by one source. `consumed_cols` tracks the combined-row width
615/// built so far (the outer-left schema slice each lateral peer sees).
616struct JoinPipeline<'a> {
617    sources: Vec<JoinSrc<'a>>,
618    masks: Vec<Option<Vec<bool>>>,
619    widths: Vec<usize>,
620    offsets: Vec<usize>,
621    /// v7.37.43 (DISTA A-2) — combined-position → source-index map; kept
622    /// in sync with `offsets` by `new` / `advance`.
623    pos_to_src: Vec<u16>,
624    working: Vec<usize>,
625    stride: usize,
626    consumed_cols: usize,
627}
628
629impl<'a> JoinPipeline<'a> {
630    /// Seed the pipeline with the primary source (one stage, stride 1).
631    fn new(
632        primary: JoinSrc<'a>,
633        mask: Option<Vec<bool>>,
634        width: usize,
635        working: Vec<usize>,
636    ) -> Self {
637        let offsets = alloc::vec![0, width];
638        let pos_to_src = build_pos_to_src(&offsets);
639        Self {
640            sources: alloc::vec![primary],
641            masks: alloc::vec![mask],
642            widths: alloc::vec![width],
643            offsets,
644            pos_to_src,
645            working,
646            stride: 1,
647            consumed_cols: width,
648        }
649    }
650
651    /// Working-set row count (tuples / stride).
652    fn rows(&self) -> usize {
653        self.working.len() / self.stride
654    }
655
656    /// Consume one peer: replace the working set with `next`, append the
657    /// peer's `source` / `mask` / width, and grow the stride + offsets.
658    fn advance(
659        &mut self,
660        next: Vec<usize>,
661        source: JoinSrc<'a>,
662        mask: Option<Vec<bool>>,
663        right_arity: usize,
664    ) {
665        self.working = next;
666        self.stride += 1;
667        self.sources.push(source);
668        self.masks.push(mask);
669        self.consumed_cols += right_arity;
670        self.offsets.push(self.consumed_cols);
671        self.widths.push(right_arity);
672        // v7.37.43 (DISTA A-2) — extend the pos_to_src table for the
673        // new peer's column span. `as u16` truncation is safe: source
674        // counts in practice are O(small).
675        let k = (self.sources.len() - 1) as u16;
676        for _ in 0..right_arity {
677            self.pos_to_src.push(k);
678        }
679    }
680}
681
682/// Per-source column mask: which columns the statement references
683/// (`None` = keep all). In-place join sources apply it at
684/// materialisation time instead of `null_out_unreferenced`.
685fn keep_mask(
686    needed: Option<&alloc::collections::BTreeSet<(String, String)>>,
687    cols: &[ColumnSchema],
688    alias: &str,
689) -> Option<Vec<bool>> {
690    let needed = needed?;
691    let keep: Vec<bool> = cols
692        .iter()
693        .map(|c| needed.contains(&(alias.to_string(), c.name.clone())))
694        .collect();
695    if keep.iter().all(|k| *k) {
696        None
697    } else {
698        Some(keep)
699    }
700}
701
702/// Split a peer's ON into hash-join `eq_pairs` — `(left combined
703/// position, right peer position)` — and the `residual` conjuncts that
704/// evaluate on matched candidates. Both empty for a LATERAL peer or a
705/// peer with no ON. The returned residual refs borrow the underlying ON
706/// expressions (not the `peer` itself, since `peer.on` is a `Copy`
707/// reference), so the caller can still mutate `peer` afterwards.
708fn extract_join_keys<'a>(
709    peer: &JoinedPeer<'a>,
710    combined_schema: &[ColumnSchema],
711    consumed_cols: usize,
712) -> (
713    Vec<(usize, usize)>,
714    // v7.39 (round 719) — the third member is the whole CONJUNCT the
715    // (left-pos, key-expr) pair came from, so the int-keyed lane can
716    // identify it in `residual` and drop the re-verification (see
717    // `join_stage_hash`).
718    Vec<(usize, &'a Expr, &'a Expr)>,
719    // v7.39 (round 720) — the MIRROR: `<peer column> = <integer-only
720    // expression over the joined left side>` (`ON b.id = a.id + 500000`,
721    // the shape the EXISTS pull-up emits). (peer-col pos, left expr,
722    // conjunct). Only the integer-only shape is collected — anything
723    // else keeps the residual path it has today.
724    Vec<(usize, &'a Expr, &'a Expr)>,
725    Vec<&'a Expr>,
726) {
727    let mut eq_pairs: Vec<(usize, usize)> = Vec::new();
728    let mut eq_exprs: Vec<(usize, &Expr, &Expr)> = Vec::new();
729    let mut eq_probe_exprs: Vec<(usize, &Expr, &Expr)> = Vec::new();
730    let mut residual: Vec<&Expr> = Vec::new();
731    if let (Some(on_expr), None) = (peer.on, peer.lateral) {
732        for sub in reorder::split_and_conjunctions(on_expr) {
733            if let Some(pair) = match_equi_pair(sub, peer, combined_schema, consumed_cols) {
734                eq_pairs.push(pair);
735                continue;
736            }
737            // v7.39 (round 590) — an equality whose peer side is COMPUTED is
738            // still a join key. `ON a.g = b.g AND a.id = b.id + 1` used to
739            // hash on `g` alone and test the second conjunct on every
740            // candidate pair, so the work was (probe rows x bucket size):
741            // over 20k rows the cost ran 22.5 ms at one row a bucket, 686 ms
742            // at 200, and past 25 SECONDS at 20,000, where PG holds 4-11 ms
743            // by hashing on both. The conjunct stays in `residual` as well,
744            // so the join's answer never depends on the key encoding.
745            if let Some((l, e)) = match_equi_expr(sub, peer, combined_schema, consumed_cols) {
746                eq_exprs.push((l, e, sub));
747                residual.push(sub);
748                continue;
749            }
750            if let Some((p, e)) = match_equi_probe_expr(sub, peer, combined_schema, consumed_cols) {
751                eq_probe_exprs.push((p, e, sub));
752                residual.push(sub);
753                continue;
754            }
755            residual.push(sub);
756        }
757    }
758    (eq_pairs, eq_exprs, eq_probe_exprs, residual)
759}
760
761/// v7.39 (round 720) — one conjunct as `<peer plain column> = <integer-only
762/// expression over the already-joined left side>`, either order. Only the
763/// integer-only shape (the classifier below) is admitted: the consumer is
764/// the i64 lane, and everything else keeps today's residual path.
765fn match_equi_probe_expr<'a>(
766    sub: &'a Expr,
767    peer: &JoinedPeer<'_>,
768    combined_schema: &[ColumnSchema],
769    consumed_cols: usize,
770) -> Option<(usize, &'a Expr)> {
771    let Expr::Binary {
772        lhs,
773        op: spg_sql::ast::BinOp::Eq,
774        rhs,
775    } = sub
776    else {
777        return None;
778    };
779    let left_slice = &combined_schema[..consumed_cols];
780    for (a, b) in [(lhs.as_ref(), rhs.as_ref()), (rhs.as_ref(), lhs.as_ref())] {
781        if let Expr::Column(c) = a
782            && let Some(p) = Engine::peer_col_pos(&peer.alias, &peer.cols, c)
783            && matches!(
784                peer.cols[p].ty,
785                spg_storage::DataType::Int
786                    | spg_storage::DataType::BigInt
787                    | spg_storage::DataType::SmallInt
788            )
789            && !matches!(b, Expr::Column(_))
790            && expr_mentions_a_column(b)
791            && int_only_left_expr(b, left_slice)
792        {
793            return Some((p, b));
794        }
795    }
796    None
797}
798
799/// One conjunct as `<left column> = <expression over the peer alone>`, in
800/// either order. The left side has to be a plain column of the part of the
801/// row already joined, because the probe reads cells and does not
802/// materialise a row to evaluate against.
803fn match_equi_expr<'a>(
804    sub: &'a Expr,
805    peer: &JoinedPeer<'_>,
806    combined_schema: &[ColumnSchema],
807    consumed_cols: usize,
808) -> Option<(usize, &'a Expr)> {
809    let Expr::Binary {
810        lhs,
811        op: spg_sql::ast::BinOp::Eq,
812        rhs,
813    } = sub
814    else {
815        return None;
816    };
817    let left_slice = &combined_schema[..consumed_cols];
818    for (a, b) in [(lhs.as_ref(), rhs.as_ref()), (rhs.as_ref(), lhs.as_ref())] {
819        if let Expr::Column(c) = a
820            && let Some(l) = Engine::composite_col_pos(left_slice, c)
821            && !matches!(b, Expr::Column(_))
822            && peer_only_key_expr(b, peer)
823            && expr_mentions_a_column(b)
824        {
825            return Some((l, b));
826        }
827    }
828    None
829}
830
831/// Can this expression be computed from one peer row, with the same answer
832/// every time? Deliberately an allowlist of node kinds rather than a walk
833/// that asks what an expression references: a node the walk did not know
834/// about, or a function whose volatility SPG cannot look up, would both be
835/// silently admitted. Columns, literals, casts, unary and arithmetic only —
836/// which leaves `ON a.k = lower(b.k)` on the old path, recorded and not done.
837fn peer_only_key_expr(e: &Expr, peer: &JoinedPeer<'_>) -> bool {
838    use spg_sql::ast::BinOp;
839    match e {
840        Expr::Column(c) => Engine::peer_col_pos(&peer.alias, &peer.cols, c).is_some(),
841        Expr::Literal(_) => true,
842        Expr::Unary { expr, .. } | Expr::Cast { expr, .. } => peer_only_key_expr(expr, peer),
843        Expr::Binary { lhs, op, rhs } => {
844            matches!(
845                op,
846                BinOp::Add | BinOp::Sub | BinOp::Mul | BinOp::Div | BinOp::IntDiv | BinOp::Mod
847            ) && peer_only_key_expr(lhs, peer)
848                && peer_only_key_expr(rhs, peer)
849        }
850        _ => false,
851    }
852}
853
854/// v7.39 (round 719) — is this key expression INTEGER-ONLY: every column
855/// an integer-family column of the peer, every literal a plain integer,
856/// every operator closed over the integers (Add / Sub / Mul — Div and Mod
857/// stay out; integer division's result type is the arm's business, not
858/// this classifier's). When it is, the computed key can live in the i64
859/// hash table and equality ON THE KEY IS the SQL `=` — no canonical-string
860/// encoding, and no residual re-verification.
861fn int_only_key_expr(e: &Expr, peer: &JoinedPeer<'_>) -> bool {
862    use spg_sql::ast::BinOp;
863    match e {
864        Expr::Column(c) => Engine::peer_col_pos(&peer.alias, &peer.cols, c).is_some_and(|p| {
865            matches!(
866                peer.cols[p].ty,
867                spg_storage::DataType::Int
868                    | spg_storage::DataType::BigInt
869                    | spg_storage::DataType::SmallInt
870            )
871        }),
872        Expr::Literal(spg_sql::ast::Literal::Integer(_)) => true,
873        Expr::Binary { lhs, op, rhs } => {
874            matches!(op, BinOp::Add | BinOp::Sub | BinOp::Mul)
875                && int_only_key_expr(lhs, peer)
876                && int_only_key_expr(rhs, peer)
877        }
878        _ => false,
879    }
880}
881
882/// v7.39 (round 720) — the round-719 classifier's MIRROR: integer-only
883/// over the already-joined LEFT side (`ON b.id = a.id + 500000` — the
884/// shape the EXISTS pull-up emits). Same allowlist, columns resolved
885/// against the combined-row prefix instead of the peer.
886fn int_only_left_expr(e: &Expr, left_slice: &[ColumnSchema]) -> bool {
887    use spg_sql::ast::BinOp;
888    match e {
889        Expr::Column(c) => Engine::composite_col_pos(left_slice, c).is_some_and(|p| {
890            matches!(
891                left_slice[p].ty,
892                spg_storage::DataType::Int
893                    | spg_storage::DataType::BigInt
894                    | spg_storage::DataType::SmallInt
895            )
896        }),
897        Expr::Literal(spg_sql::ast::Literal::Integer(_)) => true,
898        Expr::Binary { lhs, op, rhs } => {
899            matches!(op, BinOp::Add | BinOp::Sub | BinOp::Mul)
900                && int_only_left_expr(lhs, left_slice)
901                && int_only_left_expr(rhs, left_slice)
902        }
903        _ => false,
904    }
905}
906
907/// v7.39 (round 720) — evaluate an `int_only_left_expr` against one probe
908/// TUPLE, reading cells straight out of the join sources: no row
909/// materialisation, no allocation. `Ok(None)` = a NULL column (the key
910/// joins nothing, SQL `=`); overflow errors with the integer family's
911/// own sentence, as the interpreted path errors.
912fn eval_int_only_probe(
913    e: &Expr,
914    left_slice: &[ColumnSchema],
915    sources: &[JoinSrc<'_>],
916    offsets: &[usize],
917    tuple: &[usize],
918) -> Result<Option<i64>, EngineError> {
919    use spg_sql::ast::BinOp;
920    match e {
921        Expr::Column(c) => {
922            let pos = Engine::composite_col_pos(left_slice, c).expect("classifier-checked");
923            Ok(match tuple_value(sources, offsets, tuple, pos) {
924                Some(Value::BigInt(n)) => Some(*n),
925                Some(Value::Int(n)) => Some(i64::from(*n)),
926                Some(Value::SmallInt(n)) => Some(i64::from(*n)),
927                _ => None,
928            })
929        }
930        Expr::Literal(spg_sql::ast::Literal::Integer(n)) => Ok(Some(*n)),
931        Expr::Binary { lhs, op, rhs } => {
932            let (Some(a), Some(b)) = (
933                eval_int_only_probe(lhs, left_slice, sources, offsets, tuple)?,
934                eval_int_only_probe(rhs, left_slice, sources, offsets, tuple)?,
935            ) else {
936                return Ok(None);
937            };
938            let out = match op {
939                BinOp::Add => a.checked_add(b),
940                BinOp::Sub => a.checked_sub(b),
941                BinOp::Mul => a.checked_mul(b),
942                _ => unreachable!("classifier admits Add/Sub/Mul only"),
943            };
944            out.map(Some).ok_or_else(|| {
945                EngineError::Eval(crate::eval::EvalError::TypeMismatch {
946                    detail: "bigint out of range".into(),
947                })
948            })
949        }
950        _ => unreachable!("classifier admits columns/integers/arithmetic only"),
951    }
952}
953
954/// A key made only of constants would be a filter, not a join key.
955fn expr_mentions_a_column(e: &Expr) -> bool {
956    match e {
957        Expr::Column(_) => true,
958        Expr::Unary { expr, .. } | Expr::Cast { expr, .. } => expr_mentions_a_column(expr),
959        Expr::Binary { lhs, rhs, .. } => expr_mentions_a_column(lhs) || expr_mentions_a_column(rhs),
960        _ => false,
961    }
962}
963
964/// One conjunct as an equi-join key for `peer`: `<left>.<col> = <peer>.<col>`
965/// in either order, where the left side resolves inside the part of the
966/// combined row already joined. `None` when it is anything else.
967fn match_equi_pair(
968    sub: &Expr,
969    peer: &JoinedPeer<'_>,
970    combined_schema: &[ColumnSchema],
971    consumed_cols: usize,
972) -> Option<(usize, usize)> {
973    let Expr::Binary {
974        lhs,
975        op: spg_sql::ast::BinOp::Eq,
976        rhs,
977    } = sub
978    else {
979        return None;
980    };
981    let (Expr::Column(a), Expr::Column(b)) = (lhs.as_ref(), rhs.as_ref()) else {
982        return None;
983    };
984    let left_slice = &combined_schema[..consumed_cols];
985    if let (Some(l), Some(r)) = (
986        Engine::composite_col_pos(left_slice, a),
987        Engine::peer_col_pos(&peer.alias, &peer.cols, b),
988    ) {
989        return Some((l, r));
990    }
991    if let (Some(l), Some(r)) = (
992        Engine::composite_col_pos(left_slice, b),
993        Engine::peer_col_pos(&peer.alias, &peer.cols, a),
994    ) {
995        return Some((l, r));
996    }
997    None
998}
999
1000/// v7.39 (round 588) — the WHERE conjuncts that could be equi-join keys.
1001///
1002/// `FROM a, b WHERE a.id = b.id` is the ANSI-89 spelling of
1003/// `FROM a JOIN b ON a.id = b.id` and means exactly the same join, but the
1004/// equality arrives in the WHERE clause: `analyze_join_pushdown` cannot place
1005/// it on either relation (its two qualifiers name two), and
1006/// `extract_join_keys` only ever read the ON clause. The peer was left with
1007/// no key at all and fell to the nested-loop stage, which crosses the ENTIRE
1008/// peer against every surviving left row.
1009///
1010/// The rewrite is only sound while every relation in the chain is
1011/// non-nullable — under an outer join a WHERE equality filters AFTER the
1012/// NULL-filling and is not the same thing as a join condition — so one outer
1013/// join anywhere gives up on the whole statement.
1014fn where_equi_candidates<'w>(from: &FromClause, where_: Option<&'w Expr>) -> Vec<&'w Expr> {
1015    let Some(w) = where_ else { return Vec::new() };
1016    if !from
1017        .joins
1018        .iter()
1019        .all(|j| matches!(j.kind, JoinKind::Inner | JoinKind::Cross))
1020    {
1021        return Vec::new();
1022    }
1023    reorder::split_and_conjunctions(w)
1024        .into_iter()
1025        .filter(|sub| {
1026            matches!(
1027                sub,
1028                Expr::Binary { lhs, op: spg_sql::ast::BinOp::Eq, rhs }
1029                    if matches!((lhs.as_ref(), rhs.as_ref()), (Expr::Column(_), Expr::Column(_)))
1030            )
1031        })
1032        .collect()
1033}
1034
1035impl Engine {
1036    /// v7.17.0 Phase 3.P0-41 — build the per-peer descriptor for each
1037    /// join stage. A LATERAL peer can't be pre-materialised (its rows
1038    /// depend on outer columns), so it gets a sentinel carrying just
1039    /// the probed projection schema and the inner SELECT to re-run per
1040    /// outer row. A plain table with no pushed predicate is left
1041    /// deferred (the index-nested-loop path may avoid cloning it
1042    /// entirely). Everything else materialises eagerly to a
1043    /// (rows, schema) pair. `peer_preds[i]` are the WHERE conjuncts
1044    /// pushed onto peer `i` by `analyze_join_pushdown`.
1045    #[allow(clippy::type_complexity)]
1046    fn build_join_peers<'a>(
1047        &self,
1048        from: &'a FromClause,
1049        peer_preds: &[Vec<&Expr>],
1050        needed: Option<&alloc::collections::BTreeSet<(String, String)>>,
1051        budget: &mut ByteBudget,
1052    ) -> Result<Vec<JoinedPeer<'a>>, EngineError> {
1053        let mut joined: Vec<JoinedPeer<'a>> = Vec::new();
1054        for j in &from.joins {
1055            let a = j
1056                .table
1057                .alias
1058                .as_deref()
1059                .unwrap_or(j.table.name.as_str())
1060                .to_string();
1061            if let Some(inner_box) = &j.table.lateral_subquery {
1062                // v7.37 D.19 — a NON-correlated derived-table peer (a bare
1063                // `(VALUES …)`, which lowers to a UNION-ALL SELECT, or an
1064                // uncorrelated subquery) must be materialised ONCE as an eager
1065                // peer and cross-joined against every left row. Forcing it
1066                // through the per-left-row lateral path below dropped its
1067                // UNION-ALL rows, so `JOIN (VALUES ('1'),('2')) b ON true`
1068                // yielded only the first value per left row instead of the
1069                // full product. Only genuinely correlated laterals (which
1070                // reference an outer column) need per-left-row evaluation.
1071                // v7.39 (round 572) — …and that is what this now asks.
1072                //
1073                // The gate was `is_constant_values_derived`, which only
1074                // recognises a literal VALUES list, so an ordinary
1075                // uncorrelated derived table — `JOIN (SELECT … FROM t
1076                // WHERE …) b ON …`, as common a shape as SQL has — was
1077                // re-executed once per LEFT ROW. Measured on a 500k
1078                // table:
1079                //
1080                //     derived table alone                34.7 ms
1081                //     … as a join peer, 500 rows      8,552 ms
1082                //     … 2000 rows                    32,610 ms
1083                //     … 20000 rows                  >120,000 ms (cancelled)
1084                //     PG18, the 20000-row form           15.8 ms
1085                //
1086                // Linear in the LEFT side because the inner SELECT ran
1087                // again for each of its rows. `select_is_correlated` is
1088                // built to be wrong in the safe direction — its own
1089                // comment says a wrong "yes" costs only a
1090                // re-evaluation, while a wrong "no" is silently wrong —
1091                // so it is exactly the question to ask here.
1092                if is_constant_values_derived(inner_box)
1093                    || (derived_is_plain_table_select(inner_box, self.active_catalog())
1094                        && !crate::subquery::select_is_correlated(inner_box))
1095                {
1096                    let pidx = from
1097                        .joins
1098                        .iter()
1099                        .position(|jj| core::ptr::eq(jj, j))
1100                        .unwrap_or(0);
1101                    let (mut rows, mut cols) =
1102                        self.materialise_table_ref_filtered(&j.table, &peer_preds[pidx])?;
1103                    // `AS y(a, b)` renames positionally here exactly as
1104                    // it does on the per-left-row path below.
1105                    for (i, new_name) in j.table.unnest_column_aliases.iter().enumerate() {
1106                        if let Some(col) = cols.get_mut(i) {
1107                            col.name = new_name.clone();
1108                        }
1109                    }
1110                    if let Some(needed) = needed {
1111                        Self::null_out_unreferenced(&mut rows, &cols, &a, needed);
1112                    }
1113                    budget.charge(approx_rows_bytes(&rows))?;
1114                    joined.push(JoinedPeer {
1115                        eager_rows: Some(rows),
1116                        cols,
1117                        alias: a,
1118                        kind: j.kind,
1119                        on: j.on.as_ref(),
1120                        lateral: None,
1121                        join_table: None,
1122                        where_preds: Vec::new(),
1123                    });
1124                    continue;
1125                }
1126                // Probe schema by running the inner SELECT against a
1127                // NULL-padded outer context. The probe gives us the
1128                // projection's column shape; rows materialise per
1129                // left-row below.
1130                let mut schema = self.lateral_probe_schema(inner_box)?;
1131                // v7.37.16 — `AS y(a, b)` column-alias list renames the
1132                // derived table's columns positionally, exactly as the
1133                // FROM-primary derived-table path does (select.rs). The
1134                // probe returns the inner SELECT's own column names
1135                // (`column1` for a VALUES list, the inner projection
1136                // name for a subquery); without this rename a join
1137                // right-operand derived table left `y.a` unresolved
1138                // while `y.column1` / the inner name worked — a PG
1139                // divergence (PG applies the alias list identically in
1140                // FROM-primary and join-operand positions).
1141                for (i, new_name) in j.table.unnest_column_aliases.iter().enumerate() {
1142                    if let Some(col) = schema.get_mut(i) {
1143                        col.name = new_name.clone();
1144                    }
1145                }
1146                joined.push(JoinedPeer {
1147                    eager_rows: None,
1148                    cols: schema,
1149                    alias: a,
1150                    kind: j.kind,
1151                    on: j.on.as_ref(),
1152                    lateral: Some(inner_box.as_ref()),
1153                    join_table: None,
1154                    where_preds: Vec::new(),
1155                });
1156            } else {
1157                let pidx = from
1158                    .joins
1159                    .iter()
1160                    .position(|jj| core::ptr::eq(jj, j))
1161                    .unwrap_or(0);
1162                // v7.28 - defer materialisation for plain tables so the
1163                // index-nested-loop path can seek the driver and look up
1164                // only matched peer rows instead of cloning the whole
1165                // table. v7.33 — defer EVEN WITH a pushed WHERE predicate:
1166                // carry the predicate as `where_preds` for the stages to
1167                // apply as a residual on matched pairs (the eager path
1168                // here scanned + filtered the entire peer table, which on
1169                // mailrs's snippet subquery cost a full email_analysis scan
1170                // per seeked thread — 60× per IN-list group). Correctness
1171                // is backstopped by filter_join_survivors re-applying the
1172                // full WHERE to survivors.
1173                let plain = j.table.unnest_expr.is_none() && j.table.as_of_segment.is_none();
1174                if plain && let Some(t) = self.active_catalog().get(&j.table.name) {
1175                    // v7.34 (B5 ledger) — cost guard for 169ef66's INL
1176                    // pushdown: when the peer table is tiny AND a WHERE
1177                    // conjunct pushes onto it, the v7.28 eager path
1178                    // (scan + filter once, O(peer.rows + driver.rows))
1179                    // always beats INL (one peer-index seek + filter per
1180                    // driver row, O(driver.rows × log peer.rows + matched
1181                    // pair filter)). 169ef66 fixed mailrs's
1182                    // get_conversations IN(60) snippet subquery (peer
1183                    // 6k email_analysis, driver 25k messages — INL wins
1184                    // 13.7×), but regressed INBOX's outer mailboxes JOIN
1185                    // (peer = 30, driver = 25k — eager wins ~+4ms p50).
1186                    // SMALL_PEER_EAGER_ROWS at 256 keeps the IN(60) win
1187                    // (6k > 256 stays INL) while clawing back the
1188                    // small-peer case (30 ≤ 256 goes eager).
1189                    const SMALL_PEER_EAGER_ROWS: usize = 256;
1190                    let has_pushdown = !peer_preds[pidx].is_empty();
1191                    // v7.36 — drop the 7.35.1 force-eager-when-cold
1192                    // workaround. The downstream INL probe and hash
1193                    // build now thread cold-tier rows through
1194                    // `JoinSrc::Mixed` (PK-key map for INL;
1195                    // hash-iter Mixed.get for hash build). The
1196                    // nested-loop fallback's `lazy_rows` also
1197                    // appends cold rows. Small-peer + pushdown
1198                    // still takes the eager fast path.
1199                    let peer_total = t.rows().len();
1200                    if has_pushdown && peer_total <= SMALL_PEER_EAGER_ROWS {
1201                        let (mut rows, cols) =
1202                            self.materialise_table_ref_filtered(&j.table, &peer_preds[pidx])?;
1203                        if let Some(needed) = needed {
1204                            Self::null_out_unreferenced(&mut rows, &cols, &a, needed);
1205                        }
1206                        budget.charge(approx_rows_bytes(&rows))?;
1207                        joined.push(JoinedPeer {
1208                            eager_rows: Some(rows),
1209                            cols,
1210                            alias: a,
1211                            kind: j.kind,
1212                            on: j.on.as_ref(),
1213                            lateral: None,
1214                            join_table: Some(j.table.name.clone()),
1215                            where_preds: Vec::new(),
1216                        });
1217                        continue;
1218                    }
1219                    joined.push(JoinedPeer {
1220                        eager_rows: None,
1221                        cols: t.schema().columns.clone(),
1222                        alias: a,
1223                        kind: j.kind,
1224                        on: j.on.as_ref(),
1225                        lateral: None,
1226                        join_table: Some(j.table.name.clone()),
1227                        where_preds: peer_preds[pidx].iter().map(|e| (*e).clone()).collect(),
1228                    });
1229                    continue;
1230                }
1231                // Non-table peer (UNNEST / AS OF SEGMENT) — materialise
1232                // eagerly with its predicate filter applied up front.
1233                let (mut rows, cols) =
1234                    self.materialise_table_ref_filtered(&j.table, &peer_preds[pidx])?;
1235                if let Some(needed) = needed {
1236                    Self::null_out_unreferenced(&mut rows, &cols, &a, needed);
1237                }
1238                budget.charge(approx_rows_bytes(&rows))?;
1239                joined.push(JoinedPeer {
1240                    eager_rows: Some(rows),
1241                    cols,
1242                    alias: a,
1243                    kind: j.kind,
1244                    on: j.on.as_ref(),
1245                    lateral: None,
1246                    join_table: Some(j.table.name.clone()),
1247                    where_preds: Vec::new(),
1248                });
1249            }
1250        }
1251        Ok(joined)
1252    }
1253
1254    pub(crate) fn build_joined_filtered_rows(
1255        &self,
1256        from: &FromClause,
1257        where_: Option<&Expr>,
1258        cancel: CancelToken<'_>,
1259        needed: Option<&alloc::collections::BTreeSet<(String, String)>>,
1260        budget: &mut ByteBudget,
1261    ) -> Result<DeferredJoin<'_>, EngineError> {
1262        let (swapped_from, primary_preds, peer_preds) = analyze_join_pushdown(from, where_);
1263        // v7.37.x (mailrs Track A perf — SPGE ≫ PG18) — pushed conjuncts
1264        // are enforced AT the primary `filter_table_indices` (or eager
1265        // peer `materialise_table_ref_filtered`) AND/OR as a join-stage
1266        // residual via `where_preds`. Re-applying them per joined tuple
1267        // inside `filter_join_survivors` is pure waste — 30 k tuples ×
1268        // compiled-WHERE eval cost ~1 ms on the mailrs minimal probe.
1269        // Build a residual WHERE = `where_ \ pushed_conjuncts` and pass
1270        // only that to the survivor filter. Identity is by `Expr` pointer
1271        // (analyze_join_pushdown gave us borrows into `where_`'s conjunct
1272        // set, so the pointers match exactly).
1273        // v7.39 (round 588) — the set grows during the peer loop below: a
1274        // WHERE equality promoted to a peer's join key is enforced BY the
1275        // join and must not be re-applied per survivor either.
1276        let mut pushed_set: alloc::collections::BTreeSet<usize> = primary_preds
1277            .iter()
1278            .chain(peer_preds.iter().flat_map(|v| v.iter()))
1279            .map(|e| core::ptr::from_ref::<Expr>(*e) as usize)
1280            .collect();
1281        let from = swapped_from.as_ref().unwrap_or(from);
1282        let primary_alias = from
1283            .primary
1284            .alias
1285            .as_deref()
1286            .unwrap_or(from.primary.name.as_str())
1287            .to_string();
1288        // v7.31 (perf campaign) — when the primary is a plain stored
1289        // table and there are joins to run, keep it in place: filter
1290        // to row indices (same index seek / linear filter) and let
1291        // the deferred-join pipeline clone only the surviving,
1292        // referenced columns once at output time. Joinless FROMs and
1293        // non-table refs take the materialising path.
1294        //
1295        // v7.30.3 byte-budget interplay: the index path materialises
1296        // nothing (row numbers are 8 B each), so the budget charges
1297        // land where the clones happen — the materialising fallback
1298        // here, eager peers below, and the output assembly.
1299        // v7.39 (round 790) — the joins-only exclusion here was TRIED
1300        // and reverted: relaxing it, so a joinless FROM seeds the
1301        // primary by row index instead of materialising, measured
1302        // WORSE (147 MB → 178 MB on a 300k-row probe). Round 800 found
1303        // where the extra memory comes from, and it is not the output
1304        // assembly this comment used to blame.
1305        //
1306        // Peak RSS, fresh server per cell, measured either side of the
1307        // gate. The number that settles it is the baseline — taken
1308        // after seeding and a single `WHERE id = 1` read, before any
1309        // scan: 423 MB as it stands, 600 MB with the gate relaxed. One
1310        // row of output, 177 MB apart.
1311        //
1312        // Seeding the primary by index means reading rows in place out
1313        // of the stored `PersistentVec`, and touching it makes the
1314        // whole table resident. Materialising copies only the surviving
1315        // rows — one, for that warm-up — and keeps them in a compact
1316        // Vec. So the copy is not the expensive representation here;
1317        // in-place access is, and it costs the table's full residency
1318        // whatever the query then does with it.
1319        //
1320        // The 72 MB this copy costs on a full scan is real (round 798
1321        // decomposed it), but it is not recoverable by flipping this
1322        // gate. Anything that goes after it has to avoid making the
1323        // table resident, not merely avoid the copy.
1324        //
1325        // And memory is not even the strongest objection. The relaxed
1326        // build was left on the test machine by accident and a gate run
1327        // caught what the memory probes never would:
1328        // `e2e_empty_target_list_round341` failed deterministically —
1329        // a zero-column result set (`SELECT` with an empty target list)
1330        // returned no DataRows at all where three were owed. Seeding
1331        // the primary by row index does not merely cost more, it drops
1332        // rows for a projection with nothing in it.
1333        let primary_table: Option<&Table> = if !from.joins.is_empty()
1334            && from.primary.unnest_expr.is_none()
1335            && from.primary.lateral_subquery.is_none()
1336            && from.primary.as_of_segment.is_none()
1337        {
1338            self.active_catalog().get(&from.primary.name).filter(|t|
1339                // v7.36 (cold-tier coverage) — the deferred-index
1340                // primary path threads `Vec<usize>` row indices into
1341                // `JoinSrc::Stored(t.rows())` (hot-tier only), so a
1342                // primary with cold-tier rows silently dropped them
1343                // from the join. Force the materialising fallback
1344                // when ANY cold-tier row exists; the fallback rides
1345                // `materialise_table_ref_filtered` which already
1346                // covers both tiers (v7.35.1).
1347                !t.has_cold_rows_fast())
1348        } else {
1349            None
1350        };
1351        let (primary_rows, primary_cols, primary_indices) = match primary_table {
1352            Some(t) => {
1353                let idxs = self.filter_table_indices(t, &primary_alias, &primary_preds)?;
1354                // Phase C.3 step 2b — MVCC read gate on the deferred-index
1355                // primary seed. `idxs` are hot-tier physical indices into
1356                // `t.rows()` (this arm is reached only when the primary has
1357                // NO cold rows, see `has_cold_rows_fast()` filter above), so
1358                // dropping the invisible ones here keeps a dead/old version
1359                // out of the join without touching any cold-tier row. No-op
1360                // today: every hot header is frozen/committed-alive.
1361                let scan_snapshot = self.current_snapshot();
1362                let idxs: Vec<usize> = idxs
1363                    .into_iter()
1364                    .filter(|&i| t.is_row_visible(i, &scan_snapshot))
1365                    .collect();
1366                (Vec::new(), t.schema().columns.clone(), Some(idxs))
1367            }
1368            None => {
1369                let (mut rows, cols) =
1370                    self.materialise_table_ref_filtered(&from.primary, &primary_preds)?;
1371                if let Some(needed) = needed {
1372                    Self::null_out_unreferenced(&mut rows, &cols, &primary_alias, needed);
1373                }
1374                budget.charge(approx_rows_bytes(&rows))?;
1375                (rows, cols, None)
1376            }
1377        };
1378        let mut joined = self.build_join_peers(from, &peer_preds, needed, budget)?;
1379        let combined_schema = build_combined_schema(&primary_alias, &primary_cols, &joined);
1380        // v7.39 (read01 round 53) — the join's EvalContext must carry the
1381        // catalog. Without it a `::regclass` / enum / composite cast inside a
1382        // joined WHERE or ON falls back to plain text, so the canonical
1383        // `pg_class JOIN pg_index … WHERE indrelid = 't'::regclass` shape
1384        // errored on "comparison between BigInt and Text" — while the very
1385        // same predicate worked on a single-table SELECT (whose ctx does carry
1386        // the catalog). Same root as round 49's unnest(enum_range(…)).
1387        // v7.39 (round 525) — and the SESSION, for the same reason as the
1388        // catalog above: a join's WHERE is the same predicate a
1389        // single-table SELECT would carry, and `WHERE t =
1390        // current_setting('app.tenant')` failed on the joined shape while
1391        // working on the unjoined one.
1392        let join_sess = self.dml_session();
1393        let ctx = EvalContext::new(&combined_schema, None)
1394            .with_catalog(self.active_catalog())
1395            .with_session(&join_sess);
1396        if joined.is_empty() {
1397            // Joinless FROM: the primary rows ARE the combined rows —
1398            // filter and hand them back without any re-clone.
1399            let mut filtered: Vec<Row<'static>> = Vec::new();
1400            let mut memo = memoize::MemoizeCache::default();
1401            for row in primary_rows {
1402                if let Some(where_expr) = where_ {
1403                    let cond = self.eval_expr_with_correlated(
1404                        where_expr,
1405                        &row,
1406                        &ctx,
1407                        cancel,
1408                        Some(&mut memo),
1409                    )?;
1410                    if !crate::eval::predicate_is_true(&cond, "JOIN/ON", ctx.mysql_dialect)? {
1411                        continue;
1412                    }
1413                }
1414                filtered.push(row);
1415            }
1416            // v7.32 (P4 increment 2) — joinless: the survivors ARE the
1417            // primary rows; wrap them as one Owned source with identity
1418            // tuples so the deferred output type stays uniform.
1419            let width = combined_schema.len();
1420            let n = filtered.len();
1421            let offsets = alloc::vec![0, width];
1422            let pos_to_src = build_pos_to_src(&offsets);
1423            return Ok(DeferredJoin {
1424                sources: alloc::vec![JoinSrc::Owned(filtered)],
1425                offsets,
1426                pos_to_src,
1427                widths: alloc::vec![width],
1428                masks: alloc::vec![None],
1429                survivors: (0..n).collect(),
1430                stride: 1,
1431                combined_schema,
1432            });
1433        }
1434        // v7.31 (perf campaign) — deferred join materialisation: the
1435        // working set is a flat row-index tuple vec (stride = sources
1436        // joined so far, usize::MAX = a LEFT-join NULL slot), so a
1437        // combined Row materialises only where a residual-ON / lateral /
1438        // WHERE eval needs one and for the survivors handed back. Seed
1439        // the pipeline with the primary, then advance it one peer at a
1440        // time through the index-nested-loop, hash equi-join, or
1441        // nested-loop strategy.
1442        let primary_width = primary_cols.len();
1443        #[allow(clippy::type_complexity)]
1444        let (primary_source, primary_mask, working): (
1445            JoinSrc<'_>,
1446            Option<Vec<bool>>,
1447            Vec<usize>,
1448        ) = match primary_indices {
1449            Some(idxs) => {
1450                let t = primary_table.expect("stored primary");
1451                (
1452                    JoinSrc::Stored(t.rows()),
1453                    keep_mask(needed, &primary_cols, &primary_alias),
1454                    idxs,
1455                )
1456            }
1457            None => {
1458                let n = primary_rows.len();
1459                (JoinSrc::Owned(primary_rows), None, (0..n).collect())
1460            }
1461        };
1462        let where_equi = where_equi_candidates(from, where_);
1463        let mut pipe = JoinPipeline::new(primary_source, primary_mask, primary_width, working);
1464        for peer in &mut joined {
1465            if pipe.rows() > MAX_JOIN_INTERMEDIATE_ROWS {
1466                return Err(EngineError::Unsupported(alloc::format!(
1467                    "join intermediate result exceeds {MAX_JOIN_INTERMEDIATE_ROWS} rows ({} so far) - add join predicates",
1468                    pipe.rows()
1469                )));
1470            }
1471            let right_arity = peer.cols.len();
1472            let peer_mask = keep_mask(needed, &peer.cols, &peer.alias);
1473            let (mut eq_pairs, eq_exprs, eq_probe_exprs, residual) =
1474                extract_join_keys(peer, &combined_schema, pipe.consumed_cols);
1475            // v7.39 (round 588) — an ANSI-89 join writes its condition in the
1476            // WHERE clause. Give the peer those keys too, so `FROM a, b WHERE
1477            // a.id = b.id` hashes exactly like `FROM a JOIN b ON a.id = b.id`
1478            // instead of crossing all of `b` against every row of `a`.
1479            if peer.lateral.is_none() && matches!(peer.kind, JoinKind::Inner | JoinKind::Cross) {
1480                for cand in &where_equi {
1481                    if let Some(pair) =
1482                        match_equi_pair(cand, peer, &combined_schema, pipe.consumed_cols)
1483                        && !eq_pairs.contains(&pair)
1484                    {
1485                        eq_pairs.push(pair);
1486                        pushed_set.insert(core::ptr::from_ref::<Expr>(*cand) as usize);
1487                    }
1488                }
1489            }
1490            // v7.33 — a deferred peer's pushed WHERE conjuncts ride as extra
1491            // residual so the INL / hash stages drop non-matching (left,
1492            // right) pairs in place (the eager path used to pre-filter the
1493            // whole peer). Taken out of `peer` so the &mut hash call below
1494            // doesn't alias the residual borrow.
1495            let extra_preds = core::mem::take(&mut peer.where_preds);
1496            let residual: Vec<&Expr> = residual.into_iter().chain(extra_preds.iter()).collect();
1497            // v7.39 (round 725) — SEMI stays out of the INL walker (its
1498            // per-hit push has no first-match short-circuit); the hash
1499            // stage right below is where the pull-up's keys land anyway.
1500            if !matches!(peer.kind, JoinKind::Semi)
1501                && self.join_stage_inl(
1502                    &mut pipe,
1503                    peer,
1504                    &eq_pairs,
1505                    &residual,
1506                    &peer_mask,
1507                    right_arity,
1508                    &ctx,
1509                    cancel,
1510                )?
1511            {
1512                continue;
1513            }
1514            // v7.39 (round 606) — a COMPUTED key on its own is still a key.
1515            // Round 590 taught the hash stage to take `eq_exprs`, but the
1516            // gate here only ever asked about `eq_pairs`, so the machinery
1517            // was reachable only when a plain `col = col` conjunct sat
1518            // beside it. `ON a.id = b.id + 1` alone — the ordinary
1519            // previous-row / offset-by-one join, and the anti-join
1520            // `LEFT JOIN … ON a.id = b.id + 1 WHERE b.id IS NULL` — fell
1521            // through to the nested loop and crossed the whole peer against
1522            // every left row: quadratic, 701 ms at 2k rows and past 20
1523            // SECONDS at 20k where PG holds 0.4-2.7 ms.
1524            if (!eq_pairs.is_empty() || !eq_exprs.is_empty() || !eq_probe_exprs.is_empty())
1525                && peer.lateral.is_none()
1526            {
1527                self.join_stage_hash(
1528                    &mut pipe,
1529                    peer,
1530                    &eq_pairs,
1531                    &eq_exprs,
1532                    &eq_probe_exprs,
1533                    &residual,
1534                    &peer_mask,
1535                    right_arity,
1536                    &combined_schema,
1537                    &ctx,
1538                    cancel,
1539                )?;
1540                continue;
1541            }
1542            self.join_stage_nested(
1543                &mut pipe,
1544                peer,
1545                right_arity,
1546                &combined_schema,
1547                &ctx,
1548                cancel,
1549                needed,
1550                budget,
1551            )?;
1552        }
1553        // v7.39 (round 588) — built here rather than before the loop because
1554        // `pushed_set` only learns about promoted equi-keys as each peer is
1555        // planned. A conjunct that failed to promote is still in the set's
1556        // complement and is still enforced, so a shape this does not
1557        // recognise keeps the old, correct behaviour.
1558        let residual_where_owned: Option<Expr> = where_.and_then(|w| {
1559            let kept: Vec<Expr> = reorder::split_and_conjunctions(w)
1560                .into_iter()
1561                .filter(|c| !pushed_set.contains(&(core::ptr::from_ref::<Expr>(c) as usize)))
1562                .cloned()
1563                .collect();
1564            kept.into_iter().reduce(|a, b| Expr::Binary {
1565                lhs: alloc::boxed::Box::new(a),
1566                op: spg_sql::ast::BinOp::And,
1567                rhs: alloc::boxed::Box::new(b),
1568            })
1569        });
1570        let survivors =
1571            self.filter_join_survivors(&pipe, residual_where_owned.as_ref(), &ctx, cancel, budget)?;
1572        Ok(DeferredJoin {
1573            sources: pipe.sources,
1574            offsets: pipe.offsets,
1575            pos_to_src: pipe.pos_to_src,
1576            widths: pipe.widths,
1577            masks: pipe.masks,
1578            survivors,
1579            stride: pipe.stride,
1580            combined_schema,
1581        })
1582    }
1583
1584    /// v7.28 (round-22) — index-nested-loop join stage. When the working
1585    /// set is small and the peer's join column has a BTree, seek per left
1586    /// row instead of materialising the whole peer table (a correlated
1587    /// subquery body otherwise clones the full table once per outer
1588    /// group). Returns `Ok(false)` when the shape doesn't qualify, so the
1589    /// caller falls through to the hash / nested-loop strategy.
1590    #[allow(clippy::too_many_arguments)]
1591    fn join_stage_inl<'a, 'p>(
1592        &'a self,
1593        pipe: &mut JoinPipeline<'a>,
1594        peer: &JoinedPeer<'p>,
1595        eq_pairs: &[(usize, usize)],
1596        residual: &[&Expr],
1597        peer_mask: &Option<Vec<bool>>,
1598        right_arity: usize,
1599        ctx: &EvalContext,
1600        cancel: CancelToken<'_>,
1601    ) -> Result<bool, EngineError> {
1602        const INL_MAX_LEFT: usize = 1024;
1603        // v7.37.16 — RIGHT / FULL OUTER need to enumerate ALL peer rows
1604        // (to emit the unmatched ones with a NULL-filled left). The INL
1605        // probe only index-seeks the matched peer rows, so it cannot
1606        // produce the unmatched-right set. Bail to the hash stage, which
1607        // iterates every peer row and can track which matched.
1608        if matches!(peer.kind, JoinKind::Right | JoinKind::FullOuter) {
1609            return Ok(false);
1610        }
1611        let Some(tname) = &peer.join_table else {
1612            return Ok(false);
1613        };
1614        if !(peer.eager_rows.is_none() && !eq_pairs.is_empty() && pipe.rows() <= INL_MAX_LEFT) {
1615            return Ok(false);
1616        }
1617        let Some(table) = self.active_catalog().get(tname) else {
1618            return Ok(false);
1619        };
1620        let Some(idx) = peer
1621            .cols
1622            .iter()
1623            .position(|c| c.name == peer.cols[eq_pairs[0].1].name)
1624            .and_then(|pos| table.index_on(pos))
1625        else {
1626            return Ok(false);
1627        };
1628        // v7.36 — INL probe handles cold-tier locators only when the
1629        // peer's JOIN column is its single-column integer PRIMARY
1630        // KEY (the segment lookup is keyed by integer PK). For
1631        // non-PK JOINs on a cold-bearing peer, bail out so the
1632        // caller falls through to hash-join (which iterates the
1633        // peer via `Mixed` without needing locator-to-PK mapping).
1634        let has_cold = table.has_cold_rows_fast();
1635        let pk_col_pos = table
1636            .schema()
1637            .uniqueness_constraints
1638            .iter()
1639            .find(|u| u.is_primary_key && u.columns.len() == 1)
1640            .map(|u| u.columns[0]);
1641        let join_col_is_pk = pk_col_pos == Some(idx.column_position);
1642        if has_cold && !join_col_is_pk {
1643            return Ok(false);
1644        }
1645        let (cold_rows, cold_pk_map): (Vec<Row<'static>>, hashbrown::HashMap<i64, usize>) =
1646            if has_cold {
1647                crate::constraints::iter_cold_rows_with_locator_map(self.active_catalog(), table)
1648            } else {
1649                (Vec::new(), hashbrown::HashMap::new())
1650            };
1651        let stored = table.rows();
1652        let hot_len = stored.len();
1653        // Phase C.3 step 2b — MVCC read gate for the INL probe. Snapshot
1654        // computed once; a hot peer row (`ri < hot_len`) this snapshot
1655        // cannot see is skipped so it never matches. Cold rows
1656        // (`ri >= hot_len`) are frozen segment rows = always visible.
1657        // No-op today: every hot header is frozen/committed-alive.
1658        let scan_snapshot = self.current_snapshot();
1659        let (lpos0, _) = eq_pairs[0];
1660        let mut next: Vec<usize> = Vec::new();
1661        for tuple in pipe.working.chunks(pipe.stride) {
1662            cancel.check()?;
1663            let mut left_matched = false;
1664            if let Some(kv) = tuple_value(&pipe.sources, &pipe.offsets, tuple, lpos0)
1665                && !matches!(kv, Value::Null)
1666                && let Some(key) = spg_storage::IndexKey::from_value(kv)
1667            {
1668                for loc in idx.lookup_eq(&key) {
1669                    let ri = match *loc {
1670                        spg_storage::RowLocator::Hot(i) => i,
1671                        spg_storage::RowLocator::Cold { .. } => {
1672                            // Mixed-eligible branch (PK BTree
1673                            // lookup). The locator's key equals the
1674                            // PK key here; use it to find the row in
1675                            // `cold_rows` via `cold_pk_map`.
1676                            let spg_storage::IndexKey::Int(pk) = &key else {
1677                                continue;
1678                            };
1679                            match cold_pk_map.get(pk) {
1680                                Some(&off) => hot_len + off,
1681                                None => continue,
1682                            }
1683                        }
1684                    };
1685                    let right_opt: Option<&Row<'static>> = if ri < hot_len {
1686                        if !table.is_row_visible(ri, &scan_snapshot) {
1687                            continue;
1688                        }
1689                        stored.get(ri)
1690                    } else {
1691                        cold_rows.get(ri - hot_len)
1692                    };
1693                    let right = match right_opt {
1694                        Some(r) => r,
1695                        None => continue,
1696                    };
1697                    // Remaining eq pairs + residual ON check on the
1698                    // candidate only.
1699                    let mut ok = true;
1700                    for (lp, rp) in eq_pairs.iter().skip(1) {
1701                        let lv = tuple_value(&pipe.sources, &pipe.offsets, tuple, *lp);
1702                        let rv = right.values.get(*rp);
1703                        let eq = match (lv, rv) {
1704                            (Some(a), Some(b)) => {
1705                                !matches!(a, Value::Null)
1706                                    && !matches!(b, Value::Null)
1707                                    && value_cmp(a, b) == core::cmp::Ordering::Equal
1708                            }
1709                            _ => false,
1710                        };
1711                        if !eq {
1712                            ok = false;
1713                            break;
1714                        }
1715                    }
1716                    if !ok {
1717                        continue;
1718                    }
1719                    let keep = if residual.is_empty() {
1720                        true
1721                    } else {
1722                        let mut combined_vals = materialise_tuple_vals(
1723                            &pipe.sources,
1724                            &pipe.widths,
1725                            &pipe.masks,
1726                            tuple,
1727                            pipe.consumed_cols + right_arity,
1728                        );
1729                        extend_masked(&mut combined_vals, right, peer_mask.as_deref());
1730                        let combined = Row::new(combined_vals);
1731                        let mut k = true;
1732                        for r in residual {
1733                            let cond =
1734                                self.eval_expr_with_correlated(r, &combined, ctx, cancel, None)?;
1735                            if !crate::eval::predicate_is_true(&cond, "JOIN/ON", ctx.mysql_dialect)?
1736                            {
1737                                k = false;
1738                                break;
1739                            }
1740                        }
1741                        k
1742                    };
1743                    if keep {
1744                        next.extend_from_slice(tuple);
1745                        next.push(ri);
1746                        left_matched = true;
1747                    }
1748                }
1749            }
1750            if !left_matched && matches!(peer.kind, JoinKind::Left) {
1751                next.extend_from_slice(tuple);
1752                next.push(usize::MAX);
1753            }
1754        }
1755        let src = if cold_rows.is_empty() {
1756            JoinSrc::Stored(stored)
1757        } else {
1758            JoinSrc::Mixed {
1759                hot: stored,
1760                cold: cold_rows,
1761                cold_locator_map: cold_pk_map,
1762            }
1763        };
1764        pipe.advance(next, src, peer_mask.clone(), right_arity);
1765        Ok(true)
1766    }
1767
1768    /// v7.28 (round-22) — hash equi-join stage. The naive path cloned the
1769    /// full combined row for EVERY (left, right) pair before evaluating
1770    /// ON — O(L×R) materialisations (a 24k × 6k LEFT JOIN never returned).
1771    /// Build a hash on the (smaller) right side over the `eq_pairs` keys,
1772    /// probe per left tuple, and materialise only matching pairs for the
1773    /// `residual` ON conjuncts. NULL keys never match (SQL equality).
1774    #[allow(clippy::too_many_arguments)]
1775    fn join_stage_hash<'a, 'p>(
1776        &'a self,
1777        pipe: &mut JoinPipeline<'a>,
1778        peer: &mut JoinedPeer<'p>,
1779        eq_pairs: &[(usize, usize)],
1780        eq_exprs: &[(usize, &Expr, &Expr)],
1781        eq_probe_exprs: &[(usize, &Expr, &Expr)],
1782        residual: &[&Expr],
1783        peer_mask: &Option<Vec<bool>>,
1784        right_arity: usize,
1785        combined_schema: &[ColumnSchema],
1786        ctx: &EvalContext,
1787        cancel: CancelToken<'_>,
1788    ) -> Result<(), EngineError> {
1789        // Build side: eager rows if the peer was materialised (pushed
1790        // predicate / non-table ref), otherwise the stored table read in
1791        // place (v7.31 — no full-table clone + null-out just to hash it).
1792        // v7.32 (P4 increment 2) — move the eager build side into an
1793        // Owned source instead of borrowing `peer`, so the deferred
1794        // output can outlive this stage. Probe and hash-build read the
1795        // local `rights_src`.
1796        // Phase C.3 step 2b — MVCC read gate for the hash build side.
1797        // `build_gate` carries the peer `Table` + its hot-row count when
1798        // the build source is backed by the hot tier (`Stored`, or the
1799        // hot prefix of `Mixed`); a build row `ri < hot_len` this
1800        // snapshot cannot see is skipped so it never enters a bucket.
1801        // `Owned` build rows were already materialised (and filtered)
1802        // upstream, so they carry no physical hot index and stay ungated;
1803        // cold rows (`ri >= hot_len` in `Mixed`) are frozen = visible.
1804        // No-op today: every hot header is frozen/committed-alive.
1805        let (rights_src, build_gate): (JoinSrc<'a>, Option<(&'a Table, usize)>) =
1806            match peer.eager_rows.take() {
1807                Some(rows) => (JoinSrc::Owned(rows), None),
1808                None => match peer
1809                    .join_table
1810                    .as_deref()
1811                    .and_then(|n| self.active_catalog().get(n))
1812                {
1813                    // v7.36 — cold-bearing peer hashes through `Mixed`.
1814                    // Unlike INL, hash build doesn't consume the
1815                    // locator's key; it iterates the source via
1816                    // `len()/get()` and indexes each row by its
1817                    // eq_pairs values — works correctly for ANY join
1818                    // column (PK or secondary). No PK constraint.
1819                    Some(t) if t.has_cold_rows_fast() => {
1820                        let (cold, map) = crate::constraints::iter_cold_rows_with_locator_map(
1821                            self.active_catalog(),
1822                            t,
1823                        );
1824                        let hot = t.rows();
1825                        let hot_len = hot.len();
1826                        (
1827                            JoinSrc::Mixed {
1828                                hot,
1829                                cold,
1830                                cold_locator_map: map,
1831                            },
1832                            Some((t, hot_len)),
1833                        )
1834                    }
1835                    Some(t) => (JoinSrc::Stored(t.rows()), Some((t, t.rows().len()))),
1836                    None => (JoinSrc::Owned(Vec::new()), None),
1837                },
1838            };
1839        let scan_snapshot = self.current_snapshot();
1840        let n_rights = rights_src.len();
1841        // v7.29 - hashbrown over BTreeMap: the ordered map paid
1842        // O(log n) string comparisons per insert/probe (24k-row build
1843        // sides spent ~100 ms in it).
1844        // v7.36 (perf — mailrs Phase 1) — type-specialised i64 hash
1845        // table when the join keys are a single integer column on
1846        // both sides (the overwhelming case: FK to PK joins, ID
1847        // lookups). Skips the `encode_one` → String round-trip
1848        // entirely; the hash key is the i64 itself. For count_messages
1849        // / inbox / contacts / list_categories the eq_pair is
1850        // `(messages.mailbox_id, mailboxes.id)` both BigInt.
1851        let int_keyed = eq_exprs.is_empty()
1852            && eq_probe_exprs.is_empty()
1853            && eq_pairs.len() == 1
1854            && matches!(
1855                combined_schema[eq_pairs[0].0].ty,
1856                spg_storage::DataType::BigInt
1857                    | spg_storage::DataType::Int
1858                    | spg_storage::DataType::SmallInt
1859            )
1860            && {
1861                let peer_col_ty = peer.cols.get(eq_pairs[0].1).map(|c| c.ty);
1862                matches!(
1863                    peer_col_ty,
1864                    Some(
1865                        spg_storage::DataType::BigInt
1866                            | spg_storage::DataType::Int
1867                            | spg_storage::DataType::SmallInt
1868                    )
1869                )
1870            };
1871        // v7.39 (round 719) — a single COMPUTED key that is integer-only
1872        // takes the i64 lane too. `ON a.id = b.id + 1` used to pay three
1873        // taxes the plain-column key did not: a canonical-STRING hash
1874        // table (build + probe both encode), the conjunct re-verified in
1875        // `residual` against a fully materialised combined row per
1876        // matching pair (round 590's defence against key-encoding
1877        // ambiguity), and an interpreted eval per build row. On the
1878        // panel's 500k self-joins those were ~210-260 ms against PG's
1879        // ~40. For a native i64 key the ambiguity defence protects
1880        // nothing: key equality IS SQL `=` (NULLs never enter the
1881        // table), so the conjunct is dropped from residual below.
1882        let int_expr_keyed = eq_pairs.is_empty()
1883            && eq_probe_exprs.is_empty()
1884            && eq_exprs.len() == 1
1885            && matches!(
1886                combined_schema[eq_exprs[0].0].ty,
1887                spg_storage::DataType::BigInt
1888                    | spg_storage::DataType::Int
1889                    | spg_storage::DataType::SmallInt
1890            )
1891            && int_only_key_expr(eq_exprs[0].1, peer);
1892        // v7.39 (round 720) — the mirror lane: a single `<peer int
1893        // column> = <integer-only left expression>` key (the EXISTS
1894        // pull-up's shape). Build hashes the peer COLUMN (the plain
1895        // int_keyed build); the probe evaluates the left expression per
1896        // tuple straight off the join sources. Extraction already
1897        // guaranteed both sides integer-family.
1898        let int_probe_expr_keyed =
1899            eq_pairs.is_empty() && eq_exprs.is_empty() && eq_probe_exprs.len() == 1;
1900        // v7.39 (round 732) — TWO integer keys pack into one i128 (exact,
1901        // no collision): the EXISTS pull-up's mixed shape `ON b.g = a.g
1902        // AND b.id = a.id + 3` ran the canonical-STRING lane, encoding
1903        // 500k build and 500k probe keys. Any combination of plain int
1904        // pairs and int probe-exprs totalling two qualifies; eq_exprs
1905        // (peer-side computed) stay out — their build half evaluates per
1906        // peer row and is already covered by the single-key lane.
1907        let int2_keyed = eq_exprs.is_empty()
1908            && eq_pairs.len() + eq_probe_exprs.len() == 2
1909            && !eq_probe_exprs.is_empty()
1910            && eq_pairs.iter().all(|(l, r)| {
1911                matches!(
1912                    combined_schema[*l].ty,
1913                    spg_storage::DataType::BigInt
1914                        | spg_storage::DataType::Int
1915                        | spg_storage::DataType::SmallInt
1916                ) && matches!(
1917                    peer.cols.get(*r).map(|c| c.ty),
1918                    Some(
1919                        spg_storage::DataType::BigInt
1920                            | spg_storage::DataType::Int
1921                            | spg_storage::DataType::SmallInt
1922                    )
1923                )
1924            });
1925        // The residual set the matching pairs actually re-check: the
1926        // int-keyed computed conjunct comes out; everything else stays.
1927        let residual: Vec<&Expr> = if int_expr_keyed {
1928            residual
1929                .iter()
1930                .copied()
1931                .filter(|r| !core::ptr::eq(*r, eq_exprs[0].2))
1932                .collect()
1933        } else if int_probe_expr_keyed {
1934            residual
1935                .iter()
1936                .copied()
1937                .filter(|r| !core::ptr::eq(*r, eq_probe_exprs[0].2))
1938                .collect()
1939        } else if !eq_probe_exprs.is_empty() {
1940            // v7.39 (round 732) — the mixed form (`ON b.g = a.g AND
1941            // b.id = a.id + 3`, the EXISTS pull-up's two-conjunct
1942            // shape) runs the STRING lane with the probe-expr's value
1943            // in the composite key — and still re-verified that
1944            // conjunct against a fully materialised combined row per
1945            // matching pair. The probe-expr key halves are
1946            // integer-only by extraction and the canonical integer
1947            // encoding is exact (`n{n}|`), so key equality IS the
1948            // conjunct: drop it from residual, same argument as the
1949            // i64 lanes.
1950            residual
1951                .iter()
1952                .copied()
1953                .filter(|r| !eq_probe_exprs.iter().any(|(_, _, c)| core::ptr::eq(*r, *c)))
1954                .collect()
1955        } else {
1956            residual.to_vec()
1957        };
1958        // v7.39 (round 745) — a residual conjunct that reads ONLY peer
1959        // columns filters the BUILD side up front instead of re-checking
1960        // every matched pair: `JOIN d b ON a.id = b.id WHERE b.g = 7`
1961        // built a 500k-row hash table to match 5k drive rows and ran
1962        // `b.g = 7` per candidate. ON-clause semantics make this sound
1963        // for every join kind (a build row the predicate rejects can
1964        // never satisfy the ON, so its absence pads exactly the same);
1965        // WHERE-sourced peer predicates only reach here for INNER/CROSS
1966        // (the collector is kind-gated). Only compilable conjuncts move
1967        // — the interpreter path keeps its exact wording for the rest.
1968        let mut build_preds: Vec<eval::CompiledExpr> = Vec::new();
1969        let residual: Vec<&Expr> = {
1970            let peer_ctx_probe = EvalContext::new(&peer.cols, Some(peer.alias.as_str()));
1971            residual
1972                .iter()
1973                .copied()
1974                .filter(|r| {
1975                    let peer_only = {
1976                        let all = core::cell::Cell::new(true);
1977                        crate::expr_analysis::visit_expr_columns_and_subqueries(
1978                            r,
1979                            &mut |c| {
1980                                if Engine::peer_col_pos(&peer.alias, &peer.cols, c).is_none() {
1981                                    all.set(false);
1982                                }
1983                            },
1984                            &mut |_| {
1985                                all.set(false);
1986                            },
1987                        );
1988                        all.get()
1989                    };
1990                    if peer_only && eval::fully_compilable(r) && expr_mentions_a_column(r) {
1991                        build_preds.push(eval::compile_expr(r, &peer_ctx_probe));
1992                        false
1993                    } else {
1994                        true
1995                    }
1996                })
1997                .collect()
1998        };
1999        let residual = residual.as_slice();
2000        let any_int_lane = int_keyed || int_expr_keyed || int_probe_expr_keyed;
2001        let mut int2_table: hashbrown::HashMap<i128, Bucket> =
2002            hashbrown::HashMap::with_capacity(if int2_keyed { n_rights } else { 0 });
2003        let mut table: hashbrown::HashMap<String, Bucket> =
2004            hashbrown::HashMap::with_capacity(if any_int_lane { 0 } else { n_rights });
2005        let mut int_table: hashbrown::HashMap<i64, Bucket> =
2006            hashbrown::HashMap::with_capacity(if any_int_lane { n_rights } else { 0 });
2007        // v7.39 (round 590) — a key expression names the peer's own columns
2008        // and is evaluated against one peer row, so it resolves against the
2009        // PEER's schema, not the combined one the residual uses.
2010        let peer_ctx = EvalContext {
2011            columns: &peer.cols,
2012            table_alias: Some(peer.alias.as_str()),
2013            ..ctx.clone()
2014        };
2015        let mut keybuf: Vec<&Value> = Vec::with_capacity(eq_pairs.len());
2016        let mut pred_stack: Vec<Value<'static>> = Vec::new();
2017        // v7.31 (perf 3e) — scratch key buffer: build inserts allocate
2018        // only on vacant, probes never allocate.
2019        let mut keystr = String::new();
2020        // v7.39 (round 746) — SHARDED build for the integer lanes. The
2021        // build walk (visibility gate + hoisted predicates + key
2022        // extraction) ran single-threaded over the whole peer — 25 ms of
2023        // a 500k predicate scan on the panel's filtered self-join while
2024        // PG runs a parallel scan. Shards produce local i64/i128 tables
2025        // merged in SHARD ORDER, which preserves ascending row order
2026        // inside every bucket — exactly what the serial walk produced,
2027        // so match emission order is unchanged. String-lane and Mixed
2028        // (cold-bearing) builds stay serial.
2029        let mut built_parallel = false;
2030        if (any_int_lane || int2_keyed)
2031            && n_rights >= crate::PARALLEL_MIN_ROWS
2032            && !matches!(rights_src, JoinSrc::Mixed { .. })
2033            && let Some(r) = self.parallel_runner.0.as_deref()
2034        {
2035            struct ShardTables {
2036                t64: hashbrown::HashMap<i64, Bucket>,
2037                t128: hashbrown::HashMap<i128, Bucket>,
2038            }
2039            type ShardOut = Result<ShardTables, EngineError>;
2040            let n_shards = (n_rights / crate::PARALLEL_MIN_ROWS).clamp(2, 8);
2041            let chunk = n_rights.div_ceil(n_shards);
2042            let rights_ref = &rights_src;
2043            let preds_ref = &build_preds;
2044            let peer_cols = &peer.cols;
2045            let peer_alias_s = peer.alias.as_str();
2046            let mysql = ctx.mysql_dialect;
2047            let style = ctx.render_style;
2048            let cat = ctx.catalog;
2049            let eq_pairs_ref = eq_pairs;
2050            let eq_exprs_ref = eq_exprs;
2051            let eq_probe_ref = eq_probe_exprs;
2052            let results = r.run_shards(n_shards, &|si| {
2053                let lo = si * chunk;
2054                let hi = ((si + 1) * chunk).min(n_rights);
2055                let mut sctx = EvalContext::new(peer_cols, Some(peer_alias_s));
2056                sctx.mysql_dialect = mysql;
2057                sctx.render_style = style;
2058                let sctx = match cat {
2059                    Some(c) => sctx.with_catalog(c),
2060                    None => sctx,
2061                };
2062                let mut stack: Vec<Value<'static>> = Vec::new();
2063                let mut out = ShardTables {
2064                    t64: hashbrown::HashMap::new(),
2065                    t128: hashbrown::HashMap::new(),
2066                };
2067                let run = || -> ShardOut {
2068                    let mut out = out;
2069                    'srows: for ri in lo..hi {
2070                        if let Some((gt, hot_len)) = build_gate
2071                            && ri < hot_len
2072                            && !gt.is_row_visible(ri, &scan_snapshot)
2073                        {
2074                            continue;
2075                        }
2076                        let Some(right) = rights_ref.get(ri) else {
2077                            continue;
2078                        };
2079                        for c in preds_ref.iter() {
2080                            let v = eval::eval_compiled(c, right, &sctx, &mut stack)
2081                                .map_err(EngineError::Eval)?;
2082                            if !crate::eval::predicate_is_true(&v, "JOIN/ON", mysql)? {
2083                                continue 'srows;
2084                            }
2085                        }
2086                        if int2_keyed {
2087                            let mut parts = [0i64; 2];
2088                            let mut pi = 0;
2089                            for (_, rpos) in eq_pairs_ref {
2090                                match right.values.get(*rpos) {
2091                                    Some(Value::BigInt(n)) => parts[pi] = *n,
2092                                    Some(Value::Int(n)) => parts[pi] = i64::from(*n),
2093                                    Some(Value::SmallInt(n)) => parts[pi] = i64::from(*n),
2094                                    _ => continue 'srows,
2095                                }
2096                                pi += 1;
2097                            }
2098                            for (p, _, _) in eq_probe_ref {
2099                                match right.values.get(*p) {
2100                                    Some(Value::BigInt(n)) => parts[pi] = *n,
2101                                    Some(Value::Int(n)) => parts[pi] = i64::from(*n),
2102                                    Some(Value::SmallInt(n)) => parts[pi] = i64::from(*n),
2103                                    _ => continue 'srows,
2104                                }
2105                                pi += 1;
2106                            }
2107                            let key = ((parts[0] as i128) << 64) | (parts[1] as u64 as i128);
2108                            match out.t128.entry(key) {
2109                                hashbrown::hash_map::Entry::Occupied(mut o) => o.get_mut().push(ri),
2110                                hashbrown::hash_map::Entry::Vacant(v) => {
2111                                    v.insert(Bucket::One(ri));
2112                                }
2113                            }
2114                            continue;
2115                        }
2116                        let key: i64 = if !eq_pairs_ref.is_empty() || !eq_probe_ref.is_empty() {
2117                            let rpos = if eq_probe_ref.is_empty() {
2118                                eq_pairs_ref[0].1
2119                            } else {
2120                                eq_probe_ref[0].0
2121                            };
2122                            match right.values.get(rpos) {
2123                                Some(Value::BigInt(n)) => *n,
2124                                Some(Value::Int(n)) => i64::from(*n),
2125                                Some(Value::SmallInt(n)) => i64::from(*n),
2126                                _ => continue 'srows,
2127                            }
2128                        } else {
2129                            match eval::eval_expr(eq_exprs_ref[0].1, right, &sctx)
2130                                .map_err(EngineError::Eval)?
2131                            {
2132                                Value::BigInt(n) => n,
2133                                Value::Int(n) => i64::from(n),
2134                                Value::SmallInt(n) => i64::from(n),
2135                                _ => continue 'srows,
2136                            }
2137                        };
2138                        match out.t64.entry(key) {
2139                            hashbrown::hash_map::Entry::Occupied(mut o) => o.get_mut().push(ri),
2140                            hashbrown::hash_map::Entry::Vacant(v) => {
2141                                v.insert(Bucket::One(ri));
2142                            }
2143                        }
2144                    }
2145                    Ok(out)
2146                };
2147                alloc::boxed::Box::new(run())
2148            });
2149            let mut ok = true;
2150            let mut shard_tables: Vec<ShardTables> = Vec::with_capacity(n_shards);
2151            let mut first_err: Option<EngineError> = None;
2152            for boxed in results {
2153                match boxed.downcast::<ShardOut>() {
2154                    Ok(sh) => match *sh {
2155                        Ok(t) => shard_tables.push(t),
2156                        Err(e) => {
2157                            ok = false;
2158                            if first_err.is_none() {
2159                                first_err = Some(e);
2160                            }
2161                        }
2162                    },
2163                    Err(_) => ok = false,
2164                }
2165            }
2166            if let Some(e) = first_err {
2167                return Err(e);
2168            }
2169            if ok {
2170                for t in shard_tables {
2171                    for (k, b) in t.t64 {
2172                        match int_table.entry(k) {
2173                            hashbrown::hash_map::Entry::Occupied(mut o) => {
2174                                for ri in b.as_slice() {
2175                                    o.get_mut().push(*ri);
2176                                }
2177                            }
2178                            hashbrown::hash_map::Entry::Vacant(v) => {
2179                                v.insert(b);
2180                            }
2181                        }
2182                    }
2183                    for (k, b) in t.t128 {
2184                        match int2_table.entry(k) {
2185                            hashbrown::hash_map::Entry::Occupied(mut o) => {
2186                                for ri in b.as_slice() {
2187                                    o.get_mut().push(*ri);
2188                                }
2189                            }
2190                            hashbrown::hash_map::Entry::Vacant(v) => {
2191                                v.insert(b);
2192                            }
2193                        }
2194                    }
2195                }
2196                built_parallel = true;
2197            }
2198        }
2199        'build: for ri in 0..n_rights {
2200            if built_parallel {
2201                break;
2202            }
2203            if let Some((gt, hot_len)) = build_gate
2204                && ri < hot_len
2205                && !gt.is_row_visible(ri, &scan_snapshot)
2206            {
2207                continue;
2208            }
2209            let Some(right) = rights_src.get(ri) else {
2210                continue;
2211            };
2212            // v7.39 (round 745) — the hoisted peer-only predicates.
2213            if !build_preds.is_empty() {
2214                let mut keep = true;
2215                for c in &build_preds {
2216                    let v = eval::eval_compiled(c, right, &peer_ctx, &mut pred_stack)
2217                        .map_err(EngineError::Eval)?;
2218                    if !crate::eval::predicate_is_true(&v, "JOIN/ON", ctx.mysql_dialect)? {
2219                        keep = false;
2220                        break;
2221                    }
2222                }
2223                if !keep {
2224                    continue 'build;
2225                }
2226            }
2227            if int2_keyed {
2228                // Key parts in a FIXED order: plain pairs first, then
2229                // probe-exprs — the probe reads them the same way.
2230                let mut parts = [0i64; 2];
2231                let mut pi = 0;
2232                let mut null_key = false;
2233                for (_, rpos) in eq_pairs {
2234                    match right.values.get(*rpos) {
2235                        Some(Value::BigInt(n)) => parts[pi] = *n,
2236                        Some(Value::Int(n)) => parts[pi] = i64::from(*n),
2237                        Some(Value::SmallInt(n)) => parts[pi] = i64::from(*n),
2238                        _ => {
2239                            null_key = true;
2240                            break;
2241                        }
2242                    }
2243                    pi += 1;
2244                }
2245                if !null_key {
2246                    for (p, _, _) in eq_probe_exprs {
2247                        match right.values.get(*p) {
2248                            Some(Value::BigInt(n)) => parts[pi] = *n,
2249                            Some(Value::Int(n)) => parts[pi] = i64::from(*n),
2250                            Some(Value::SmallInt(n)) => parts[pi] = i64::from(*n),
2251                            _ => {
2252                                null_key = true;
2253                                break;
2254                            }
2255                        }
2256                        pi += 1;
2257                    }
2258                }
2259                if null_key {
2260                    continue 'build;
2261                }
2262                let key = ((parts[0] as i128) << 64) | (parts[1] as u64 as i128);
2263                match int2_table.entry(key) {
2264                    hashbrown::hash_map::Entry::Occupied(mut o) => o.get_mut().push(ri),
2265                    hashbrown::hash_map::Entry::Vacant(v) => {
2266                        v.insert(Bucket::One(ri));
2267                    }
2268                }
2269                continue;
2270            }
2271            if any_int_lane {
2272                let key = if int_keyed || int_probe_expr_keyed {
2273                    // Plain-column build: the key column is the eq_pair's
2274                    // right side, or the mirror lane's peer column.
2275                    let rpos = if int_keyed {
2276                        eq_pairs[0].1
2277                    } else {
2278                        eq_probe_exprs[0].0
2279                    };
2280                    match right.values.get(rpos) {
2281                        Some(Value::BigInt(n)) => *n,
2282                        Some(Value::Int(n)) => i64::from(*n),
2283                        Some(Value::SmallInt(n)) => i64::from(*n),
2284                        _ => continue 'build,
2285                    }
2286                } else {
2287                    // Computed integer key: evaluate against the peer
2288                    // row. NULL joins nothing (SQL `=`); any non-integer
2289                    // value cannot happen under `int_only_key_expr`, and
2290                    // an arithmetic error (overflow) propagates, as it
2291                    // does on every other evaluation path.
2292                    match eval::eval_expr(eq_exprs[0].1, right, &peer_ctx)
2293                        .map_err(EngineError::Eval)?
2294                    {
2295                        Value::BigInt(n) => n,
2296                        Value::Int(n) => i64::from(n),
2297                        Value::SmallInt(n) => i64::from(n),
2298                        _ => continue 'build,
2299                    }
2300                };
2301                // v7.37.x (docker-fair NOTEX hash-build attack) — most
2302                // FK-to-PK joins are unique on the build side, so the
2303                // bucket is a one-element Vec. `or_default()` lands as
2304                // a 0-cap Vec then the push grows it through 1 → 4
2305                // (two allocs); pre-sizing to 1 cuts those to one.
2306                // For the NOTEX 12.5 k-row build side this saves
2307                // ~12.5 k × ~100 ns ≈ 1.25 ms per query.
2308                match int_table.entry(key) {
2309                    hashbrown::hash_map::Entry::Occupied(mut o) => o.get_mut().push(ri),
2310                    hashbrown::hash_map::Entry::Vacant(v) => {
2311                        v.insert(Bucket::One(ri));
2312                    }
2313                }
2314                continue;
2315            }
2316            keybuf.clear();
2317            for (_, rpos) in eq_pairs {
2318                match right.values.get(*rpos) {
2319                    Some(v) if !matches!(v, Value::Null) => keybuf.push(v),
2320                    _ => continue 'build,
2321                }
2322            }
2323            aggregate::encode_key_refs_into(&keybuf, &mut keystr);
2324            // v7.39 (round 590) — then the computed components, in the order
2325            // the probe will read them. A NULL never matches under `=`, so a
2326            // row whose key expression is NULL joins nothing and is left out
2327            // of the table entirely.
2328            for (_, e, _) in eq_exprs {
2329                let v = eval::eval_expr(e, right, &peer_ctx).map_err(EngineError::Eval)?;
2330                if matches!(v, Value::Null) {
2331                    continue 'build;
2332                }
2333                aggregate::push_canonical_key(&mut keystr, &v);
2334            }
2335            // v7.39 (round 720) — then the PROBE-side computed keys'
2336            // build halves: the peer COLUMN each one equates to. Without
2337            // this, `ON b.g = a.g AND b.id = a.id + 1` hashed on `g`
2338            // alone and re-verified the second conjunct per candidate
2339            // pair — the exact quadratic round 590 fixed, resurrected in
2340            // mirror image (measured: a 500k self-join never returned).
2341            // The canonical integer encoding is type-agnostic (`n{n}|`),
2342            // so the probe's evaluated i64 meets the column's own value.
2343            for (p, _, _) in eq_probe_exprs {
2344                match right.values.get(*p) {
2345                    Some(v) if !matches!(v, Value::Null) => {
2346                        aggregate::push_canonical_key(&mut keystr, v);
2347                    }
2348                    _ => continue 'build,
2349                }
2350            }
2351            match table.get_mut(keystr.as_str()) {
2352                Some(b) => b.push(ri),
2353                None => {
2354                    table.insert(keystr.clone(), Bucket::One(ri));
2355                }
2356            }
2357        }
2358        let mut next: Vec<usize> = Vec::new();
2359        // v7.37.16 — RIGHT / FULL OUTER: track which peer (build-side)
2360        // rows joined with at least one drive tuple so the unmatched
2361        // ones can be emitted (NULL-filled left) after the probe loop.
2362        // Empty (unallocated) for INNER / LEFT — no per-row cost there.
2363        let track_right = matches!(peer.kind, JoinKind::Right | JoinKind::FullOuter);
2364        let mut peer_matched: Vec<bool> = if track_right {
2365            alloc::vec![false; n_rights]
2366        } else {
2367            Vec::new()
2368        };
2369        let mut probebuf: Vec<&Value> = Vec::with_capacity(eq_pairs.len());
2370        for tuple in pipe.working.chunks(pipe.stride) {
2371            cancel.check()?;
2372            let mut left_matched = false;
2373            let mut left_has_null = false;
2374            let int2_probe_key: Option<i128> = if int2_keyed {
2375                let mut parts = [0i64; 2];
2376                let mut pi = 0;
2377                let mut nul = false;
2378                for (lpos, _) in eq_pairs {
2379                    match tuple_value(&pipe.sources, &pipe.offsets, tuple, *lpos) {
2380                        Some(Value::BigInt(n)) => parts[pi] = *n,
2381                        Some(Value::Int(n)) => parts[pi] = i64::from(*n),
2382                        Some(Value::SmallInt(n)) => parts[pi] = i64::from(*n),
2383                        _ => {
2384                            nul = true;
2385                            break;
2386                        }
2387                    }
2388                    pi += 1;
2389                }
2390                if !nul {
2391                    for (_, e, _) in eq_probe_exprs {
2392                        match eval_int_only_probe(
2393                            e,
2394                            &combined_schema[..pipe.consumed_cols],
2395                            &pipe.sources,
2396                            &pipe.offsets,
2397                            tuple,
2398                        )? {
2399                            Some(k) => parts[pi] = k,
2400                            None => {
2401                                nul = true;
2402                                break;
2403                            }
2404                        }
2405                        pi += 1;
2406                    }
2407                }
2408                if nul {
2409                    left_has_null = true;
2410                    None
2411                } else {
2412                    Some(((parts[0] as i128) << 64) | (parts[1] as u64 as i128))
2413                }
2414            } else {
2415                None
2416            };
2417            let int_probe_key: Option<i64> = if int2_keyed {
2418                None
2419            } else if int_probe_expr_keyed {
2420                match eval_int_only_probe(
2421                    eq_probe_exprs[0].1,
2422                    &combined_schema[..pipe.consumed_cols],
2423                    &pipe.sources,
2424                    &pipe.offsets,
2425                    tuple,
2426                )? {
2427                    Some(k) => Some(k),
2428                    None => {
2429                        left_has_null = true;
2430                        None
2431                    }
2432                }
2433            } else if any_int_lane {
2434                let lpos = if int_keyed {
2435                    eq_pairs[0].0
2436                } else {
2437                    eq_exprs[0].0
2438                };
2439                match tuple_value(&pipe.sources, &pipe.offsets, tuple, lpos) {
2440                    Some(Value::BigInt(n)) => Some(*n),
2441                    Some(Value::Int(n)) => Some(i64::from(*n)),
2442                    Some(Value::SmallInt(n)) => Some(i64::from(*n)),
2443                    _ => {
2444                        left_has_null = true;
2445                        None
2446                    }
2447                }
2448            } else {
2449                probebuf.clear();
2450                for (lpos, _) in eq_pairs {
2451                    match tuple_value(&pipe.sources, &pipe.offsets, tuple, *lpos) {
2452                        Some(v) if !matches!(v, Value::Null) => probebuf.push(v),
2453                        _ => {
2454                            left_has_null = true;
2455                            break;
2456                        }
2457                    }
2458                }
2459                if !left_has_null {
2460                    aggregate::encode_key_refs_into(&probebuf, &mut keystr);
2461                    for (lpos, _, _) in eq_exprs {
2462                        match tuple_value(&pipe.sources, &pipe.offsets, tuple, *lpos) {
2463                            Some(v) if !matches!(v, Value::Null) => {
2464                                aggregate::push_canonical_key(&mut keystr, v)
2465                            }
2466                            _ => {
2467                                left_has_null = true;
2468                                break;
2469                            }
2470                        }
2471                    }
2472                }
2473                if !left_has_null {
2474                    for (_, e, _) in eq_probe_exprs {
2475                        match eval_int_only_probe(
2476                            e,
2477                            &combined_schema[..pipe.consumed_cols],
2478                            &pipe.sources,
2479                            &pipe.offsets,
2480                            tuple,
2481                        )? {
2482                            Some(k) => {
2483                                aggregate::push_canonical_key(&mut keystr, &Value::BigInt(k))
2484                            }
2485                            None => {
2486                                left_has_null = true;
2487                                break;
2488                            }
2489                        }
2490                    }
2491                }
2492                None
2493            };
2494            let cands_opt: Option<&Bucket> = if left_has_null {
2495                None
2496            } else if int2_keyed {
2497                int2_table.get(&int2_probe_key.unwrap())
2498            } else if any_int_lane {
2499                int_table.get(&int_probe_key.unwrap())
2500            } else {
2501                table.get(keystr.as_str())
2502            };
2503            if let Some(cands) = cands_opt {
2504                for &ri in cands.as_slice() {
2505                    let keep = if residual.is_empty() {
2506                        true
2507                    } else {
2508                        let right = rights_src.get(ri).expect("hash candidate row");
2509                        let mut combined_vals = materialise_tuple_vals(
2510                            &pipe.sources,
2511                            &pipe.widths,
2512                            &pipe.masks,
2513                            tuple,
2514                            pipe.consumed_cols + right_arity,
2515                        );
2516                        extend_masked(&mut combined_vals, right, peer_mask.as_deref());
2517                        let combined = Row::new(combined_vals);
2518                        let mut ok = true;
2519                        for r in residual {
2520                            let cond =
2521                                self.eval_expr_with_correlated(r, &combined, ctx, cancel, None)?;
2522                            if !crate::eval::predicate_is_true(&cond, "JOIN/ON", ctx.mysql_dialect)?
2523                            {
2524                                ok = false;
2525                                break;
2526                            }
2527                        }
2528                        ok
2529                    };
2530                    if keep {
2531                        next.extend_from_slice(tuple);
2532                        next.push(ri);
2533                        left_matched = true;
2534                        if track_right {
2535                            peer_matched[ri] = true;
2536                        }
2537                        // v7.39 (round 725) — SEMI: one pairing per drive
2538                        // row is the answer; the rest of the bucket is
2539                        // EXISTS's dead work.
2540                        if matches!(peer.kind, JoinKind::Semi) {
2541                            break;
2542                        }
2543                    }
2544                }
2545            }
2546            // LEFT (and FULL OUTER) keep unmatched drive rows with a
2547            // NULL-filled peer (`usize::MAX` sentinel → NULL columns).
2548            if !left_matched && matches!(peer.kind, JoinKind::Left | JoinKind::FullOuter) {
2549                next.extend_from_slice(tuple);
2550                next.push(usize::MAX);
2551            }
2552        }
2553        // v7.37.16 — RIGHT / FULL OUTER: append the peer rows that no
2554        // drive tuple matched, NULL-filling every prior source column
2555        // (a `usize::MAX` sentinel for each of the `stride` drive slots)
2556        // and carrying the real peer row index. Same visibility gate as
2557        // the build loop so an invisible / NULL-key peer row is emitted
2558        // once and only when it truly exists.
2559        if track_right {
2560            for ri in 0..n_rights {
2561                if peer_matched[ri] {
2562                    continue;
2563                }
2564                if let Some((gt, hot_len)) = build_gate
2565                    && ri < hot_len
2566                    && !gt.is_row_visible(ri, &scan_snapshot)
2567                {
2568                    continue;
2569                }
2570                if rights_src.get(ri).is_none() {
2571                    continue;
2572                }
2573                for _ in 0..pipe.stride {
2574                    next.push(usize::MAX);
2575                }
2576                next.push(ri);
2577            }
2578        }
2579        pipe.advance(next, rights_src, peer_mask.clone(), right_arity);
2580        debug_assert!(pipe.consumed_cols <= combined_schema.len());
2581        Ok(())
2582    }
2583
2584    /// Nested-loop join stage — the fallback for LATERAL peers and
2585    /// non-equi ON. A deferred plain-table peer materialises here
2586    /// (pruned), since every (left, right) pair gets evaluated anyway.
2587    #[allow(clippy::too_many_arguments)]
2588    fn join_stage_nested<'a, 'p>(
2589        &'a self,
2590        pipe: &mut JoinPipeline<'a>,
2591        peer: &mut JoinedPeer<'p>,
2592        right_arity: usize,
2593        combined_schema: &[ColumnSchema],
2594        ctx: &EvalContext,
2595        cancel: CancelToken<'_>,
2596        needed: Option<&alloc::collections::BTreeSet<(String, String)>>,
2597        budget: &mut ByteBudget,
2598    ) -> Result<(), EngineError> {
2599        let lazy_rows: Option<Vec<Row<'static>>> =
2600            if peer.eager_rows.is_none() && peer.lateral.is_none() {
2601                let tname = peer.join_table.as_deref().unwrap_or("");
2602                // v7.37.15 Phase B — visibility-gated nested-loop
2603                // fallback peer scan.
2604                let snap = self.current_snapshot();
2605                let mut rows: Vec<Row<'static>> = self
2606                    .active_catalog()
2607                    .get(tname)
2608                    .map(|t| t.scan_visible(&snap).map(|(_, r)| r.clone()).collect())
2609                    .unwrap_or_default();
2610                // v7.36 — nested-loop fallback materialises the peer
2611                // into `lazy_rows`. Append cold-tier rows so the fall-
2612                // back stays correct after the force-eager-when-cold
2613                // guard was lifted in `build_join_peers`.
2614                if let Some(t) = self.active_catalog().get(tname)
2615                    && t.has_cold_rows_fast()
2616                {
2617                    rows.extend(crate::constraints::iter_cold_rows_of_parent(
2618                        self.active_catalog(),
2619                        t,
2620                    ));
2621                }
2622                if let Some(needed) = needed {
2623                    Self::null_out_unreferenced(&mut rows, &peer.cols, &peer.alias, needed);
2624                }
2625                budget.charge(approx_rows_bytes(&rows))?;
2626                Some(rows)
2627            } else {
2628                None
2629            };
2630        // Lateral results are per-outer-row, so matched right rows persist
2631        // in a stage arena the tuples can index.
2632        let mut arena: Vec<Row<'static>> = Vec::new();
2633        let rights_eager: Option<&[Row<'static>]> =
2634            peer.eager_rows.as_deref().or(lazy_rows.as_deref());
2635        let mut next: Vec<usize> = Vec::new();
2636        let right_or_full = matches!(peer.kind, JoinKind::Right | JoinKind::FullOuter);
2637        // v7.37.16 — RIGHT / FULL OUTER over a *derived-table* right
2638        // operand (VALUES / non-correlated subquery). `build_join_peers`
2639        // routes every derived table through the "lateral" branch even
2640        // when it is not correlated; materialise it ONCE here into a
2641        // fixed row set so the unmatched-peer rows can be enumerated. A
2642        // truly correlated peer would give per-left-row-varying rows, but
2643        // PG rejects RIGHT/FULL LATERAL, so a single NULL-outer
2644        // materialisation is the correct fixed set. Also handles the
2645        // empty-drive case (no left tuples → every peer row unmatched).
2646        let lateral_fixed: Option<Vec<Row<'static>>> =
2647            if right_or_full && let Some(inner) = peer.lateral {
2648                // Materialise the derived table directly (no outer-column
2649                // substitution): a RIGHT/FULL peer must be non-correlated
2650                // (PG rejects RIGHT/FULL LATERAL), so its rows are the same
2651                // for every drive row and independent of the outer context.
2652                // Use the union-aware entry: a multi-row `VALUES (…),(…)` is
2653                // stored as a head SELECT + `stmt.unions` tails, so the bare
2654                // (non-union) executor would return only the first row.
2655                match self.exec_select_cancel(inner, cancel)? {
2656                    QueryResult::Rows { rows, .. } => Some(rows),
2657                    _ => {
2658                        return Err(EngineError::Unsupported(
2659                            "derived-table join operand must be a SELECT".into(),
2660                        ));
2661                    }
2662                }
2663            } else {
2664                None
2665            };
2666        // A fixed peer-row index space exists for the non-lateral eager
2667        // path OR the just-materialised `lateral_fixed` set. Both let
2668        // RIGHT / FULL OUTER track which peer rows matched and emit the
2669        // unmatched ones (NULL-filled left) after the loop.
2670        let track_right = right_or_full && (peer.lateral.is_none() || lateral_fixed.is_some());
2671        let fixed_peer_len = match &lateral_fixed {
2672            Some(f) => f.len(),
2673            None => rights_eager.map(<[_]>::len).unwrap_or(0),
2674        };
2675        let mut peer_matched: Vec<bool> = if track_right {
2676            alloc::vec![false; fixed_peer_len]
2677        } else {
2678            Vec::new()
2679        };
2680        for tuple in pipe.working.chunks(pipe.stride) {
2681            cancel.check()?;
2682            let mut left_matched = false;
2683            let left_vals = materialise_tuple_vals(
2684                &pipe.sources,
2685                &pipe.widths,
2686                &pipe.masks,
2687                tuple,
2688                pipe.consumed_cols,
2689            );
2690            let per_left_rrows: Cow<'_, [Row]> = match (&lateral_fixed, peer.lateral) {
2691                // RIGHT/FULL derived-table peer — the single fixed set.
2692                (Some(fixed), _) => Cow::Borrowed(fixed.as_slice()),
2693                (None, Some(inner)) => {
2694                    // Substitute outer columns and run the inner SELECT
2695                    // against the current left row's slice of the
2696                    // combined schema.
2697                    let outer_schema = &combined_schema[..pipe.consumed_cols];
2698                    let left_row = Row::new(left_vals.clone());
2699                    let rows =
2700                        self.materialise_lateral_for_outer(inner, outer_schema, &left_row)?;
2701                    Cow::Owned(rows)
2702                }
2703                (None, None) => Cow::Borrowed(rights_eager.expect("non-lateral peer eager")),
2704            };
2705            for (ri, right) in per_left_rrows.as_ref().iter().enumerate() {
2706                let mut combined_vals = left_vals.clone();
2707                combined_vals.extend(right.values.iter().cloned());
2708                let combined = Row::new(combined_vals);
2709                let keep = if let Some(on_expr) = peer.on {
2710                    // v7.24.1 — correlated-aware (subqueries in ON
2711                    // referencing earlier join columns).
2712                    let cond =
2713                        self.eval_expr_with_correlated(on_expr, &combined, ctx, cancel, None)?;
2714                    crate::eval::predicate_is_true(&cond, "JOIN/ON", ctx.mysql_dialect)?
2715                } else {
2716                    true
2717                };
2718                if keep {
2719                    next.extend_from_slice(tuple);
2720                    if peer.lateral.is_some() && lateral_fixed.is_none() {
2721                        // Correlated / INNER-LEFT lateral: per-outer-row
2722                        // arena (rows vary per left tuple).
2723                        let mut cv = combined.values;
2724                        let rv = cv.split_off(left_vals.len());
2725                        arena.push(Row::new(rv));
2726                        next.push(arena.len() - 1);
2727                    } else {
2728                        // Fixed peer index space (non-lateral eager, or
2729                        // the RIGHT/FULL `lateral_fixed` set).
2730                        next.push(ri);
2731                        if track_right {
2732                            peer_matched[ri] = true;
2733                        }
2734                    }
2735                    left_matched = true;
2736                    // v7.39 (round 725) — SEMI keeps one pairing (see the
2737                    // hash stage; this loop is its safety net).
2738                    if matches!(peer.kind, JoinKind::Semi) {
2739                        break;
2740                    }
2741                }
2742            }
2743            if !left_matched && matches!(peer.kind, JoinKind::Left | JoinKind::FullOuter) {
2744                next.extend_from_slice(tuple);
2745                next.push(usize::MAX);
2746            }
2747        }
2748        // v7.37.16 — RIGHT / FULL OUTER: append unmatched peer rows with
2749        // a NULL-filled left (`usize::MAX` for each drive slot).
2750        if track_right {
2751            for (ri, matched) in peer_matched.iter().enumerate() {
2752                if *matched {
2753                    continue;
2754                }
2755                for _ in 0..pipe.stride {
2756                    next.push(usize::MAX);
2757                }
2758                next.push(ri);
2759            }
2760        }
2761        if next.len() / (pipe.stride + 1) > MAX_JOIN_INTERMEDIATE_ROWS {
2762            return Err(EngineError::Unsupported(alloc::format!(
2763                "join intermediate result exceeds {MAX_JOIN_INTERMEDIATE_ROWS} rows ({} so far) - add join predicates",
2764                next.len() / (pipe.stride + 1)
2765            )));
2766        }
2767        let source = if let Some(fixed) = lateral_fixed {
2768            // RIGHT/FULL derived-table peer — the fixed set is the source.
2769            JoinSrc::Owned(fixed)
2770        } else if peer.lateral.is_some() {
2771            JoinSrc::Owned(arena)
2772        } else if let Some(lz) = lazy_rows {
2773            JoinSrc::Owned(lz)
2774        } else {
2775            // v7.32 (P4 increment 2) — move (not borrow) the eager peer
2776            // rows; `rights_eager` has finished its nested-loop borrow.
2777            JoinSrc::Owned(peer.eager_rows.take().expect("non-lateral peer eager"))
2778        };
2779        // Fallback sources are pre-pruned (eager / lazy null-out) or
2780        // lateral projections; nothing left for a mask to drop.
2781        pipe.advance(next, source, None, right_arity);
2782        debug_assert!(pipe.consumed_cols <= combined_schema.len());
2783        Ok(())
2784    }
2785
2786    /// v7.24 (round-16 B) — final WHERE filter over the joined working
2787    /// set. The compiled path reads cells by reference through
2788    /// `RowRef::Tuple` (`eval_compiled_ref`) WITHOUT materialising a
2789    /// combined Row; only a correlated WHERE (subqueries) materialises,
2790    /// once, per surviving probe, through the memoized correlated-aware
2791    /// evaluator. Survivors are returned as their row-index tuples — the
2792    /// aggregate path borrows them, projection / window callers
2793    /// `materialise()`.
2794    fn filter_join_survivors(
2795        &self,
2796        pipe: &JoinPipeline<'_>,
2797        where_: Option<&Expr>,
2798        ctx: &EvalContext,
2799        cancel: CancelToken<'_>,
2800        budget: &mut ByteBudget,
2801    ) -> Result<Vec<usize>, EngineError> {
2802        // v7.37.x (mailrs Track A perf — paired with v7.37.15
2803        // pushdown-strip) — when every conjunct was pushed onto its
2804        // source (eager peer filter / primary index seek / join-stage
2805        // residual), `residual_where` is None and every joined tuple
2806        // is already a survivor. Skip the per-tuple eval-or-true loop:
2807        // budget-charge a single rectangular approximation of the
2808        // whole working set, then bulk-copy the tuple indices via
2809        // `to_vec()`. On the mailrs minimal 100k shape this turns a
2810        // 100 k-iter per-tuple loop into a single allocation +
2811        // memcpy.
2812        if where_.is_none() {
2813            // Approximate total bytes by per-row cost × row count
2814            // (mirrors what the per-tuple charge would sum). Empty
2815            // working set short-circuits to a no-op.
2816            let n_rows = if pipe.stride == 0 {
2817                0
2818            } else {
2819                pipe.working.len() / pipe.stride
2820            };
2821            if n_rows > 0 {
2822                let sample_tuple = &pipe.working[..pipe.stride];
2823                let per_tuple =
2824                    approx_tuple_bytes(&pipe.sources, &pipe.offsets, &pipe.masks, sample_tuple);
2825                budget.charge(per_tuple.saturating_mul(n_rows))?;
2826            }
2827            cancel.check()?;
2828            return Ok(pipe.working.clone());
2829        }
2830        let mut memo = memoize::MemoizeCache::default();
2831        let compiled_where: Option<eval::CompiledExpr> = where_
2832            .filter(|w| eval::fully_compilable(w))
2833            .map(|w| eval::compile_expr(w, ctx));
2834        let mut survivors: Vec<usize> = Vec::new();
2835        for tuple in pipe.working.chunks(pipe.stride) {
2836            let rr = RowRef::Tuple {
2837                sources: &pipe.sources,
2838                offsets: &pipe.offsets,
2839                pos_to_src: &pipe.pos_to_src,
2840                tuple,
2841            };
2842            // v7.37.9 T3 S2 — declare eval_stack inside the per-tuple
2843            // loop so its `'val` lifetime binds to `rr`'s local scope.
2844            // The Vec allocation per tuple is amortised by the row's
2845            // existing work; alternative (outer-scope Vec) hits the
2846            // lifetime-contamination wall under `'row: 'val`.
2847            let mut eval_stack: Vec<Value<'_>> = Vec::new();
2848            let pass = if let Some(cw) = &compiled_where {
2849                matches!(
2850                    eval::eval_compiled_ref(cw, rr, ctx, &mut eval_stack)
2851                        .map_err(EngineError::Eval)?,
2852                    Value::Bool(true)
2853                )
2854            } else if let Some(where_expr) = where_ {
2855                let row = rr.as_row();
2856                matches!(
2857                    self.eval_expr_with_correlated(where_expr, &row, ctx, cancel, Some(&mut memo))?,
2858                    Value::Bool(true)
2859                )
2860            } else {
2861                true
2862            };
2863            if !pass {
2864                continue;
2865            }
2866            // v7.30.3 byte budget — survivors hold 8 B row numbers, but
2867            // the live data they reference is what the meter must track;
2868            // `approx_tuple_bytes` sums it by reference (no clone),
2869            // mirroring the bytes the old materialised path charged.
2870            budget.charge(approx_tuple_bytes(
2871                &pipe.sources,
2872                &pipe.offsets,
2873                &pipe.masks,
2874                tuple,
2875            ))?;
2876            survivors.extend_from_slice(tuple);
2877        }
2878        Ok(survivors)
2879    }
2880
2881    /// v7.17.0 Phase 3.P0-41 — probe a LATERAL subquery's projection
2882    /// schema by running it once with a NULL-padded outer context.
2883    /// The probe never materialises real outer rows; it just executes
2884    /// the inner SELECT with `outer_alias.col` references substituted
2885    /// to NULL so the projection's type inference is exercised.
2886    fn lateral_probe_schema(
2887        &self,
2888        inner: &SelectStatement,
2889    ) -> Result<Vec<ColumnSchema>, EngineError> {
2890        // Substitute every qualified column reference whose qualifier
2891        // does NOT match an in-subquery FROM alias with NULL. The
2892        // safest probe is to walk the inner SELECT and replace any
2893        // `<qual>.<col>` whose qual isn't bound inside the subquery
2894        // with a Null literal. For the v7.17 probe we just run the
2895        // unmodified subquery and surface the columns; if it fails
2896        // (e.g. references an outer column the probe can't resolve),
2897        // we synthesise a best-effort schema from the SELECT items
2898        // by inferring a single Text-typed column per projection.
2899        match self.execute_readonly_select_for_lateral_probe(inner) {
2900            Ok(QueryResult::Rows { columns, .. }) => Ok(columns),
2901            // Best-effort fallback: each SELECT item becomes a TEXT
2902            // column. Real schemas only differ when the inner SELECT
2903            // references outer columns at projection-time; those
2904            // queries surface via the substitution path during
2905            // per-row execution and still return the right values.
2906            _ => {
2907                // `SELECT * FROM <srf>(… outer.col …)` — the wrapped
2908                // correlated-SRF shape. The probe can't evaluate the
2909                // outer reference, but the SRF ref itself dictates
2910                // the schema: column-alias list first, then the
2911                // executor's natural defaults (alias / fn name), plus
2912                // the WITH ORDINALITY counter.
2913                if let [SelectItem::Wildcard] = inner.items.as_slice()
2914                    && let Some(from) = &inner.from
2915                    && from.joins.is_empty()
2916                    && (from.primary.unnest_expr.is_some()
2917                        || from.primary.generate_series_args.is_some())
2918                {
2919                    let t = &from.primary;
2920                    let elem_dtype = if t.generate_series_args.is_some() {
2921                        DataType::BigInt
2922                    } else {
2923                        DataType::Text
2924                    };
2925                    let first = t
2926                        .unnest_column_aliases
2927                        .first()
2928                        .cloned()
2929                        .or_else(|| t.alias.clone())
2930                        .unwrap_or_else(|| t.name.clone());
2931                    let mut out = alloc::vec![ColumnSchema::new(first, elem_dtype, true)];
2932                    if t.with_ordinality {
2933                        let ord = t
2934                            .unnest_column_aliases
2935                            .get(1)
2936                            .cloned()
2937                            .unwrap_or_else(|| "ordinality".to_string());
2938                        out.push(ColumnSchema::new(ord, DataType::BigInt, false));
2939                    }
2940                    return Ok(out);
2941                }
2942                // v7.39 (read01 round 69) — `SELECT * FROM <user fn>(…)`, which is
2943                // what a correlated `LATERAL f(t.c)` wraps into. A wildcard has no
2944                // name to give, so without this the column came back as `col0` and
2945                // the alias (`AS d`) resolved to nothing. Take the shape the
2946                // function DECLARES: `RETURNS TABLE(id int, v text)` names its
2947                // columns, and a `SETOF <scalar>` is one column named after the
2948                // call's alias.
2949                // v7.39 (round 205, JSON_TABLE) — a wrapped correlated
2950                // JSON_TABLE (`SELECT * FROM JSON_TABLE(t.col, …)`):
2951                // its column shape is STATIC (COLUMNS list), so infer
2952                // it directly without evaluating the doc (which still
2953                // references the outer column at schema time).
2954                if let Some(from) = &inner.from
2955                    && from.joins.is_empty()
2956                    && matches!(inner.items.as_slice(), [SelectItem::Wildcard])
2957                    && let Some(jt) = from.primary.json_table.as_deref()
2958                {
2959                    return Ok(crate::select::json_table_schema_pub(&jt.columns));
2960                }
2961                if let Some(from) = &inner.from
2962                    && from.joins.is_empty()
2963                    && matches!(inner.items.as_slice(), [SelectItem::Wildcard])
2964                    && let Some((fn_name, _)) = from.primary.table_fn_call.as_deref()
2965                {
2966                    let cat = self.active_catalog();
2967                    let overloads = cat.functions_named(fn_name);
2968                    if let Some(def) = overloads.first() {
2969                        let declared = def.returns.trim();
2970                        let upper = declared.to_ascii_uppercase();
2971                        if let Some(rest) = upper.strip_prefix("TABLE(") {
2972                            let _ = rest;
2973                            let raw = &declared["TABLE(".len()..declared.len() - 1];
2974                            let cols: Vec<ColumnSchema> = raw
2975                                .split(',')
2976                                .map(|decl| {
2977                                    let cname = decl.split_whitespace().next().unwrap_or("col");
2978                                    ColumnSchema::new(cname.to_string(), DataType::Text, true)
2979                                })
2980                                .collect();
2981                            return Ok(cols);
2982                        }
2983                        let cname = from
2984                            .primary
2985                            .alias
2986                            .clone()
2987                            .unwrap_or_else(|| fn_name.clone());
2988                        return Ok(alloc::vec![ColumnSchema::new(cname, DataType::Text, true)]);
2989                    }
2990                }
2991                let mut out: Vec<ColumnSchema> = Vec::new();
2992                for (i, item) in inner.items.iter().enumerate() {
2993                    let name = match item {
2994                        SelectItem::Expr { alias: Some(a), .. } => a.clone(),
2995                        SelectItem::Expr { expr, .. } => synth_lateral_col_name(expr, i),
2996                        SelectItem::Wildcard | SelectItem::QualifiedWildcard(_) => {
2997                            alloc::format!("col{i}")
2998                        }
2999                    };
3000                    out.push(ColumnSchema::new(name, DataType::Text, true));
3001                }
3002                Ok(out)
3003            }
3004        }
3005    }
3006
3007    /// v7.17.0 Phase 3.P0-41 — try the inner LATERAL subquery against
3008    /// the engine in read-only mode for schema-probe purposes. Failure
3009    /// is expected when the subquery references an outer column the
3010    /// probe can't resolve; the caller falls back to a best-effort
3011    /// schema based on the SELECT items.
3012    fn execute_readonly_select_for_lateral_probe(
3013        &self,
3014        inner: &SelectStatement,
3015    ) -> Result<QueryResult, EngineError> {
3016        self.exec_bare_select_cancel(inner, CancelToken::none())
3017    }
3018
3019    /// v7.17.0 Phase 3.P0-41 — materialise a LATERAL subquery's rows
3020    /// for one outer-row context. Walks the inner SELECT, replaces
3021    /// every `<outer_alias>.<col>` reference whose alias appears in
3022    /// the outer schema with the literal value from the outer row,
3023    /// then runs the rewritten SELECT against the engine.
3024    fn materialise_lateral_for_outer(
3025        &self,
3026        inner: &SelectStatement,
3027        outer_schema: &[ColumnSchema],
3028        outer_row: &Row<'static>,
3029    ) -> Result<Vec<Row<'static>>, EngineError> {
3030        let mut substituted = inner.clone();
3031        substitute_outer_columns_multi(&mut substituted, outer_row, outer_schema);
3032        let result = self.exec_bare_select_cancel(&substituted, CancelToken::none())?;
3033        match result {
3034            QueryResult::Rows { rows, .. } => Ok(rows),
3035            _ => Err(EngineError::Unsupported(
3036                "LATERAL subquery must be a SELECT (cannot be a write statement)".into(),
3037            )),
3038        }
3039    }
3040
3041    /// v7.30.3 (mailrs round-26) — bounded execution for the backfill
3042    /// shape that walked prod into reclaim livelock:
3043    ///
3044    ///   SELECT … FROM big b JOIN small s ON b.k = s.k
3045    ///   WHERE … ORDER BY … LIMIT n
3046    ///
3047    /// The general join path materialises the FULL join+filter result
3048    /// (≈2× the table's fat columns on a fresh backfill scan) before
3049    /// LIMIT truncates to n rows. Here the primary streams row-by-row
3050    /// against a hash of the materialised peer, and accepted rows feed
3051    /// a keep = LIMIT+OFFSET bounded top-N heap — peak memory scales
3052    /// with the answer, not the table. Returns Ok(None) when the shape
3053    /// doesn't qualify; the caller falls through to the general path,
3054    /// which the byte budget guards.
3055    /// v7.34.5 (mailrs prod #5 / `content_worker` 250 k) — walker-
3056    /// driven sibling of `try_streamed_inner_join_topn`. When the
3057    /// outer ORDER BY is on an indexed primary column, drive the
3058    /// primary scan via the BTree iterator in the requested
3059    /// direction so rows arrive already in ORDER BY order; the join
3060    /// + WHERE filter + early-stop run unchanged afterwards, BUT
3061    /// the heap-based top-N (which still walks every primary row)
3062    /// becomes a plain `Vec` that breaks after `LIMIT + OFFSET`
3063    /// survivors. Mirrors the single-table `try_pk_walk_top_n`
3064    /// eligibility gates plus the `try_streamed_inner_join_topn`
3065    /// join-shape gates. Returns `None` on any miss → the legacy
3066    /// heap streamer + general path handle it.
3067    /// v7.37.x (docker-fair NOTEX attack) — short-circuit
3068    ///   SELECT COUNT(*) FROM A LEFT JOIN B ON B.k = A.fk WHERE B.k IS NULL
3069    /// (the v7.37.27 NOT-EXISTS pull-up output shape). Materialising
3070    /// every (outer, NULL-padded right) tuple just to count survivors
3071    /// is wasted work — build a `HashSet<i64>` of B's unique join
3072    /// values (B.k must be UNIQUE / PK on a single integer column), scan
3073    /// A's storage, and increment the counter on each miss. PG's Merge
3074    /// Anti-Join does the same shape over both PK indexes. Returns
3075    /// `None` on any eligibility miss; the general join + aggregate
3076    /// path handles non-matching shapes.
3077    pub(crate) fn try_count_star_left_anti_join_fast(
3078        &self,
3079        stmt: &SelectStatement,
3080        from: &FromClause,
3081    ) -> Result<Option<QueryResult>, EngineError> {
3082        use spg_sql::ast::{JoinKind, SelectItem};
3083        ANTI_JOIN_FAST_PATH_TRIED.fetch_add(1, core::sync::atomic::Ordering::Relaxed);
3084        if stmt.distinct
3085            || stmt.limit_with_ties
3086            || stmt.group_by.is_some()
3087            || stmt.having.is_some()
3088            || !stmt.unions.is_empty()
3089            || !stmt.order_by.is_empty()
3090            || stmt.limit.is_some()
3091            || stmt.offset.is_some()
3092        {
3093            return Ok(None);
3094        }
3095        if from.joins.len() != 1 {
3096            return Ok(None);
3097        }
3098        let join = &from.joins[0];
3099        if !matches!(join.kind, JoinKind::Left) {
3100            return Ok(None);
3101        }
3102        // Gate: outer + inner must be plain catalog tables.
3103        let plain = |t: &spg_sql::ast::TableRef| {
3104            t.unnest_expr.is_none()
3105                && t.lateral_subquery.is_none()
3106                && t.as_of_segment.is_none()
3107                && t.generate_series_args.is_none()
3108        };
3109        if !plain(&from.primary) || !plain(&join.table) {
3110            return Ok(None);
3111        }
3112        let outer_alias = from
3113            .primary
3114            .alias
3115            .as_deref()
3116            .unwrap_or(from.primary.name.as_str());
3117        let inner_alias = join
3118            .table
3119            .alias
3120            .as_deref()
3121            .unwrap_or(join.table.name.as_str());
3122        // Items must be a single `COUNT(*)`.
3123        if stmt.items.len() != 1 {
3124            return Ok(None);
3125        }
3126        let SelectItem::Expr { expr, .. } = &stmt.items[0] else {
3127            return Ok(None);
3128        };
3129        let is_count_star = matches!(expr, Expr::FunctionCall { name, args }
3130            if name.eq_ignore_ascii_case("count_star") && args.is_empty());
3131        if !is_count_star {
3132            return Ok(None);
3133        }
3134        // ON clause: single equality on outer.col = inner.col (or
3135        // commuted). Capture (outer_col, inner_col).
3136        let Some(on) = join.on.as_ref() else {
3137            return Ok(None);
3138        };
3139        // v7.39 (round 744) — the ON accepts a plain pair (the r178
3140        // shape) OR `outer.col = <integer-only expression over inner>`
3141        // (the computed key the round-721 pull-up emits). For the
3142        // computed shape the IS NULL column must be one the key
3143        // expression READS: a real match forces the expression non-NULL,
3144        // hence every referenced column non-NULL — so the filter selects
3145        // exactly the pad rows. Any other inner column could be NULL on
3146        // a MATCHED row and the count would be wrong.
3147        enum InnerKey {
3148            Col(String),
3149            Expr(Expr),
3150        }
3151        let (outer_col, inner_key, null_cols): (String, InnerKey, Vec<String>) =
3152            if let Some((oc, ic)) = analyse_join_eq(on, outer_alias, inner_alias)? {
3153                let nulls = alloc::vec![ic.clone()];
3154                (oc, InnerKey::Col(ic), nulls)
3155            } else if let Some((oc, e)) = analyse_join_eq_expr(on, outer_alias, inner_alias) {
3156                let mut cols: Vec<String> = Vec::new();
3157                collect_inner_int_cols(&e, &mut cols);
3158                (oc, InnerKey::Expr(e), cols)
3159            } else {
3160                return Ok(None);
3161            };
3162        // WHERE clause: single `inner_alias.<col> IS NULL` predicate
3163        // (canonical anti-join filter), col constrained as above.
3164        let Some(where_expr) = stmt.where_.as_ref() else {
3165            return Ok(None);
3166        };
3167        if !null_cols
3168            .iter()
3169            .any(|c| is_inner_is_null(where_expr, inner_alias, c))
3170        {
3171            return Ok(None);
3172        }
3173        // Set membership is duplicate-insensitive for an ANTI count, so
3174        // no uniqueness gate is needed on either shape; the columns just
3175        // have to be integer-family so the i64 set is exact.
3176        let catalog = self.active_catalog();
3177        let Some(inner_table) = catalog.get(join.table.name.as_str()) else {
3178            return Ok(None);
3179        };
3180        let inner_schema = inner_table.schema();
3181        let int_col_pos = |name: &str| -> Option<usize> {
3182            inner_schema
3183                .columns
3184                .iter()
3185                .position(|c| c.name.eq_ignore_ascii_case(name))
3186                .filter(|&p| {
3187                    matches!(
3188                        inner_schema.columns[p].ty,
3189                        spg_storage::DataType::BigInt
3190                            | spg_storage::DataType::Int
3191                            | spg_storage::DataType::SmallInt
3192                    )
3193                })
3194        };
3195        let inner_pos: Option<usize> = match &inner_key {
3196            InnerKey::Col(c) => {
3197                let Some(p) = int_col_pos(c) else {
3198                    return Ok(None);
3199                };
3200                Some(p)
3201            }
3202            InnerKey::Expr(_) => {
3203                // Every column the expression reads must be inner int.
3204                if !null_cols.iter().all(|c| int_col_pos(c).is_some()) {
3205                    return Ok(None);
3206                }
3207                None
3208            }
3209        };
3210        let Some(outer_table) = catalog.get(from.primary.name.as_str()) else {
3211            return Ok(None);
3212        };
3213        let outer_schema = outer_table.schema();
3214        let Some(outer_pos) = outer_schema
3215            .columns
3216            .iter()
3217            .position(|c| c.name.eq_ignore_ascii_case(&outer_col))
3218        else {
3219            return Ok(None);
3220        };
3221        let outer_ty = outer_schema.columns[outer_pos].ty;
3222        if !matches!(
3223            outer_ty,
3224            spg_storage::DataType::BigInt
3225                | spg_storage::DataType::Int
3226                | spg_storage::DataType::SmallInt
3227        ) {
3228            return Ok(None);
3229        }
3230        // Build the antiset.
3231        let read_int = |v: &Value| -> Option<i64> {
3232            match v {
3233                Value::BigInt(n) => Some(*n),
3234                Value::Int(n) => Some(i64::from(*n)),
3235                Value::SmallInt(n) => Some(i64::from(*n)),
3236                _ => None,
3237            }
3238        };
3239        // Phase C.3 step 2b — MVCC read gate for the anti-join count
3240        // fast path. Both loops iterate hot-tier rows (`.rows()`) by
3241        // physical index, so a row this snapshot cannot see must neither
3242        // seed the antiset nor be counted. No-op today: every hot header
3243        // is frozen/committed-alive.
3244        let scan_snapshot = self.current_snapshot();
3245        let mut antiset: hashbrown::HashSet<i64> =
3246            hashbrown::HashSet::with_capacity(inner_table.row_count());
3247        let inner_ctx = self.ev_ctx(&inner_schema.columns, Some(inner_alias));
3248        for (i, row) in inner_table.rows().iter().enumerate() {
3249            if !inner_table.is_row_visible(i, &scan_snapshot) {
3250                continue;
3251            }
3252            match (&inner_key, inner_pos) {
3253                (InnerKey::Col(_), Some(p)) => {
3254                    if let Some(v) = row.values.get(p)
3255                        && let Some(k) = read_int(v)
3256                    {
3257                        antiset.insert(k);
3258                    }
3259                }
3260                (InnerKey::Expr(e), _) => {
3261                    let v = eval::eval_expr(e, row, &inner_ctx).map_err(EngineError::Eval)?;
3262                    if let Some(k) = read_int(&v) {
3263                        antiset.insert(k);
3264                    }
3265                }
3266                _ => unreachable!("Col always carries a position"),
3267            }
3268        }
3269        // Walk outer; count rows whose key isn't in the set OR whose key
3270        // is NULL (a NULL outer key has no join match either way).
3271        let mut count: i64 = 0;
3272        for (i, row) in outer_table.rows().iter().enumerate() {
3273            if !outer_table.is_row_visible(i, &scan_snapshot) {
3274                continue;
3275            }
3276            match row.values.get(outer_pos) {
3277                Some(v) => match read_int(v) {
3278                    Some(k) => {
3279                        if !antiset.contains(&k) {
3280                            count += 1;
3281                        }
3282                    }
3283                    None => count += 1,
3284                },
3285                None => count += 1,
3286            }
3287        }
3288        let columns = alloc::vec![ColumnSchema::new(
3289            "count".to_string(),
3290            spg_storage::DataType::BigInt,
3291            false,
3292        )];
3293        let rows = alloc::vec![Row::new(alloc::vec![Value::BigInt(count)])];
3294        let _ = outer_alias;
3295        let _ = outer_col;
3296        ANTI_JOIN_FAST_PATH_FIRED.fetch_add(1, core::sync::atomic::Ordering::Relaxed);
3297        Ok(Some(QueryResult::Rows { columns, rows }))
3298    }
3299
3300    pub(crate) fn try_streamed_inner_join_walk_topn(
3301        &self,
3302        stmt: &SelectStatement,
3303        from: &FromClause,
3304        cancel: CancelToken<'_>,
3305    ) -> Result<Option<QueryResult>, EngineError> {
3306        let Some(limit) = stmt.limit_literal() else {
3307            return Ok(None);
3308        };
3309        if stmt.offset.is_some() && stmt.offset_literal().is_none() {
3310            return Ok(None);
3311        }
3312        if stmt.distinct
3313            || stmt.limit_with_ties
3314            || stmt.group_by.is_some()
3315            || stmt.having.is_some()
3316            || aggregate::uses_aggregate(stmt)
3317        {
3318            return Ok(None);
3319        }
3320        if from.joins.len() != 1 {
3321            return Ok(None);
3322        }
3323        let j = &from.joins[0];
3324        if !matches!(j.kind, JoinKind::Inner) {
3325            return Ok(None);
3326        }
3327        let plain = |t: &TableRef| {
3328            t.unnest_expr.is_none() && t.lateral_subquery.is_none() && t.as_of_segment.is_none()
3329        };
3330        if !plain(&from.primary) || !plain(&j.table) {
3331            return Ok(None);
3332        }
3333        let Some(on_expr) = j.on.as_ref() else {
3334            return Ok(None);
3335        };
3336        let Some(primary_table) = self.active_catalog().get(&from.primary.name) else {
3337            return Ok(None);
3338        };
3339        if self.active_catalog().get(&j.table.name).is_none() {
3340            return Ok(None);
3341        }
3342        let primary_alias = from
3343            .primary
3344            .alias
3345            .as_deref()
3346            .unwrap_or(from.primary.name.as_str())
3347            .to_string();
3348        // Walker eligibility — single-key ORDER BY on a btree-indexed
3349        // primary column.
3350        if stmt.order_by.len() != 1 {
3351            return Ok(None);
3352        }
3353        let order = &stmt.order_by[0];
3354        let Expr::Column(order_col) = &order.expr else {
3355            return Ok(None);
3356        };
3357        if let Some(q) = &order_col.qualifier
3358            && !q.eq_ignore_ascii_case(&primary_alias)
3359        {
3360            return Ok(None);
3361        }
3362        let primary_cols = primary_table.schema().columns.clone();
3363        let Some(order_col_pos) = primary_cols
3364            .iter()
3365            .position(|c| c.name.eq_ignore_ascii_case(&order_col.name))
3366        else {
3367            return Ok(None);
3368        };
3369        let Some(order_index) = primary_table.index_on(order_col_pos) else {
3370            return Ok(None);
3371        };
3372        if !matches!(order_index.kind, spg_storage::IndexKind::BTree(_)) {
3373            return Ok(None);
3374        }
3375        // Peer side: same materialise + prune as the heap streamer.
3376        let peer_alias = j
3377            .table
3378            .alias
3379            .as_deref()
3380            .unwrap_or(j.table.name.as_str())
3381            .to_string();
3382        let mut needed = alloc::collections::BTreeSet::new();
3383        let prunable = collect_qualified_refs(stmt, &mut needed).is_some();
3384        let mut budget = ByteBudget::new(self.max_query_bytes);
3385        let (mut peer_rows, peer_cols) = self.materialise_table_ref_filtered(&j.table, &[])?;
3386        if prunable {
3387            Self::null_out_unreferenced(&mut peer_rows, &peer_cols, &peer_alias, &needed);
3388        }
3389        budget.charge(approx_rows_bytes(&peer_rows))?;
3390        let mut combined_schema: Vec<ColumnSchema> = Vec::new();
3391        for col in &primary_cols {
3392            combined_schema.push(ColumnSchema::new(
3393                alloc::format!("{primary_alias}.{}", col.name),
3394                col.ty,
3395                col.nullable,
3396            ));
3397        }
3398        for col in &peer_cols {
3399            combined_schema.push(ColumnSchema::new(
3400                alloc::format!("{peer_alias}.{}", col.name),
3401                col.ty,
3402                col.nullable,
3403            ));
3404        }
3405        // v7.39 (read01 round 53) — the join's EvalContext must carry the
3406        // catalog. Without it a `::regclass` / enum / composite cast inside a
3407        // joined WHERE or ON falls back to plain text, so the canonical
3408        // `pg_class JOIN pg_index … WHERE indrelid = 't'::regclass` shape
3409        // errored on "comparison between BigInt and Text" — while the very
3410        // same predicate worked on a single-table SELECT (whose ctx does carry
3411        // the catalog). Same root as round 49's unnest(enum_range(…)).
3412        // v7.39 (round 525) — and the SESSION, for the same reason as the
3413        // catalog above: a join's WHERE is the same predicate a
3414        // single-table SELECT would carry, and `WHERE t =
3415        // current_setting('app.tenant')` failed on the joined shape while
3416        // working on the unjoined one.
3417        let join_sess = self.dml_session();
3418        let ctx = EvalContext::new(&combined_schema, None)
3419            .with_catalog(self.active_catalog())
3420            .with_session(&join_sess);
3421        let left_arity = primary_cols.len();
3422        let mut eq_pairs: Vec<(usize, usize)> = Vec::new();
3423        let mut residual: Vec<&Expr> = Vec::new();
3424        for sub in reorder::split_and_conjunctions(on_expr) {
3425            let mut matched = None;
3426            if let Expr::Binary {
3427                lhs,
3428                op: spg_sql::ast::BinOp::Eq,
3429                rhs,
3430            } = sub
3431                && let (Expr::Column(a), Expr::Column(b)) = (lhs.as_ref(), rhs.as_ref())
3432            {
3433                let left_slice = &combined_schema[..left_arity];
3434                if let (Some(l), Some(r)) = (
3435                    Self::composite_col_pos(left_slice, a),
3436                    Self::peer_col_pos(&peer_alias, &peer_cols, b),
3437                ) {
3438                    matched = Some((l, r));
3439                } else if let (Some(l), Some(r)) = (
3440                    Self::composite_col_pos(left_slice, b),
3441                    Self::peer_col_pos(&peer_alias, &peer_cols, a),
3442                ) {
3443                    matched = Some((l, r));
3444                }
3445            }
3446            match matched {
3447                Some(pair) => eq_pairs.push(pair),
3448                None => residual.push(sub),
3449            }
3450        }
3451        if eq_pairs.is_empty() {
3452            return Ok(None);
3453        }
3454        // Hash the peer on the equality key (same as the heap streamer).
3455        let mut htable: hashbrown::HashMap<String, Vec<usize>> =
3456            hashbrown::HashMap::with_capacity(peer_rows.len());
3457        let mut keybuf: Vec<Value<'static>> = Vec::with_capacity(eq_pairs.len());
3458        'build: for (ri, right) in peer_rows.iter().enumerate() {
3459            keybuf.clear();
3460            for (_, rpos) in &eq_pairs {
3461                let v = right.values.get(*rpos).cloned().unwrap_or(Value::Null);
3462                if matches!(v, Value::Null) {
3463                    continue 'build;
3464                }
3465                keybuf.push(v);
3466            }
3467            htable
3468                .entry(aggregate::encode_key(&keybuf))
3469                .or_default()
3470                .push(ri);
3471        }
3472        let keep_mask: Vec<bool> = primary_cols
3473            .iter()
3474            .map(|c| !prunable || needed.contains(&(primary_alias.clone(), c.name.clone())))
3475            .collect();
3476        let keep = (limit as usize).saturating_add(stmt.offset_literal().map_or(0, |o| o as usize));
3477        let mut where_memo = memoize::MemoizeCache::default();
3478        let mut plain_sink: Vec<Row<'static>> = Vec::with_capacity(keep.min(1024));
3479        // v7.37.6 (mailrs content_worker — Attack #2 from
3480        // `v7.37.5-content-worker-prod-decomposition.md`): pre-split
3481        // the WHERE predicate into conjuncts that reference only the
3482        // outer (`primary_alias`) and conjuncts that touch the peer
3483        // alias (mixed). Outer-only conjuncts can be evaluated against
3484        // `left` BEFORE we materialise `combined_vals`, which lets the
3485        // 25 k-iter content_worker hot path skip the 12-cell per-row
3486        // clone when the InList probe (`m.id NOT IN`) misses — which it
3487        // does on > 99 % of rows in the prod snapshot.
3488        //
3489        // We also split the ON residual the same way so a `mb.foo = …`
3490        // predicate that's been mis-folded into the residual still goes
3491        // through the slow combined-row path, and an `m.foo = …` one
3492        // gates before the materialise.
3493        //
3494        // Implementation notes:
3495        // - We allocate the split once per query (the walker iterates,
3496        //   the split does not).
3497        // - Outer-only conjuncts evaluate against
3498        //   `&combined_schema[..left_arity]` paired with `left`, so the
3499        //   column resolver indices match the way they would on the
3500        //   combined row (positions 0..left_arity are identical).
3501        // - The full WHERE still re-runs on the post-materialise path
3502        //   only for the conjuncts the split could not classify as
3503        //   outer-only (this keeps the semantics identical even when an
3504        //   unknown / subquery node is present; `expr_references_alias`
3505        //   is conservative).
3506        let outer_schema: &[ColumnSchema] = &combined_schema[..left_arity];
3507        let outer_ctx = EvalContext::new(outer_schema, None).with_catalog(self.active_catalog());
3508        let where_conjuncts: Vec<&Expr> = stmt
3509            .where_
3510            .as_ref()
3511            .map(|w| reorder::split_and_conjunctions(w))
3512            .unwrap_or_default();
3513        let (where_outer_only, where_mixed): (Vec<&Expr>, Vec<&Expr>) =
3514            where_conjuncts.iter().copied().partition(|e| {
3515                crate::joinfold::expr_references_alias(e, &primary_alias)
3516                    && !crate::joinfold::expr_references_any_other_alias(e, &primary_alias)
3517            });
3518        let (residual_outer_only, residual_mixed): (Vec<&Expr>, Vec<&Expr>) =
3519            residual.iter().copied().partition(|e| {
3520                crate::joinfold::expr_references_alias(e, &primary_alias)
3521                    && !crate::joinfold::expr_references_any_other_alias(e, &primary_alias)
3522            });
3523        let mut outer_memo = memoize::MemoizeCache::default();
3524        // Walker drive: walk primary via btree index in ORDER BY
3525        // direction. Rows arrive already sorted; plain_sink + early
3526        // stop replaces the heap.
3527        let walker: alloc::boxed::Box<
3528            dyn Iterator<Item = (&spg_storage::IndexKey, &Vec<spg_storage::RowLocator>)>,
3529        > = if order.desc {
3530            alloc::boxed::Box::new(order_index.iter_desc())
3531        } else {
3532            alloc::boxed::Box::new(order_index.iter_asc())
3533        };
3534        let primary_table_name = primary_table.schema().name.clone();
3535        // Phase C.3 step 2b — MVCC read gate for the streamed-join
3536        // walker's primary. Snapshot computed once; a hot primary row
3537        // this snapshot cannot see is skipped so a dead/old version never
3538        // drives a join tuple. Cold locators are frozen = always visible.
3539        // No-op today: every hot header is frozen/committed-alive.
3540        let scan_snapshot = self.current_snapshot();
3541        'walk: for (key, locators) in walker {
3542            cancel.check()?;
3543            for loc in locators {
3544                // v7.34.6 (mailrs prod #6) — cold-tier dispatch on the
3545                // walker. Pre-v7.34.6 bailed the whole walker on the
3546                // first cold locator, which is exactly the prod-803MB
3547                // shape: messages at scale has older rows promoted to
3548                // cold segments, so the ORDER BY id DESC walk hits a
3549                // cold locator on the very first batch and the entire
3550                // plan falls back to the 82ms NOT-IN scan-and-sort.
3551                // `Catalog::resolve_cold_locator` reads one cold
3552                // segment page + decodes the dense row body, which
3553                // ports the walker's early-stop across the tier
3554                // boundary at ~µs per row.
3555                let left_cow: Cow<'_, Row> = match *loc {
3556                    spg_storage::RowLocator::Hot(i) => {
3557                        if !primary_table.is_row_visible(i, &scan_snapshot) {
3558                            continue;
3559                        }
3560                        match primary_table.rows().get(i) {
3561                            Some(r) => Cow::Borrowed(r),
3562                            None => continue,
3563                        }
3564                    }
3565                    spg_storage::RowLocator::Cold { segment_id, .. } => {
3566                        match self.active_catalog().resolve_cold_locator(
3567                            &primary_table_name,
3568                            segment_id,
3569                            key,
3570                        ) {
3571                            Some(r) => Cow::Owned(r),
3572                            None => continue,
3573                        }
3574                    }
3575                };
3576                let left: &Row<'static> = left_cow.as_ref();
3577                keybuf.clear();
3578                let mut left_has_null = false;
3579                for (lpos, _) in &eq_pairs {
3580                    let v = left.values.get(*lpos).cloned().unwrap_or(Value::Null);
3581                    if matches!(v, Value::Null) {
3582                        left_has_null = true;
3583                        break;
3584                    }
3585                    keybuf.push(v);
3586                }
3587                if left_has_null {
3588                    continue;
3589                }
3590                let Some(cands) = htable.get(&aggregate::encode_key(&keybuf)) else {
3591                    continue;
3592                };
3593                // v7.37.6 — gate the outer-only WHERE conjuncts +
3594                // outer-only ON residual on `left` before we ever clone
3595                // into `combined_vals`. content_worker's `m.size > 0
3596                // AND m.id NOT IN (…25 k…)` is outer-only on `m.*`; if
3597                // the InList probe misses (>99 %), we skip the 12-cell
3598                // clone + extend + Row::new + budget charge for every
3599                // peer candidate this outer row hashed to.
3600                let mut outer_ok = true;
3601                for r in &residual_outer_only {
3602                    let cond = self.eval_expr_with_correlated(r, left, &outer_ctx, cancel, None)?;
3603                    if !crate::eval::predicate_is_true(&cond, "JOIN/ON", ctx.mysql_dialect)? {
3604                        outer_ok = false;
3605                        break;
3606                    }
3607                }
3608                if !outer_ok {
3609                    continue;
3610                }
3611                for w in &where_outer_only {
3612                    let cond = self.eval_expr_with_correlated(
3613                        w,
3614                        left,
3615                        &outer_ctx,
3616                        cancel,
3617                        Some(&mut outer_memo),
3618                    )?;
3619                    if !crate::eval::predicate_is_true(&cond, "JOIN/ON", ctx.mysql_dialect)? {
3620                        outer_ok = false;
3621                        break;
3622                    }
3623                }
3624                if !outer_ok {
3625                    continue;
3626                }
3627                for &ri in cands {
3628                    let right = &peer_rows[ri];
3629                    let mut combined_vals: Vec<Value<'static>> =
3630                        Vec::with_capacity(left_arity + peer_cols.len());
3631                    for (i, v) in left.values.iter().enumerate() {
3632                        combined_vals.push(if keep_mask.get(i).copied().unwrap_or(true) {
3633                            v.clone()
3634                        } else {
3635                            Value::Null
3636                        });
3637                    }
3638                    combined_vals.extend(right.values.iter().cloned());
3639                    let combined = Row::new(combined_vals);
3640                    let mut ok = true;
3641                    for r in &residual_mixed {
3642                        let cond =
3643                            self.eval_expr_with_correlated(r, &combined, &ctx, cancel, None)?;
3644                        if !crate::eval::predicate_is_true(&cond, "JOIN/ON", ctx.mysql_dialect)? {
3645                            ok = false;
3646                            break;
3647                        }
3648                    }
3649                    if !ok {
3650                        continue;
3651                    }
3652                    for w in &where_mixed {
3653                        let cond = self.eval_expr_with_correlated(
3654                            w,
3655                            &combined,
3656                            &ctx,
3657                            cancel,
3658                            Some(&mut where_memo),
3659                        )?;
3660                        if !crate::eval::predicate_is_true(&cond, "JOIN/ON", ctx.mysql_dialect)? {
3661                            ok = false;
3662                            break;
3663                        }
3664                    }
3665                    if !ok {
3666                        continue;
3667                    }
3668                    budget.charge(approx_row_bytes(&combined))?;
3669                    plain_sink.push(combined);
3670                    if plain_sink.len() >= keep {
3671                        break 'walk;
3672                    }
3673                }
3674            }
3675        }
3676        // Already in ORDER BY order from the walk.
3677        let mut output = plain_sink;
3678        apply_offset_and_limit(&mut output, stmt.offset_literal(), stmt.limit_literal());
3679        let projection =
3680            build_projection(&stmt.items, &combined_schema, "", self.backslash_escapes)?;
3681        let mut proj_memo = memoize::MemoizeCache::default();
3682        let mut rows: Vec<Row<'static>> = Vec::with_capacity(output.len());
3683        for row in &output {
3684            let mut values = Vec::with_capacity(projection.len());
3685            for p in &projection {
3686                values.push(self.eval_expr_with_correlated(
3687                    &p.expr,
3688                    row,
3689                    &ctx,
3690                    cancel,
3691                    Some(&mut proj_memo),
3692                )?);
3693            }
3694            rows.push(Row::new(values));
3695        }
3696        let columns: Vec<ColumnSchema> = projection
3697            .into_iter()
3698            .map(|p| ColumnSchema::new(p.output_name, p.ty, p.nullable))
3699            .collect();
3700        Ok(Some(QueryResult::Rows { columns, rows }))
3701    }
3702
3703    pub(crate) fn try_streamed_inner_join_topn(
3704        &self,
3705        stmt: &SelectStatement,
3706        from: &FromClause,
3707        cancel: CancelToken<'_>,
3708    ) -> Result<Option<QueryResult>, EngineError> {
3709        // Shape gate — any bail lands on the general path.
3710        let Some(limit) = stmt.limit_literal() else {
3711            return Ok(None);
3712        };
3713        if stmt.offset.is_some() && stmt.offset_literal().is_none() {
3714            return Ok(None);
3715        }
3716        if stmt.distinct
3717            || stmt.group_by.is_some()
3718            || stmt.having.is_some()
3719            || aggregate::uses_aggregate(stmt)
3720        {
3721            return Ok(None);
3722        }
3723        if from.joins.len() != 1 {
3724            return Ok(None);
3725        }
3726        let j = &from.joins[0];
3727        if !matches!(j.kind, JoinKind::Inner) {
3728            return Ok(None);
3729        }
3730        let plain = |t: &TableRef| {
3731            t.unnest_expr.is_none() && t.lateral_subquery.is_none() && t.as_of_segment.is_none()
3732        };
3733        if !plain(&from.primary) || !plain(&j.table) {
3734            return Ok(None);
3735        }
3736        let Some(on_expr) = j.on.as_ref() else {
3737            return Ok(None);
3738        };
3739        // Plain catalog tables only — views / virtual tables keep the
3740        // general path's materialise_table_ref fallback.
3741        let Some(primary_table) = self.active_catalog().get(&from.primary.name) else {
3742            return Ok(None);
3743        };
3744        if self.active_catalog().get(&j.table.name).is_none() {
3745            return Ok(None);
3746        }
3747        let primary_alias = from
3748            .primary
3749            .alias
3750            .as_deref()
3751            .unwrap_or(from.primary.name.as_str())
3752            .to_string();
3753        let peer_alias = j
3754            .table
3755            .alias
3756            .as_deref()
3757            .unwrap_or(j.table.name.as_str())
3758            .to_string();
3759        let mut needed = alloc::collections::BTreeSet::new();
3760        let prunable = collect_qualified_refs(stmt, &mut needed).is_some();
3761        // Peer side: materialise + prune exactly like the general
3762        // path; the budget still guards a degenerately fat peer.
3763        let mut budget = ByteBudget::new(self.max_query_bytes);
3764        let (mut peer_rows, peer_cols) = self.materialise_table_ref_filtered(&j.table, &[])?;
3765        if prunable {
3766            Self::null_out_unreferenced(&mut peer_rows, &peer_cols, &peer_alias, &needed);
3767        }
3768        budget.charge(approx_rows_bytes(&peer_rows))?;
3769        let primary_cols = primary_table.schema().columns.clone();
3770        let mut combined_schema: Vec<ColumnSchema> = Vec::new();
3771        for col in &primary_cols {
3772            combined_schema.push(ColumnSchema::new(
3773                alloc::format!("{primary_alias}.{}", col.name),
3774                col.ty,
3775                col.nullable,
3776            ));
3777        }
3778        for col in &peer_cols {
3779            combined_schema.push(ColumnSchema::new(
3780                alloc::format!("{peer_alias}.{}", col.name),
3781                col.ty,
3782                col.nullable,
3783            ));
3784        }
3785        // v7.39 (read01 round 53) — the join's EvalContext must carry the
3786        // catalog. Without it a `::regclass` / enum / composite cast inside a
3787        // joined WHERE or ON falls back to plain text, so the canonical
3788        // `pg_class JOIN pg_index … WHERE indrelid = 't'::regclass` shape
3789        // errored on "comparison between BigInt and Text" — while the very
3790        // same predicate worked on a single-table SELECT (whose ctx does carry
3791        // the catalog). Same root as round 49's unnest(enum_range(…)).
3792        // v7.39 (round 525) — and the SESSION, for the same reason as the
3793        // catalog above: a join's WHERE is the same predicate a
3794        // single-table SELECT would carry, and `WHERE t =
3795        // current_setting('app.tenant')` failed on the joined shape while
3796        // working on the unjoined one.
3797        let join_sess = self.dml_session();
3798        let ctx = EvalContext::new(&combined_schema, None)
3799            .with_catalog(self.active_catalog())
3800            .with_session(&join_sess);
3801        // Hash-joinable left = right equality pairs from ON; anything
3802        // else stays as a residual conjunct on the candidate row.
3803        let left_arity = primary_cols.len();
3804        let mut eq_pairs: Vec<(usize, usize)> = Vec::new();
3805        let mut residual: Vec<&Expr> = Vec::new();
3806        for sub in reorder::split_and_conjunctions(on_expr) {
3807            let mut matched = None;
3808            if let Expr::Binary {
3809                lhs,
3810                op: spg_sql::ast::BinOp::Eq,
3811                rhs,
3812            } = sub
3813                && let (Expr::Column(a), Expr::Column(b)) = (lhs.as_ref(), rhs.as_ref())
3814            {
3815                let left_slice = &combined_schema[..left_arity];
3816                if let (Some(l), Some(r)) = (
3817                    Self::composite_col_pos(left_slice, a),
3818                    Self::peer_col_pos(&peer_alias, &peer_cols, b),
3819                ) {
3820                    matched = Some((l, r));
3821                } else if let (Some(l), Some(r)) = (
3822                    Self::composite_col_pos(left_slice, b),
3823                    Self::peer_col_pos(&peer_alias, &peer_cols, a),
3824                ) {
3825                    matched = Some((l, r));
3826                }
3827            }
3828            match matched {
3829                Some(pair) => eq_pairs.push(pair),
3830                None => residual.push(sub),
3831            }
3832        }
3833        if eq_pairs.is_empty() {
3834            return Ok(None); // nested-loop shapes stay on the general path
3835        }
3836        // Hash the peer on the equality key (NULL keys never match).
3837        let mut htable: hashbrown::HashMap<String, Vec<usize>> =
3838            hashbrown::HashMap::with_capacity(peer_rows.len());
3839        let mut keybuf: Vec<Value<'static>> = Vec::with_capacity(eq_pairs.len());
3840        'build: for (ri, right) in peer_rows.iter().enumerate() {
3841            keybuf.clear();
3842            for (_, rpos) in &eq_pairs {
3843                let v = right.values.get(*rpos).cloned().unwrap_or(Value::Null);
3844                if matches!(v, Value::Null) {
3845                    continue 'build;
3846                }
3847                keybuf.push(v);
3848            }
3849            htable
3850                .entry(aggregate::encode_key(&keybuf))
3851                .or_default()
3852                .push(ri);
3853        }
3854        // Streamed twin of null_out_unreferenced: clone only the
3855        // referenced primary columns into each candidate row.
3856        let keep_mask: Vec<bool> = primary_cols
3857            .iter()
3858            .map(|c| !prunable || needed.contains(&(primary_alias.clone(), c.name.clone())))
3859            .collect();
3860        let keep = (limit as usize).saturating_add(stmt.offset_literal().map_or(0, |o| o as usize));
3861        let descs: alloc::rc::Rc<[bool]> = stmt
3862            .order_by
3863            .iter()
3864            .map(|o| o.desc)
3865            .collect::<Vec<bool>>()
3866            .into();
3867        let mut where_memo = memoize::MemoizeCache::default();
3868        let mut heap: alloc::collections::BinaryHeap<TopNEntry> =
3869            alloc::collections::BinaryHeap::new();
3870        let mut plain_sink: Vec<Row<'static>> = Vec::new();
3871        let mut seq: u64 = 0;
3872        // v7.36 (cold-tier coverage) — extend the primary scan with
3873        // the cold-tier rows so `ORDER BY <non-indexed> LIMIT N`
3874        // doesn't lose half a freezer-promoted table when the walker
3875        // shape isn't a match. Hot rows borrow from `PersistentVec`;
3876        // cold rows are pre-materialised once and yielded in order.
3877        let primary_cold = self.iter_cold_rows_of_table(primary_table);
3878        // v7.37.15 Phase B — visibility gate the primary join scan.
3879        // The cold tier still iterates directly because cold rows have
3880        // no per-row header (they're frozen segments — equivalent to
3881        // RowHeader::frozen() for visibility purposes). Phase D wires
3882        // per-segment all-visible bitmaps so cold scans skip the
3883        // visibility check entirely.
3884        let snap = self.current_snapshot();
3885        'scan: for left in primary_table
3886            .scan_visible(&snap)
3887            .map(|(_, r)| r)
3888            .chain(primary_cold.iter())
3889        {
3890            cancel.check()?;
3891            if keep == 0 {
3892                break 'scan;
3893            }
3894            keybuf.clear();
3895            let mut left_has_null = false;
3896            for (lpos, _) in &eq_pairs {
3897                let v = left.values.get(*lpos).cloned().unwrap_or(Value::Null);
3898                if matches!(v, Value::Null) {
3899                    left_has_null = true;
3900                    break;
3901                }
3902                keybuf.push(v);
3903            }
3904            if left_has_null {
3905                continue;
3906            }
3907            let Some(cands) = htable.get(&aggregate::encode_key(&keybuf)) else {
3908                continue;
3909            };
3910            for &ri in cands {
3911                let right = &peer_rows[ri];
3912                let mut combined_vals: Vec<Value<'static>> =
3913                    Vec::with_capacity(left_arity + peer_cols.len());
3914                for (i, v) in left.values.iter().enumerate() {
3915                    combined_vals.push(if keep_mask.get(i).copied().unwrap_or(true) {
3916                        v.clone()
3917                    } else {
3918                        Value::Null
3919                    });
3920                }
3921                combined_vals.extend(right.values.iter().cloned());
3922                let combined = Row::new(combined_vals);
3923                let mut ok = true;
3924                for r in &residual {
3925                    let cond = self.eval_expr_with_correlated(r, &combined, &ctx, cancel, None)?;
3926                    if !crate::eval::predicate_is_true(&cond, "JOIN/ON", ctx.mysql_dialect)? {
3927                        ok = false;
3928                        break;
3929                    }
3930                }
3931                if !ok {
3932                    continue;
3933                }
3934                if let Some(w) = stmt.where_.as_ref() {
3935                    let cond = self.eval_expr_with_correlated(
3936                        w,
3937                        &combined,
3938                        &ctx,
3939                        cancel,
3940                        Some(&mut where_memo),
3941                    )?;
3942                    if !crate::eval::predicate_is_true(&cond, "JOIN/ON", ctx.mysql_dialect)? {
3943                        continue;
3944                    }
3945                }
3946                if stmt.order_by.is_empty() {
3947                    budget.charge(approx_row_bytes(&combined))?;
3948                    plain_sink.push(combined);
3949                    if plain_sink.len() >= keep {
3950                        break 'scan;
3951                    }
3952                } else {
3953                    let keys = build_order_keys(&stmt.order_by, &combined, &ctx)?;
3954                    let entry = TopNEntry {
3955                        keys,
3956                        descs: alloc::rc::Rc::clone(&descs),
3957                        seq,
3958                        row: combined,
3959                    };
3960                    seq += 1;
3961                    if heap.len() < keep {
3962                        budget.charge(approx_row_bytes(&entry.row))?;
3963                        heap.push(entry);
3964                    } else if let Some(top) = heap.peek()
3965                        && entry < *top
3966                    {
3967                        if let Some(evicted) = heap.pop() {
3968                            budget.release(approx_row_bytes(&evicted.row));
3969                        }
3970                        budget.charge(approx_row_bytes(&entry.row))?;
3971                        heap.push(entry);
3972                    }
3973                }
3974            }
3975        }
3976        let mut output: Vec<Row<'static>> = if stmt.order_by.is_empty() {
3977            plain_sink
3978        } else {
3979            heap.into_sorted_vec().into_iter().map(|e| e.row).collect()
3980        };
3981        apply_offset_and_limit(&mut output, stmt.offset_literal(), stmt.limit_literal());
3982        let projection =
3983            build_projection(&stmt.items, &combined_schema, "", self.backslash_escapes)?;
3984        let mut proj_memo = memoize::MemoizeCache::default();
3985        let mut rows: Vec<Row<'static>> = Vec::with_capacity(output.len());
3986        for row in &output {
3987            let mut values = Vec::with_capacity(projection.len());
3988            for p in &projection {
3989                values.push(self.eval_expr_with_correlated(
3990                    &p.expr,
3991                    row,
3992                    &ctx,
3993                    cancel,
3994                    Some(&mut proj_memo),
3995                )?);
3996            }
3997            rows.push(Row::new(values));
3998        }
3999        let columns: Vec<ColumnSchema> = projection
4000            .into_iter()
4001            .map(|p| ColumnSchema::new(p.output_name, p.ty, p.nullable))
4002            .collect();
4003        Ok(Some(QueryResult::Rows { columns, rows }))
4004    }
4005}
4006
4007/// v7.17.0 Phase 3.P0-41 — synthesise a column name for a LATERAL
4008/// projection item that has no explicit alias. PG names anonymous
4009/// projection items by the function call's name or by `column<i>`.
4010/// SPG mirrors the latter (lower-overhead than walking arbitrary
4011/// Expr shapes) so the probe-schema fallback path produces stable
4012/// names for the lateral peer's columns.
4013pub(crate) fn synth_lateral_col_name(expr: &Expr, idx: usize) -> String {
4014    match expr {
4015        // Bare column reference — use the column's own name.
4016        Expr::Column(c) => c.name.clone(),
4017        // Function call — use the function name (PG canonical:
4018        // `count` / `max` / `lower` …).
4019        Expr::FunctionCall { name, .. } => name.clone(),
4020        // Cast — drill into the inner expression.
4021        Expr::Cast { expr: inner, .. } => synth_lateral_col_name(inner, idx),
4022        // Everything else falls back to PG's `column<N>` placeholder.
4023        _ => alloc::format!("column{}", idx + 1),
4024    }
4025}
4026
4027/// v7.17.0 Phase 3.P0-41 — substitute every `<alias>.<col>` Expr
4028/// reference whose `<alias>.<col>` exists in the outer composite
4029/// schema with the matching value from the outer row. Walks the
4030/// entire SELECT body (items, WHERE, GROUP BY, HAVING, ORDER BY,
4031/// UNION peers) so any depth of outer reference inside the
4032/// LATERAL subquery resolves before execution.
4033/// True when `e` is a compile-time constant — no column ref, function call,
4034/// subquery, or other outer-touching construct. Used to recognise a bare
4035/// `(VALUES …)` derived table (whose rows are pure literals) so it can be
4036/// eager-materialised as a join peer rather than forced through the per-left-row
4037/// lateral path (see D.19).
4038fn expr_is_constant(e: &Expr) -> bool {
4039    match e {
4040        Expr::Literal(_) | Expr::Placeholder(_) => true,
4041        Expr::Unary { expr, .. } | Expr::Cast { expr, .. } => expr_is_constant(expr),
4042        Expr::Binary { lhs, rhs, .. } => expr_is_constant(lhs) && expr_is_constant(rhs),
4043        _ => false,
4044    }
4045}
4046
4047/// True when `s` is a constant `(VALUES …)`-shaped derived table: the head and
4048/// every UNION-ALL peer has no FROM clause and projects only constant
4049/// expressions. Such a table references nothing outer, so it is safe to
4050/// materialise once and cross-join — unlike a correlated lateral (e.g.
4051/// `generate_series(1, outer.col)`), which must stay per-left-row.
4052/// v7.39 (round 572) — is this derived table a plain SELECT over stored
4053/// tables, with nothing set-returning in its FROM?
4054///
4055/// `select_is_correlated` answers about columns in the projection, the
4056/// WHERE and the nested subqueries. It does NOT see an outer reference
4057/// carried in a set-returning function's ARGUMENTS — `LATERAL
4058/// generate_series(1, lo.n)` and `LATERAL unnest(t.arr)` parse into a
4059/// synthesised SELECT whose correlation lives in `generate_series_args`
4060/// / `unnest_expr`, and it reports those as uncorrelated. Fifteen
4061/// lateral e2e tests said so the moment the gate widened.
4062///
4063/// So the eager path asks this first: every FROM item must be a named
4064/// stored table. An SRF anywhere keeps the per-left-row evaluation it
4065/// needs.
4066fn derived_is_plain_table_select(s: &SelectStatement, cat: &crate::Catalog) -> bool {
4067    let Some(from) = &s.from else {
4068        return false;
4069    };
4070    let plain = |t: &spg_sql::ast::TableRef| {
4071        t.unnest_expr.is_none()
4072            && t.generate_series_args.is_none()
4073            && t.lateral_subquery.is_none()
4074            // A set-returning FUNCTION reads as a named FROM item —
4075            // `LATERAL f(t.col)`, `jsonb_each_text(t.j)`, `json_table(…)`
4076            // all put the function's name here and their correlation in
4077            // the arguments. Resolving the name against the catalog is
4078            // what tells a stored table from one of those.
4079            && cat.get(&t.name).is_some()
4080    };
4081    plain(&from.primary) && from.joins.iter().all(|j| plain(&j.table))
4082}
4083
4084fn is_constant_values_derived(s: &SelectStatement) -> bool {
4085    use spg_sql::ast::SelectItem;
4086    let peer_ok = |p: &SelectStatement| -> bool {
4087        p.from.is_none()
4088            && p.where_.is_none()
4089            && p.having.is_none()
4090            && p.items.iter().all(|it| match it {
4091                SelectItem::Expr { expr, .. } => expr_is_constant(expr),
4092                SelectItem::Wildcard | SelectItem::QualifiedWildcard(_) => false,
4093            })
4094    };
4095    peer_ok(s) && s.unions.iter().all(|(_, peer)| peer_ok(peer))
4096}
4097
4098pub(crate) fn substitute_outer_columns_multi(
4099    stmt: &mut SelectStatement,
4100    outer_row: &Row<'static>,
4101    outer_schema: &[ColumnSchema],
4102) {
4103    substitute_outer_in_select(stmt, outer_row, outer_schema);
4104}
4105
4106/// v4.23: walk every Expr in `stmt` and replace each Column ref
4107/// that targets the outer scope (qualifier matches the outer
4108/// table alias) with a Literal carrying the outer row's value.
4109/// Conservative: only qualified refs are substituted, so the user
4110/// must write `outer_alias.col` to reference an outer column. This
4111/// matches PG's lexical scoping for correlated subqueries and
4112/// avoids accidentally rebinding inner columns of the same name.
4113fn substitute_outer_in_select(
4114    stmt: &mut SelectStatement,
4115    outer_row: &Row<'static>,
4116    outer_schema: &[ColumnSchema],
4117) {
4118    // A FROM-less SELECT (`LATERAL (SELECT <outer col> …)`) has no inner
4119    // scope of its own, so an unqualified column in its projection /
4120    // predicates must resolve to the outer row. A SELECT with a FROM
4121    // keeps the conservative qualified-only rule: bare names resolve
4122    // against its own tables first (PG lexical scoping).
4123    let bare = stmt.from.is_none();
4124    for item in &mut stmt.items {
4125        if let SelectItem::Expr { expr, .. } = item {
4126            substitute_outer_in_expr(expr, outer_row, outer_schema, bare);
4127        }
4128    }
4129    // v7.37.43-T4.5 — walk FROM-side SRF argument expressions
4130    // (`unnest(<expr>)` / `generate_series(<args>)` /
4131    // `jsonb_each_text(<expr>)`) so a LATERAL SRF with an outer-
4132    // column reference gets the reference substituted before
4133    // per-row execution.
4134    if let Some(from) = &mut stmt.from {
4135        substitute_outer_in_table_ref(&mut from.primary, outer_row, outer_schema);
4136        for j in &mut from.joins {
4137            substitute_outer_in_table_ref(&mut j.table, outer_row, outer_schema);
4138            if let Some(on) = &mut j.on {
4139                substitute_outer_in_expr(on, outer_row, outer_schema, bare);
4140            }
4141        }
4142    }
4143    if let Some(w) = &mut stmt.where_ {
4144        substitute_outer_in_expr(w, outer_row, outer_schema, bare);
4145    }
4146    if let Some(gs) = &mut stmt.group_by {
4147        for g in gs {
4148            substitute_outer_in_expr(g, outer_row, outer_schema, bare);
4149        }
4150    }
4151    if let Some(h) = &mut stmt.having {
4152        substitute_outer_in_expr(h, outer_row, outer_schema, bare);
4153    }
4154    for o in &mut stmt.order_by {
4155        substitute_outer_in_expr(&mut o.expr, outer_row, outer_schema, bare);
4156    }
4157    for (_, peer) in &mut stmt.unions {
4158        substitute_outer_in_select(peer, outer_row, outer_schema);
4159    }
4160}
4161
4162fn substitute_outer_in_table_ref(
4163    t: &mut spg_sql::ast::TableRef,
4164    outer_row: &Row<'static>,
4165    outer_schema: &[ColumnSchema],
4166) {
4167    // A set-returning FROM item's argument is evaluated with no inner
4168    // scope, so its unqualified column refs are outer refs
4169    // (`LATERAL unnest(<outer array col>)`).
4170    if let Some((_, arg)) = t.jsonb_each_text_arg.as_mut() {
4171        substitute_outer_in_expr(arg, outer_row, outer_schema, true);
4172    }
4173    if let Some(arg) = t.unnest_expr.as_deref_mut() {
4174        substitute_outer_in_expr(arg, outer_row, outer_schema, true);
4175    }
4176    // v7.39 (read01 round 69) — a user function on a JOIN's right side
4177    // (`t JOIN LATERAL dbl(t.id) AS d ON true`): its ARGUMENTS reference the
4178    // outer row, so they take the substitution too. Without this the call would
4179    // see an unresolved column and the correlation would silently not happen.
4180    if let Some(call) = t.table_fn_call.as_deref_mut() {
4181        for a in call.1.iter_mut() {
4182            substitute_outer_in_expr(a, outer_row, outer_schema, true);
4183        }
4184    }
4185    if let Some(args) = t.generate_series_args.as_mut() {
4186        for a in args.iter_mut() {
4187            substitute_outer_in_expr(a, outer_row, outer_schema, true);
4188        }
4189    }
4190    if let Some(inner) = t.lateral_subquery.as_deref_mut() {
4191        substitute_outer_in_select(inner, outer_row, outer_schema);
4192    }
4193    // v7.39 (round 205, JSON_TABLE) — the document expr (and PASSING
4194    // values) are evaluated with no inner scope, so their unqualified
4195    // / outer-qualified column refs are outer refs (implicit LATERAL:
4196    // `t, JSON_TABLE(t.arr, …)`).
4197    if let Some(jt) = t.json_table.as_deref_mut() {
4198        substitute_outer_in_expr(&mut jt.doc, outer_row, outer_schema, true);
4199        for (_, e) in jt.passing.iter_mut() {
4200            substitute_outer_in_expr(e, outer_row, outer_schema, true);
4201        }
4202    }
4203}
4204
4205/// Index of the outer column a reference targets, or `None`. Qualified
4206/// refs (`outer_alias.col`) match the composite outer-schema name. When
4207/// `bare_ok` — i.e. the expression is evaluated with no inner FROM scope
4208/// of its own (a FROM-less LATERAL subquery, or a set-returning FROM
4209/// item's argument) — an *unqualified* ref also resolves to the outer
4210/// scope, matching the bare column against the last segment of each
4211/// outer name, but only when that match is unique (an ambiguous bare ref
4212/// is left for the normal resolver to reject, as PG does).
4213fn outer_col_index(
4214    outer_schema: &[ColumnSchema],
4215    qualifier: Option<&str>,
4216    name: &str,
4217    bare_ok: bool,
4218) -> Option<usize> {
4219    match qualifier {
4220        Some(q) => {
4221            let composite = alloc::format!("{q}.{name}");
4222            outer_schema
4223                .iter()
4224                .position(|sc| sc.name.eq_ignore_ascii_case(&composite))
4225        }
4226        None if bare_ok => {
4227            let mut found = None;
4228            for (i, sc) in outer_schema.iter().enumerate() {
4229                let bare = sc.name.rsplit('.').next().unwrap_or(sc.name.as_str());
4230                if bare.eq_ignore_ascii_case(name) {
4231                    if found.is_some() {
4232                        return None; // ambiguous — don't guess
4233                    }
4234                    found = Some(i);
4235                }
4236            }
4237            found
4238        }
4239        None => None,
4240    }
4241}
4242
4243/// Materialise an outer-row value as a substitutable `Expr`. Array values
4244/// have no scalar `Literal` form, so they become an `ARRAY[…]`
4245/// constructor of element literals — this is what lets a correlated
4246/// `unnest(<outer array>)` expand per row.
4247fn outer_value_to_expr(v: Value<'static>) -> Option<Expr> {
4248    match v {
4249        Value::TextArray(items) => Some(Expr::Array(
4250            items
4251                .into_iter()
4252                .map(|it| {
4253                    Expr::Literal(match it {
4254                        Some(s) => spg_sql::ast::Literal::String(s),
4255                        None => spg_sql::ast::Literal::Null,
4256                    })
4257                })
4258                .collect(),
4259        )),
4260        Value::IntArray(items) => Some(Expr::Array(
4261            items
4262                .into_iter()
4263                .map(|it| {
4264                    Expr::Literal(match it {
4265                        Some(n) => spg_sql::ast::Literal::Integer(i64::from(n)),
4266                        None => spg_sql::ast::Literal::Null,
4267                    })
4268                })
4269                .collect(),
4270        )),
4271        Value::BigIntArray(items) => Some(Expr::Array(
4272            items
4273                .into_iter()
4274                .map(|it| {
4275                    Expr::Literal(match it {
4276                        Some(n) => spg_sql::ast::Literal::Integer(n),
4277                        None => spg_sql::ast::Literal::Null,
4278                    })
4279                })
4280                .collect(),
4281        )),
4282        other => value_to_literal_expr(other).ok(),
4283    }
4284}
4285
4286fn substitute_outer_in_expr(
4287    e: &mut Expr,
4288    outer_row: &Row<'static>,
4289    outer_schema: &[ColumnSchema],
4290    bare_ok: bool,
4291) {
4292    if let Expr::Column(c) = e
4293        && let Some(idx) = outer_col_index(outer_schema, c.qualifier.as_deref(), &c.name, bare_ok)
4294    {
4295        let v = outer_row.values.get(idx).cloned().unwrap_or(Value::Null);
4296        if let Some(lit) = outer_value_to_expr(v) {
4297            *e = lit;
4298            return;
4299        }
4300    }
4301    let mut rec = |e: &mut Expr| substitute_outer_in_expr(e, outer_row, outer_schema, bare_ok);
4302    match e {
4303        Expr::Binary { lhs, rhs, .. } => {
4304            rec(lhs);
4305            rec(rhs);
4306        }
4307        Expr::Unary { expr: inner, .. } => rec(inner),
4308        Expr::FunctionCall { args, .. } => {
4309            for a in args {
4310                rec(a);
4311            }
4312        }
4313        Expr::Cast { expr: inner, .. } => rec(inner),
4314        // v7.38 (read01 LATERAL) — recurse the array/subscript nodes too,
4315        // so `LATERAL unnest(ARRAY[<outer col>, …])` and `arr[<outer>]`
4316        // substitute the outer reference (previously these fell through
4317        // to the no-op arm, stranding the reference unresolved).
4318        Expr::Array(items) => {
4319            for it in items {
4320                rec(it);
4321            }
4322        }
4323        Expr::ArraySubscript { target, index } => {
4324            rec(target);
4325            rec(index);
4326        }
4327        Expr::ArraySlice { target, lo, hi } => {
4328            rec(target);
4329            if let Some(lo) = lo {
4330                rec(lo);
4331            }
4332            if let Some(hi) = hi {
4333                rec(hi);
4334            }
4335        }
4336        Expr::InList { expr, list, .. } => {
4337            rec(expr);
4338            for it in list {
4339                rec(it);
4340            }
4341        }
4342        Expr::Case {
4343            operand,
4344            branches,
4345            else_branch,
4346        } => {
4347            if let Some(op) = operand {
4348                rec(op);
4349            }
4350            for (cond, val) in branches {
4351                rec(cond);
4352                rec(val);
4353            }
4354            if let Some(e) = else_branch {
4355                rec(e);
4356            }
4357        }
4358        _ => {}
4359    }
4360}
4361
4362/// v7.28 (round-22) — single-table predicate pushdown + table-order
4363/// swap analysis, run once before the join pipeline. Splits the WHERE
4364/// conjuncts into per-table predicate lists (the primary plus one per
4365/// INNER peer) so each table can be filtered — with an index seek when
4366/// a conjunct is `col = literal` — BEFORE it joins. Pushed conjuncts
4367/// stay in WHERE too (idempotent), so correctness never depends on the
4368/// pushdown.
4369///
4370/// When the primary has no pushed predicate but the first INNER peer
4371/// does, and the swap is provably safe (equi-joins commute and output
4372/// columns resolve by composite name, so downstream projection is
4373/// order-independent; restricted to the first join with an ON whose
4374/// qualifiers all live in {primary, first peer}), it returns an owned
4375/// FromClause with the primary and that peer swapped — the join then
4376/// starts from the filtered side instead of cloning the whole
4377/// unfiltered primary (e.g. a correlated subquery body like
4378/// `FROM email_analysis e2 JOIN messages m2 … WHERE m2.thread_id =
4379/// '<outer>'`).
4380///
4381/// Returns `(swapped_from, primary_preds, peer_preds)`; `swapped_from`
4382/// is `Some` only when a swap happened, and the caller rebinds `from`
4383/// to it. The returned predicate refs borrow from `where_`.
4384fn analyze_join_pushdown<'w>(
4385    from: &FromClause,
4386    where_: Option<&'w Expr>,
4387) -> (Option<FromClause>, Vec<&'w Expr>, Vec<Vec<&'w Expr>>) {
4388    let primary_alias = from
4389        .primary
4390        .alias
4391        .as_deref()
4392        .unwrap_or(from.primary.name.as_str());
4393    let mut primary_preds: Vec<&Expr> = Vec::new();
4394    let mut peer_preds: Vec<Vec<&Expr>> = alloc::vec![Vec::new(); from.joins.len()];
4395    // v7.37.16 — a RIGHT / FULL OUTER join anywhere in the chain makes
4396    // the primary (left/drive) side nullable: unmatched peer rows emit
4397    // NULL-filled primary columns. A WHERE predicate on the primary must
4398    // then be applied AFTER the join, never pushed onto the primary scan
4399    // (pushing `l.k IS NULL` below `l RIGHT JOIN r` would filter l first
4400    // and wrongly keep all NULL-primary rows). Leaving such predicates in
4401    // the residual WHERE keeps them correct. Peer pushdown is already
4402    // gated to INNER peers below, so it needs no extra guard.
4403    let primary_nullable = from
4404        .joins
4405        .iter()
4406        .any(|j| matches!(j.kind, JoinKind::Right | JoinKind::FullOuter));
4407    if let Some(w) = where_ {
4408        for sub in reorder::split_and_conjunctions(w) {
4409            if expr_has_subquery(sub) || aggregate::contains_aggregate(sub) {
4410                continue;
4411            }
4412            let mut quals: Vec<&str> = Vec::new();
4413            let mut all_qualified = true;
4414            collect_column_qualifiers(sub, &mut quals, &mut all_qualified);
4415            if !all_qualified || quals.is_empty() {
4416                continue;
4417            }
4418            let q0 = quals[0];
4419            if !quals.iter().all(|q| q.eq_ignore_ascii_case(q0)) {
4420                continue;
4421            }
4422            if q0.eq_ignore_ascii_case(primary_alias) {
4423                if !primary_nullable {
4424                    primary_preds.push(sub);
4425                }
4426                continue;
4427            }
4428            for (i, j) in from.joins.iter().enumerate() {
4429                // v7.39 (round 588) — a comma join parses as `Cross`, and a
4430                // cross peer is exactly as non-nullable as an inner one, so a
4431                // single-relation WHERE conjunct belongs on its scan just the
4432                // same. Without this the `b` of `FROM a, b WHERE … b.id < 100`
4433                // was scanned whole.
4434                if matches!(j.kind, JoinKind::Inner | JoinKind::Cross)
4435                    && j.table.lateral_subquery.is_none()
4436                    && q0.eq_ignore_ascii_case(
4437                        j.table.alias.as_deref().unwrap_or(j.table.name.as_str()),
4438                    )
4439                {
4440                    peer_preds[i].push(sub);
4441                    break;
4442                }
4443            }
4444        }
4445    }
4446    // Safety: swapping reorders which table joins FIRST, so it is only
4447    // legal when the FIRST join's ON references no table beyond
4448    // {primary, first peer} (a later peer's ON may name the original
4449    // primary, which must already be in the combined row when that peer
4450    // joins). Restrict to i == 0 AND an ON whose qualifiers all live in
4451    // those two tables.
4452    if primary_preds.is_empty()
4453        && let Some(j0) = from.joins.first()
4454        && matches!(j0.kind, JoinKind::Inner)
4455        && j0.table.lateral_subquery.is_none()
4456        && !peer_preds[0].is_empty()
4457    {
4458        let peer_alias = j0.table.alias.as_deref().unwrap_or(j0.table.name.as_str());
4459        let on_safe = j0.on.as_ref().is_some_and(|on| {
4460            let mut quals: Vec<&str> = Vec::new();
4461            let mut all_q = true;
4462            collect_column_qualifiers(on, &mut quals, &mut all_q);
4463            all_q
4464                && quals.iter().all(|q| {
4465                    q.eq_ignore_ascii_case(primary_alias) || q.eq_ignore_ascii_case(peer_alias)
4466                })
4467        });
4468        if on_safe {
4469            let mut from_owned = from.clone();
4470            core::mem::swap(&mut from_owned.primary, &mut from_owned.joins[0].table);
4471            let primary_preds = peer_preds[0].drain(..).collect();
4472            return (Some(from_owned), primary_preds, peer_preds);
4473        }
4474    }
4475    (None, primary_preds, peer_preds)
4476}
4477
4478/// Build the combined output schema for a join: every primary column
4479/// then every peer column, each qualified `<alias>.<col>` so the
4480/// deferred-join cell lookups and downstream projection resolve by
4481/// composite name.
4482fn build_combined_schema(
4483    primary_alias: &str,
4484    primary_cols: &[ColumnSchema],
4485    joined: &[JoinedPeer<'_>],
4486) -> Vec<ColumnSchema> {
4487    // v7.39 (round 688) — the qualified copy carries what lives outside the
4488    // DataType lattice. `ColumnSchema::new` knows name, type and
4489    // nullability, so a column's collation stopped here and `ORDER BY a.loc`
4490    // over a join sorted by bytes. Proven on the path by panicking inside
4491    // this function and watching the query hit it (round 687).
4492    let carry = |name: alloc::string::String, col: &ColumnSchema| {
4493        let mut c = ColumnSchema::new(name, col.ty, col.nullable);
4494        c.collation_name = col.collation_name.clone();
4495        c.user_enum_type = col.user_enum_type.clone();
4496        c
4497    };
4498    let mut combined_schema: Vec<ColumnSchema> = Vec::new();
4499    for col in primary_cols {
4500        combined_schema.push(carry(alloc::format!("{primary_alias}.{}", col.name), col));
4501    }
4502    for peer in joined {
4503        for col in &peer.cols {
4504            combined_schema.push(carry(alloc::format!("{}.{}", peer.alias, col.name), col));
4505        }
4506    }
4507    combined_schema
4508}
4509
4510/// v7.37.x — helper for `try_count_star_left_anti_join_fast`. Recognises
4511/// `outer.X = inner.Y` (commuted accepted) and returns the column names.
4512fn analyse_join_eq(
4513    on: &Expr,
4514    outer_alias: &str,
4515    inner_alias: &str,
4516) -> Result<Option<(String, String)>, EngineError> {
4517    use spg_sql::ast::BinOp;
4518    let Expr::Binary {
4519        lhs,
4520        op: BinOp::Eq,
4521        rhs,
4522    } = on
4523    else {
4524        return Ok(None);
4525    };
4526    let (Expr::Column(a), Expr::Column(b)) = (lhs.as_ref(), rhs.as_ref()) else {
4527        return Ok(None);
4528    };
4529    fn col_alias(c: &spg_sql::ast::ColumnName) -> Option<&str> {
4530        c.qualifier.as_deref()
4531    }
4532    let pair_o_then_i = (col_alias(a), col_alias(b));
4533    if matches!(pair_o_then_i, (Some(aq), Some(bq))
4534        if aq.eq_ignore_ascii_case(outer_alias) && bq.eq_ignore_ascii_case(inner_alias))
4535    {
4536        return Ok(Some((a.name.clone(), b.name.clone())));
4537    }
4538    if matches!(pair_o_then_i, (Some(aq), Some(bq))
4539        if aq.eq_ignore_ascii_case(inner_alias) && bq.eq_ignore_ascii_case(outer_alias))
4540    {
4541        return Ok(Some((b.name.clone(), a.name.clone())));
4542    }
4543    Ok(None)
4544}
4545
4546/// v7.39 (round 744) — recognise `outer.col = <integer-only expression
4547/// over the inner alias>` (commuted accepted). The expression allowlist
4548/// mirrors `int_only_key_expr`: inner-qualified columns, integer
4549/// literals, Add/Sub/Mul.
4550fn analyse_join_eq_expr(on: &Expr, outer_alias: &str, inner_alias: &str) -> Option<(String, Expr)> {
4551    use spg_sql::ast::BinOp;
4552    let Expr::Binary {
4553        lhs,
4554        op: BinOp::Eq,
4555        rhs,
4556    } = on
4557    else {
4558        return None;
4559    };
4560    fn inner_only_int(e: &Expr, inner_alias: &str) -> bool {
4561        use spg_sql::ast::BinOp;
4562        match e {
4563            Expr::Column(c) => c
4564                .qualifier
4565                .as_deref()
4566                .is_some_and(|q| q.eq_ignore_ascii_case(inner_alias)),
4567            Expr::Literal(spg_sql::ast::Literal::Integer(_)) => true,
4568            Expr::Binary { lhs, op, rhs } => {
4569                matches!(op, BinOp::Add | BinOp::Sub | BinOp::Mul)
4570                    && inner_only_int(lhs, inner_alias)
4571                    && inner_only_int(rhs, inner_alias)
4572            }
4573            _ => false,
4574        }
4575    }
4576    for (a, b) in [(lhs.as_ref(), rhs.as_ref()), (rhs.as_ref(), lhs.as_ref())] {
4577        if let Expr::Column(c) = a
4578            && c.qualifier
4579                .as_deref()
4580                .is_some_and(|q| q.eq_ignore_ascii_case(outer_alias))
4581            && !matches!(b, Expr::Column(_))
4582            && inner_only_int(b, inner_alias)
4583            && expr_mentions_a_column(b)
4584        {
4585            return Some((c.name.clone(), b.clone()));
4586        }
4587    }
4588    None
4589}
4590
4591/// The inner columns an `analyse_join_eq_expr` key reads.
4592fn collect_inner_int_cols(e: &Expr, out: &mut Vec<String>) {
4593    match e {
4594        Expr::Column(c) => out.push(c.name.clone()),
4595        Expr::Binary { lhs, rhs, .. } => {
4596            collect_inner_int_cols(lhs, out);
4597            collect_inner_int_cols(rhs, out);
4598        }
4599        _ => {}
4600    }
4601}
4602
4603/// v7.37.x — recognise `<inner_alias>.<inner_col> IS NULL`.
4604fn is_inner_is_null(e: &Expr, inner_alias: &str, inner_col: &str) -> bool {
4605    let Expr::IsNull { expr, negated } = e else {
4606        return false;
4607    };
4608    if *negated {
4609        return false;
4610    }
4611    let Expr::Column(c) = expr.as_ref() else {
4612        return false;
4613    };
4614    c.qualifier
4615        .as_deref()
4616        .is_some_and(|q| q.eq_ignore_ascii_case(inner_alias))
4617        && c.name.eq_ignore_ascii_case(inner_col)
4618}
4619
4620pub static ANTI_JOIN_FAST_PATH_TRIED: core::sync::atomic::AtomicU64 =
4621    core::sync::atomic::AtomicU64::new(0);
4622pub static ANTI_JOIN_FAST_PATH_FIRED: core::sync::atomic::AtomicU64 =
4623    core::sync::atomic::AtomicU64::new(0);
4624
4625#[cfg(test)]
4626mod r655_rowref_size {
4627    /// v7.39 (round 655/656) — `RowRef` is 64 bytes because its `Tuple`
4628    /// variant carries four slice references for the join path. A scan
4629    /// only ever uses `Owned`, an 8-byte pointer.
4630    ///
4631    /// Round 655 measured what that cost: a scalar aggregate's working
4632    /// memory was O(rows) at ~81 bytes/row — 7.0 MB at 100k, 19.8 at
4633    /// 250k, 40.4 at 500k, 79.6 at 1M — because
4634    /// `run_single_table_aggregate` collected one `RowRef` per surviving
4635    /// row on top of the `Vec<&Row>` it already had. Round 656 gave
4636    /// `AggRows` a `Ptrs` arm that reads those pointers directly:
4637    /// **81 -> 17 bytes/row**; round 657 reserved the survivor vector
4638    /// when there is no WHERE, taking it to **15** — a 5.4x cut overall,
4639    /// measured at all four sizes.
4640    ///
4641    /// The size is pinned because the enum is still built per row inside
4642    /// the loop — on the stack now, so it costs nothing, but a bigger
4643    /// variant would start costing again in registers and moves. A
4644    /// failure here means "re-measure scan RSS at 100k/250k/500k/1M",
4645    /// not "this is forbidden".
4646    fn rowref_stays_small() {
4647        assert_eq!(
4648            core::mem::size_of::<super::RowRef<'_>>(),
4649            64,
4650            "RowRef changed size; a table scan allocates one per row, so \
4651             this multiplies by the row count — re-measure scan RSS"
4652        );
4653    }
4654}