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}