Skip to main content

rete_core/
index.rs

1//! Permutation index set (SPO / POS / OSP) over integer triples (SPEC.md §6).
2//!
3//! The three orders together answer every triple-pattern shape: for any subset
4//! of bound `(s, p, o)` components, at least one permutation sorts those bound
5//! components into a leading prefix, turning the lookup into a range scan.
6//!
7//! Each permutation is **tiled** (format v0.2): consecutive runs of whole
8//! a-groups packed to a byte budget ([`INDEX_TILE_BUDGET`]), each tile a
9//! self-contained [`TripleBlock`]. A bound leading id routes to exactly one
10//! tile (binary search over the tile ranges), then jumps to its a-group via
11//! the tile's lazily-built [`GroupDirectory`]; unbound scans chain all tiles
12//! with per-tile zone-map pruning. On disk each tile is compressed
13//! independently, so a ranged reader can fetch just the tiles a query needs.
14//! v0.1 single-block sections are still read (one tile per permutation).
15
16use std::sync::Arc;
17
18use crate::chunk_cache::{CacheEntry, ChunkCache};
19use crate::triples::{GroupDirectory, Triple, TripleBlock, TripleBlockBuilder};
20
21/// A triple pattern: `None` is an unbound variable, `Some(id)` a bound term.
22pub type Pattern = (Option<u32>, Option<u32>, Option<u32>);
23
24/// Default per-tile budget in (uncompressed) encoded bytes. Tiles are the
25/// independently-compressed, independently-fetchable units of a permutation
26/// section: ~64 KiB matches both zstd's sweet spot and one HTTP range read.
27/// The budget a build uses is [`crate::write_opts::tile_budget`] (`--tile-kib`).
28pub const INDEX_TILE_BUDGET: usize = crate::write_opts::DEFAULT_TILE_BUDGET_BYTES;
29
30/// Geometric tile-prefetch ramp for an unbound (multi-tile) scan (see
31/// [`GraphIndex::scan_iter`]). The first coalesced batch faults this many
32/// tiles; each subsequent batch doubles up to [`PREFETCH_WINDOW_MAX`]. Small
33/// enough that `LIMIT 1` fetches only a few tiles, large enough that a full
34/// scan still coalesces into a handful of range reads.
35const PREFETCH_WINDOW_START: usize = 4;
36const PREFETCH_WINDOW_MAX: usize = 512;
37
38/// Fetches one tile's (uncompressed) block image on demand: the bridge that
39/// lets a remote `GraphIndex` fault tiles in over a `RangeReader` without this
40/// module knowing about I/O. `None` = the fetch failed; the index records the
41/// failure ([`GraphIndex::load_incomplete`]) and the scan sees an empty tile —
42/// callers running over remote data MUST check the flag after evaluating.
43pub type TileLoader = Box<dyn Fn(usize, usize) -> Option<Vec<u8>> + Send + Sync>;
44
45/// Fetches **many** tiles of one section in one round trip: given the section
46/// and the (ascending) indices of the tiles wanted, returns each tile's
47/// uncompressed block image in the same order. The ranged reader implements
48/// this by coalescing byte-adjacent tile ranges into single range reads, so a
49/// full-section scan costs a handful of requests instead of one per tile.
50/// `None` = the batch failed as a whole; callers fall back to the per-tile
51/// [`TileLoader`] (which records per-tile failures).
52pub type TileBulkLoader = Box<dyn Fn(usize, &[usize]) -> Option<Vec<Vec<u8>>> + Send + Sync>;
53
54/// One tile of a permutation section: a fully self-contained [`TripleBlock`]
55/// over a consecutive run of whole a-groups, plus the leading-id range it
56/// covers (for routing without parsing or fetching) and its lazily-built group
57/// directory (for intra-tile point lookups). The block image itself may be
58/// local (built/decoded eagerly) or faulted in by the index's [`TileLoader`].
59pub struct Tile {
60    min_a: u32,
61    max_a: u32,
62    /// Optional **tile synopsis**: the inclusive min/max of the two non-leading
63    /// columns `(min_b, max_b, min_c, max_c)`, read from the directory trailer of
64    /// a synopsis-carrying file. Lets a scan prune this tile by a bound secondary
65    /// component *before* faulting it (remote reads). `None` = no synopsis (an
66    /// older file, or a locally-built/opened tile) — then nothing is pruned early
67    /// and the in-tile zone map prunes after the tile is in hand, as before.
68    pub(crate) syn: Option<(u32, u32, u32, u32)>,
69    /// Encoded byte length of this tile (compressed on-disk size for a remote
70    /// tile, in-memory image size for a local one; 0 = unknown). Feeds the join
71    /// planner's fatness gates without faulting any data.
72    len: u32,
73    /// Build-time staging only: a locally-built or locally-decoded tile's body
74    /// waits here until [`GraphIndex::from_sections`] moves it into the shared
75    /// [`ChunkCache`] (keyed by `(section, tile)`) and clears it. `None`
76    /// afterward, and for a remote tile whose body faults in through the loader.
77    /// The long-lived retention is the cache, not this field.
78    staged: Option<Vec<u8>>,
79}
80
81impl Tile {
82    fn local(min_a: u32, max_a: u32, bytes: Vec<u8>) -> Self {
83        let len = bytes.len().min(u32::MAX as usize) as u32;
84        Tile {
85            min_a,
86            max_a,
87            syn: None,
88            len,
89            staged: Some(bytes),
90        }
91    }
92
93    fn remote(min_a: u32, max_a: u32, syn: Option<(u32, u32, u32, u32)>) -> Self {
94        Tile {
95            min_a,
96            max_a,
97            syn,
98            len: 0,
99            staged: None,
100        }
101    }
102
103    /// Encoded byte length (see the field doc); falls back to a still-staged
104    /// body's size when the directory didn't provide one.
105    pub(crate) fn encoded_len(&self) -> u64 {
106        if self.len > 0 {
107            self.len as u64
108        } else {
109            self.staged.as_ref().map_or(0, |d| d.len() as u64)
110        }
111    }
112
113    /// Leading-component (permuted `a`) range this tile covers, inclusive.
114    pub fn leading_range(&self) -> (u32, u32) {
115        (self.min_a, self.max_a)
116    }
117
118    /// Could this tile hold a triple with the given bound non-leading components,
119    /// per its synopsis? `true` when there is no synopsis (can't rule it out) or
120    /// the bound `b`/`c` fall inside the recorded ranges. Conservative: a `false`
121    /// is a *proven* miss (the in-tile zone map would reject the same tile), so
122    /// skipping the fetch never drops a result.
123    pub(crate) fn syn_admits(&self, pb: Option<u32>, pc: Option<u32>) -> bool {
124        match self.syn {
125            None => true,
126            Some((min_b, max_b, min_c, max_c)) => {
127                let ok = |v: Option<u32>, lo: u32, hi: u32| v.is_none_or(|x| lo <= x && x <= hi);
128                ok(pb, min_b, max_b) && ok(pc, min_c, max_c)
129            }
130        }
131    }
132}
133
134/// Fetches stored leaf blocks of a two-level directory (format `0x06`, SPEC.md
135/// §4.4): absolute `(offset, len)` ranges in, their bytes out, in order — one
136/// batched read. `None` = the batch failed.
137pub type LeafFetch = Arc<dyn Fn(&[(u64, u64)]) -> Option<Vec<Vec<u8>>> + Send + Sync>;
138
139/// One leaf of a permutation's tile directory, as the `0x06` root describes it.
140#[derive(Debug, Clone)]
141pub struct TileLeaf {
142    /// Index of the leaf's first tile in the permutation.
143    pub first: usize,
144    /// Tiles in the leaf.
145    pub n: usize,
146    /// Leading-id range of the leaf (first tile's min, last tile's max).
147    pub min_a: u32,
148    /// See `min_a`.
149    pub max_a: u32,
150    /// Absolute range of the stored leaf block.
151    pub leaf: (u64, u64),
152    /// Absolute offset of the leaf's first tile.
153    pub tile_base: u64,
154}
155
156/// A permutation's tile directory whose entries live in leaf blocks faulted
157/// on demand and kept for the session (format `0x06`). Routing narrows by the
158/// leaves' leading-id ranges first, so a bound lookup faults one leaf.
159/// A faulted tile-directory leaf: each tile and its absolute offset.
160type TileCell = std::sync::OnceLock<Box<[(Tile, u64)]>>;
161
162pub struct LeafedTiles {
163    leaves: Vec<TileLeaf>,
164    cells: Vec<TileCell>,
165    total: usize,
166    total_bytes: u64,
167    fetch: LeafFetch,
168    codec: u8,
169    /// Decompression bound on a leaf.
170    max_block: u64,
171    failed: std::sync::atomic::AtomicBool,
172}
173
174/// The tile standing in for one whose leaf could not be read: it routes
175/// nowhere and its synopsis admits nothing. The failure is recorded.
176static UNREADABLE_TILE: Tile = Tile {
177    min_a: u32::MAX,
178    max_a: u32::MAX,
179    syn: Some((1, 0, 1, 0)),
180    len: 0,
181    staged: None,
182};
183
184impl LeafedTiles {
185    /// Bound every leaf decompression at `max` bytes (from the `0x06` root).
186    pub fn with_max_block(mut self, max: u64) -> Self {
187        self.max_block = max;
188        self
189    }
190
191    /// A leafed directory of `total` tiles and `total_bytes` stored bytes.
192    pub fn new(
193        leaves: Vec<TileLeaf>,
194        total: usize,
195        total_bytes: u64,
196        fetch: LeafFetch,
197        codec: u8,
198    ) -> Self {
199        let cells = leaves.iter().map(|_| std::sync::OnceLock::new()).collect();
200        LeafedTiles {
201            leaves,
202            cells,
203            total,
204            total_bytes,
205            fetch,
206            codec,
207            max_block: u64::MAX,
208            failed: std::sync::atomic::AtomicBool::new(false),
209        }
210    }
211
212    /// How many tiles hold leading id `a`, from the root alone and without
213    /// reading anything (the `rete cost` preview). A resident leaf is counted
214    /// exactly; one that is not is estimated from what the root records of it:
215    /// a leaf whose whole range is `a` holds `n` such tiles, and a leaf shared
216    /// with other ids holds its tiles' share of the range (at least one). Skewed
217    /// ids make the shared-leaf figure rough; [`Self::admitted_estimate`] reads
218    /// instead.
219    pub(crate) fn admitted_unfaulted(&self, a: u32) -> f64 {
220        let mut tiles = 0.0;
221        for (li, lf) in self.leaves.iter().enumerate() {
222            if a < lf.min_a || a > lf.max_a {
223                continue;
224            }
225            tiles += match self.cells[li].get() {
226                Some(v) => v
227                    .iter()
228                    .filter(|(t, _)| t.min_a <= a && a <= t.max_a)
229                    .count() as f64,
230                None if lf.min_a == lf.max_a => lf.n as f64,
231                None => {
232                    let width = (lf.max_a as f64 - lf.min_a as f64) + 1.0;
233                    (lf.n as f64 / width).max(1.0)
234                }
235            };
236        }
237        tiles
238    }
239
240    /// The `(first tile's min_b, last tile's max_b)` of leaf `li`, faulting it;
241    /// `None` when the leaf cannot be read or a tile carries no synopsis.
242    fn leaf_b_span(&self, li: usize) -> Option<(u32, u32)> {
243        self.ensure_leaves(&[li]);
244        let v = self.cells[li].get()?;
245        let first = v.first()?.0.syn?;
246        let last = v.last()?.0.syn?;
247        Some((first.0, last.1))
248    }
249
250    /// Exact count of leaf `li`'s tiles that hold `a` and admit `pb`/`pc`,
251    /// faulting the leaf; the unfaulted figure if it cannot be read.
252    fn count_in_leaf(&self, li: usize, a: u32, pb: Option<u32>, pc: Option<u32>) -> f64 {
253        self.ensure_leaves(&[li]);
254        match self.cells[li].get() {
255            Some(v) => v
256                .iter()
257                .filter(|(t, _)| t.min_a <= a && a <= t.max_a && t.syn_admits(pb, pc))
258                .count() as f64,
259            None => {
260                let lf = &self.leaves[li];
261                let width = (lf.max_a as f64 - lf.min_a as f64) + 1.0;
262                (lf.n as f64 / width).max(1.0)
263            }
264        }
265    }
266
267    /// The tiles that hold leading id `a` and admit the bound second and third
268    /// keys `pb`/`pc`, counted from the root plus a few leaf reads (all kept
269    /// for the session; the lookup that follows needs them anyway).
270    ///
271    /// The planner's row estimates stand on this, so it must neither say "one
272    /// tile" for a predicate that fills a leaf nor ignore a bound class:
273    /// `?s rdf:type <small class>` is a few tiles of an 851 M-row predicate.
274    /// - A leaf shared with other ids (at most two per id) is read and counted
275    ///   exactly, synopsis included.
276    /// - Leaves wholly inside `a` cost nothing when only `a` is bound (each is
277    ///   `n` tiles). With `pb` bound they are ordered by `pb` (tiles are cut in
278    ///   `(a, b, c)` order), so the admitted tiles are a contiguous run that two
279    ///   binary searches over the leaves' `b` spans bound: O(log leaves) reads,
280    ///   then the run's end leaves are counted exactly and the leaves between
281    ///   them are all `n`.
282    pub(crate) fn admitted_estimate(&self, a: u32, pb: Option<u32>, pc: Option<u32>) -> f64 {
283        let first = self.leaves.partition_point(|lf| lf.max_a < a);
284        let mut last = first;
285        while last < self.leaves.len() && self.leaves[last].min_a <= a {
286            last += 1;
287        }
288        if first >= last {
289            return 0.0;
290        }
291        let mut tiles = 0.0;
292        // Pure leaves are contiguous: every shared one is first or last.
293        let pure = |li: usize| self.leaves[li].min_a == a && self.leaves[li].max_a == a;
294        let (mut p0, mut p1) = (first, last);
295        if !pure(first) {
296            tiles += self.count_in_leaf(first, a, pb, pc);
297            p0 += 1;
298        }
299        if p1 > p0 && !pure(p1 - 1) {
300            tiles += self.count_in_leaf(p1 - 1, a, pb, pc);
301            p1 -= 1;
302        }
303        if p0 >= p1 {
304            return tiles;
305        }
306        // Wholly-`a` leaves p0..p1.
307        let Some(b) = pb else {
308            return tiles + self.leaves[p0..p1].iter().map(|l| l.n as f64).sum::<f64>();
309        };
310        // First leaf whose last tile reaches `b`; first leaf that starts past it.
311        let search = |past: bool| -> Option<usize> {
312            let (mut lo, mut hi) = (p0, p1);
313            while lo < hi {
314                let mid = lo + (hi - lo) / 2;
315                let (min_b, max_b) = self.leaf_b_span(mid)?;
316                if if past { min_b > b } else { max_b >= b } {
317                    hi = mid;
318                } else {
319                    lo = mid + 1;
320                }
321            }
322            Some(lo)
323        };
324        let (Some(f), Some(g)) = (search(false), search(true)) else {
325            // A leaf could not be read: no narrowing, count them all.
326            return tiles + self.leaves[p0..p1].iter().map(|l| l.n as f64).sum::<f64>();
327        };
328        if f >= g {
329            return tiles;
330        }
331        // Leaves f..g may hold `b`; those strictly between the ends are all `b`.
332        tiles += self.count_in_leaf(f, a, pb, pc);
333        if g - 1 > f {
334            tiles += self.count_in_leaf(g - 1, a, pb, pc);
335            tiles += self.leaves[f + 1..g - 1]
336                .iter()
337                .map(|l| l.n as f64)
338                .sum::<f64>();
339        }
340        tiles
341    }
342
343    fn leaf_of(&self, ti: usize) -> usize {
344        self.leaves
345            .partition_point(|l| l.first <= ti)
346            .saturating_sub(1)
347    }
348
349    fn decode_leaf(&self, li: usize, stored: &[u8]) -> Option<Box<[(Tile, u64)]>> {
350        let l = &self.leaves[li];
351        let raw = crate::file::decompress_bounded(self.codec, stored, self.max_block).ok()?;
352        let ents = crate::v06::decode_tile_leaf(&raw, l.n).ok()?;
353        let mut at = l.tile_base;
354        let mut out = Vec::with_capacity(ents.len());
355        for e in ents {
356            let mut t = Tile::remote(e.min_a, e.max_a, Some(e.syn));
357            t.len = e.clen.min(u32::MAX as u64) as u32;
358            out.push((t, at));
359            at = at.checked_add(e.clen)?;
360        }
361        Some(out.into_boxed_slice())
362    }
363
364    /// Fault every not-yet-resident leaf of `lis` in one batched read.
365    pub fn ensure_leaves(&self, lis: &[usize]) {
366        // Callers pass one leaf per pattern or term, so the same leaf can come
367        // many times over (a join probing 460 k subjects of one leaf): fetch
368        // each leaf once.
369        let mut missing: Vec<usize> = lis
370            .iter()
371            .copied()
372            .filter(|&li| li < self.cells.len() && self.cells[li].get().is_none())
373            .collect();
374        missing.sort_unstable();
375        missing.dedup();
376        if missing.is_empty() {
377            return;
378        }
379        let ranges: Vec<(u64, u64)> = missing.iter().map(|&li| self.leaves[li].leaf).collect();
380        match (self.fetch)(&ranges) {
381            Some(blobs) if blobs.len() == missing.len() => {
382                for (&li, b) in missing.iter().zip(blobs) {
383                    match self.decode_leaf(li, &b) {
384                        Some(v) => {
385                            let _ = self.cells[li].set(v);
386                        }
387                        None => self
388                            .failed
389                            .store(true, std::sync::atomic::Ordering::Relaxed),
390                    }
391                }
392            }
393            _ => self
394                .failed
395                .store(true, std::sync::atomic::Ordering::Relaxed),
396        }
397    }
398
399    /// Fault the leaves covering tiles `[start, end)`.
400    pub fn ensure_span(&self, start: usize, end: usize) {
401        if start >= end || start >= self.total {
402            return;
403        }
404        let a = self.leaf_of(start);
405        let b = self.leaf_of(end.min(self.total) - 1);
406        self.ensure_leaves(&(a..=b).collect::<Vec<_>>());
407    }
408
409    fn entry(&self, ti: usize) -> Option<&(Tile, u64)> {
410        if ti >= self.total {
411            return None;
412        }
413        let li = self.leaf_of(ti);
414        if self.cells[li].get().is_none() {
415            self.ensure_leaves(&[li]);
416        }
417        self.cells[li].get()?.get(ti - self.leaves[li].first)
418    }
419
420    /// Tile `ti`'s absolute stored range, faulting its leaf.
421    pub fn range(&self, ti: usize) -> Option<(u64, u64)> {
422        self.entry(ti).map(|(t, off)| (*off, t.len as u64))
423    }
424
425    /// Did a leaf fault fail?
426    pub fn load_failed(&self) -> bool {
427        self.failed.load(std::sync::atomic::Ordering::Relaxed)
428    }
429
430    /// Forget a recorded leaf-fault failure.
431    pub fn reset_load_failure(&self) {
432        self.failed
433            .store(false, std::sync::atomic::Ordering::Relaxed);
434    }
435
436    /// Bytes held by the leaves faulted so far.
437    fn resident_bytes(&self) -> u64 {
438        let n: usize = self
439            .cells
440            .iter()
441            .filter_map(|c| c.get())
442            .map(|v| v.len())
443            .sum();
444        (n * (std::mem::size_of::<Tile>() + 8)) as u64
445            + (self.leaves.len() * std::mem::size_of::<TileLeaf>()) as u64
446    }
447}
448
449/// A permutation's tile directory: resident in full (`0x05`, built indexes),
450/// or two-level with leaves faulted on demand (`0x06`).
451pub enum TileList {
452    /// Every tile descriptor resident.
453    Full(Vec<Tile>),
454    /// Root resident, leaves on demand.
455    Leafed(Arc<LeafedTiles>),
456}
457
458impl Default for TileList {
459    fn default() -> Self {
460        TileList::Full(Vec::new())
461    }
462}
463
464impl std::ops::Index<usize> for TileList {
465    type Output = Tile;
466    fn index(&self, ti: usize) -> &Tile {
467        match self {
468            TileList::Full(v) => &v[ti],
469            TileList::Leafed(l) => match l.entry(ti) {
470                Some((t, _)) => t,
471                None => {
472                    l.failed.store(true, std::sync::atomic::Ordering::Relaxed);
473                    &UNREADABLE_TILE
474                }
475            },
476        }
477    }
478}
479
480impl TileList {
481    /// Number of tiles.
482    pub fn len(&self) -> usize {
483        match self {
484            TileList::Full(v) => v.len(),
485            TileList::Leafed(l) => l.total,
486        }
487    }
488
489    /// No tiles?
490    pub fn is_empty(&self) -> bool {
491        self.len() == 0
492    }
493
494    /// The resident descriptors of a full list (`None` for a leafed one).
495    pub fn as_full(&self) -> Option<&[Tile]> {
496        match self {
497            TileList::Full(v) => Some(v),
498            TileList::Leafed(_) => None,
499        }
500    }
501
502    /// Every tile, in order, faulting all leaves first (one batch).
503    pub fn iter(&self) -> impl Iterator<Item = &Tile> {
504        if let TileList::Leafed(l) = self {
505            l.ensure_span(0, l.total);
506        }
507        (0..self.len()).map(move |i| &self[i])
508    }
509
510    /// The first tile, if any.
511    pub fn first(&self) -> Option<&Tile> {
512        (!self.is_empty()).then(|| &self[0])
513    }
514
515    /// The last tile, if any.
516    pub fn last(&self) -> Option<&Tile> {
517        (!self.is_empty()).then(|| &self[self.len() - 1])
518    }
519
520    /// Fault the leaves covering `[start, end)` in one batch (a no-op on a full
521    /// list).
522    pub fn ensure_span(&self, start: usize, end: usize) {
523        if let TileList::Leafed(l) = self {
524            l.ensure_span(start, end);
525        }
526    }
527
528    /// Index of the first tile whose `max_a >= a` (`len()` if none). A leafed
529    /// list narrows to the one leaf that can hold it before faulting.
530    pub fn first_max_ge(&self, a: u32) -> usize {
531        match self {
532            TileList::Full(v) => v.partition_point(|t| t.max_a < a),
533            TileList::Leafed(l) => {
534                let li = l.leaves.partition_point(|lf| lf.max_a < a);
535                if li >= l.leaves.len() {
536                    return l.total;
537                }
538                let lf = &l.leaves[li];
539                l.ensure_leaves(&[li]);
540                // Binary search inside the faulted leaf: stepping with `self[i]`
541                // costs a leaf lookup per tile, and a lookup runs this per probe
542                // row (94,700 probes x a leaf of thousands of tiles was 126 s).
543                match l.cells[li].get() {
544                    Some(v) => lf.first + v.partition_point(|(t, _)| t.max_a < a),
545                    None => lf.first,
546                }
547            }
548        }
549    }
550
551    /// Index of the first tile whose `min_a > a` (`len()` if none): the end of
552    /// the run of tiles that can hold leading id `a`. Faults one leaf at most.
553    pub fn first_min_gt(&self, a: u32) -> usize {
554        match self {
555            TileList::Full(v) => v.partition_point(|t| t.min_a <= a),
556            TileList::Leafed(l) => {
557                let next = l.leaves.partition_point(|lf| lf.min_a <= a);
558                if next == 0 {
559                    return 0;
560                }
561                let li = next - 1;
562                let lf = &l.leaves[li];
563                l.ensure_leaves(&[li]);
564                match l.cells[li].get() {
565                    Some(v) => lf.first + v.partition_point(|(t, _)| t.min_a <= a),
566                    None => lf.first,
567                }
568            }
569        }
570    }
571
572    /// Stored bytes of every tile (from the root on a leafed list).
573    pub fn total_bytes(&self) -> u64 {
574        match self {
575            TileList::Full(v) => v.iter().map(Tile::encoded_len).sum(),
576            TileList::Leafed(l) => l.total_bytes,
577        }
578    }
579
580    /// Stored bytes of the tiles whose leading range holds `a`, **without
581    /// faulting**: exact where the leaf is resident, else the mean tile size times the
582    /// tiles the root says hold it — the planner's cost
583    /// estimate must not cost a round trip.
584    pub fn routed_bytes(&self, a: u32) -> u64 {
585        match self {
586            TileList::Full(v) => v
587                .iter()
588                .filter(|t| t.min_a <= a && a <= t.max_a)
589                .map(Tile::encoded_len)
590                .sum(),
591            TileList::Leafed(l) => {
592                let lis: Vec<usize> = l
593                    .leaves
594                    .iter()
595                    .enumerate()
596                    .filter(|(_, lf)| lf.min_a <= a && a <= lf.max_a)
597                    .map(|(i, _)| i)
598                    .collect();
599                if lis.is_empty() {
600                    return 0;
601                }
602                if lis.iter().all(|&li| l.cells[li].get().is_some()) {
603                    lis.iter()
604                        .flat_map(|&li| l.cells[li].get().into_iter().flat_map(|v| v.iter()))
605                        .filter(|(t, _)| t.min_a <= a && a <= t.max_a)
606                        .map(|(t, _)| t.encoded_len())
607                        .sum()
608                } else {
609                    // The mean tile size times the tiles the root says hold
610                    // `a`: a predicate filling many tiles is many tiles' bytes.
611                    let mean = l.total_bytes as f64 / l.total.max(1) as f64;
612                    (l.admitted_unfaulted(a).max(1.0) * mean) as u64
613                }
614            }
615        }
616    }
617
618    /// Is the leaf holding tile `ti` resident (always, for a full list)?
619    pub fn resident_at(&self, ti: usize) -> bool {
620        match self {
621            TileList::Full(_) => true,
622            TileList::Leafed(l) => ti < l.total && l.cells[l.leaf_of(ti)].get().is_some(),
623        }
624    }
625
626    /// The absolute stored range of tile `ti` (leafed lists only).
627    pub fn range(&self, ti: usize) -> Option<(u64, u64)> {
628        match self {
629            TileList::Full(_) => None,
630            TileList::Leafed(l) => l.range(ti),
631        }
632    }
633
634    fn load_failed(&self) -> bool {
635        matches!(self, TileList::Leafed(l) if l.load_failed())
636    }
637
638    fn reset_load_failure(&self) {
639        if let TileList::Leafed(l) = self {
640            l.reset_load_failure();
641        }
642    }
643
644    fn directory_bytes(&self) -> u64 {
645        match self {
646            TileList::Full(v) => (v.capacity() * std::mem::size_of::<Tile>()) as u64,
647            TileList::Leafed(l) => l.resident_bytes(),
648        }
649    }
650}
651
652/// What a routed scan of one pattern will touch, from the tile directories
653/// alone — see [`GraphIndex::scan_plan`].
654#[derive(Debug, Clone, Copy, PartialEq, Eq)]
655pub struct ScanPlan {
656    /// The permutation the scan routes to (always inside [`PermSet::CORE`]).
657    pub permutation: IndexPermutation,
658    /// Tiles in that permutation's section — the denominator.
659    pub tiles_total: usize,
660    /// Tiles left after routing on the bound leading component.
661    pub tiles_routed: usize,
662    /// Tiles left after the synopsis prune — what the scan will actually fetch.
663    pub tiles_admitted: usize,
664    /// Encoded (on-disk) bytes of the admitted tiles: the index-side upper
665    /// bound on the scan's range reads.
666    pub tile_bytes: u64,
667    /// Encoded bytes of the whole section — what an unpruned scan would read.
668    pub section_bytes: u64,
669}
670
671/// Which stored permutation to scan. The full **six** orders (a SPARQL engine's
672/// classic set, as in QLever): together they sort the triples on *every* prefix
673/// of `(s, p, o)` columns, so for any bound-component prefix and any free
674/// component there is a permutation that routes on the prefix **and** yields the
675/// stream sorted on that free component — the precondition for a merge join.
676#[derive(Debug, Clone, Copy, PartialEq, Eq)]
677pub enum IndexPermutation {
678    Spo,
679    Sop,
680    Pso,
681    Pos,
682    Osp,
683    Ops,
684}
685
686/// The number of permutations the format can address (the width of every
687/// per-permutation array). How many a *given file* stores is its [`PermSet`].
688pub(crate) const NUM_PERMS: usize = 6;
689
690/// The six permutations in section order — the canonical iteration list. The
691/// original three (SPO, POS, OSP) lead, so a default scan's tie-break (and its
692/// provenance) is unchanged from the 3-permutation format; the three added orders
693/// (SOP, PSO, OPS) follow and exist to give a merge join a stream sorted on the
694/// join key.
695pub(crate) const ALL_PERMS: [IndexPermutation; NUM_PERMS] = [
696    IndexPermutation::Spo,
697    IndexPermutation::Pos,
698    IndexPermutation::Osp,
699    IndexPermutation::Sop,
700    IndexPermutation::Pso,
701    IndexPermutation::Ops,
702];
703
704/// **Which** permutations a file stores — a 6-bit set over
705/// [`IndexPermutation::section_index`], recorded in the header so a reader
706/// never has to assume six.
707///
708/// Every legal set is a superset of [`PermSet::CORE`] = {SPO, POS, OSP}. That
709/// is not a convention: those three **tie the longest bound prefix on all eight
710/// triple-pattern shapes**, so [`GraphIndex::best_permutation`] never selects
711/// outside them and a lean file routes every pattern exactly where a full one
712/// does — same tiles, same rows. What the other three (SOP, PSO, OPS) add is a
713/// *sort order*: they let [`GraphIndex::permutation_sorted_on`] hand a
714/// sort-merge join a co-sorted stream for three of the twelve
715/// (bound-set, join-column) shapes it otherwise has to hash-join. Absent, the
716/// merge seed declines and the hash path answers — the cost is a fast path,
717/// never correctness (`bgp::try_merge_join` already returns `Option`).
718///
719/// `CORE` is the smallest legal set, [`PermSet::ALL`] is the default and what
720/// every file written before this existed carries.
721#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
722pub struct PermSet(u8);
723
724impl Default for PermSet {
725    fn default() -> Self {
726        PermSet::ALL
727    }
728}
729
730impl PermSet {
731    /// All six permutations — the default, and what a `0` header byte means.
732    pub const ALL: PermSet = PermSet(0b0011_1111);
733    /// The three routing permutations {SPO, POS, OSP}: every pattern shape
734    /// routes as well as it does with six, no merge-join sort orders.
735    pub const CORE: PermSet = PermSet(0b0000_0111);
736
737    /// Parse a raw mask. `Err` when it is not a superset of [`PermSet::CORE`]
738    /// or addresses a permutation this build has no section index for — a
739    /// pattern would then route to a permutation the file does not carry, which
740    /// is the one failure mode that must never be silent.
741    pub fn from_bits(bits: u8) -> Result<PermSet, &'static str> {
742        if bits & !PermSet::ALL.0 != 0 {
743            return Err("permutation mask addresses more than six permutations");
744        }
745        if bits & PermSet::CORE.0 != PermSet::CORE.0 {
746            return Err("permutation mask must contain SPO, POS and OSP");
747        }
748        Ok(PermSet(bits))
749    }
750
751    /// The raw 6-bit mask.
752    pub fn bits(self) -> u8 {
753        self.0
754    }
755
756    /// Is `perm` stored in this file?
757    pub fn contains(self, perm: IndexPermutation) -> bool {
758        self.0 & (1 << perm.section_index()) != 0
759    }
760
761    /// How many permutation sections the file's index container holds.
762    pub fn len(self) -> usize {
763        self.0.count_ones() as usize
764    }
765
766    /// Never true: every legal set contains [`PermSet::CORE`]. Present because
767    /// a public `len` without it is a clippy error, and the honest answer is
768    /// more useful than an `allow`.
769    pub fn is_empty(self) -> bool {
770        false
771    }
772
773    /// The stored permutations in [`ALL_PERMS`] (= container) order.
774    pub fn iter(self) -> impl Iterator<Item = IndexPermutation> {
775        ALL_PERMS.into_iter().filter(move |p| self.contains(*p))
776    }
777
778    /// `perm`'s position **inside the index container**, which is its rank
779    /// among the stored permutations — not [`IndexPermutation::section_index`],
780    /// which is its slot in the format's fixed six-wide addressing. The two
781    /// coincide only for [`PermSet::ALL`].
782    pub fn position(self, perm: IndexPermutation) -> Option<usize> {
783        self.contains(perm)
784            .then(|| (self.0 & ((1u8 << perm.section_index()) - 1)).count_ones() as usize)
785    }
786
787    /// Names in container order, e.g. `["SPO", "POS", "OSP"]`.
788    pub fn names(self) -> Vec<&'static str> {
789        self.iter().map(|p| p.name()).collect()
790    }
791
792    /// Does this set carry the three merge-join orders?
793    pub fn has_merge_orders(self) -> bool {
794        self == PermSet::ALL
795    }
796}
797
798impl IndexPermutation {
799    /// Stable display name used in CLI diagnostics and provenance records.
800    pub fn name(self) -> &'static str {
801        match self {
802            IndexPermutation::Spo => "SPO",
803            IndexPermutation::Sop => "SOP",
804            IndexPermutation::Pso => "PSO",
805            IndexPermutation::Pos => "POS",
806            IndexPermutation::Osp => "OSP",
807            IndexPermutation::Ops => "OPS",
808        }
809    }
810
811    /// Section number inside the stored permutation container.
812    pub fn section_index(self) -> usize {
813        match self {
814            IndexPermutation::Spo => 0,
815            IndexPermutation::Pos => 1,
816            IndexPermutation::Osp => 2,
817            IndexPermutation::Sop => 3,
818            IndexPermutation::Pso => 4,
819            IndexPermutation::Ops => 5,
820        }
821    }
822
823    /// The canonical `(s, p, o)` roles in this permutation's `(a, b, c)` slots,
824    /// as indices (0=s, 1=p, 2=o). The single source of truth for `forward` /
825    /// `back` / `order_pattern`.
826    pub(crate) const fn roles(self) -> [usize; 3] {
827        match self {
828            IndexPermutation::Spo => [0, 1, 2],
829            IndexPermutation::Sop => [0, 2, 1],
830            IndexPermutation::Pso => [1, 0, 2],
831            IndexPermutation::Pos => [1, 2, 0],
832            IndexPermutation::Osp => [2, 0, 1],
833            IndexPermutation::Ops => [2, 1, 0],
834        }
835    }
836
837    /// Map a canonical `(s, p, o)` triple into this permutation's `(a, b, c)`.
838    pub(crate) fn forward(self, t: Triple) -> Triple {
839        let c = [t.0, t.1, t.2];
840        let r = self.roles();
841        (c[r[0]], c[r[1]], c[r[2]])
842    }
843
844    /// Map a stored `(a, b, c)` back to canonical `(s, p, o)`.
845    pub(crate) fn back(self, abc: Triple) -> Triple {
846        let a = [abc.0, abc.1, abc.2];
847        let r = self.roles();
848        let mut out = [0u32; 3];
849        out[r[0]] = a[0];
850        out[r[1]] = a[1];
851        out[r[2]] = a[2];
852        (out[0], out[1], out[2])
853    }
854
855    /// The pattern's bound components in this permutation's component order.
856    pub(crate) fn order_pattern(self, p: Pattern) -> [Option<u32>; 3] {
857        let c = [p.0, p.1, p.2];
858        let r = self.roles();
859        [c[r[0]], c[r[1]], c[r[2]]]
860    }
861
862    /// Length of the leading run of bound components — higher is more selective.
863    fn leading_bound(self, pat: Pattern) -> usize {
864        self.order_pattern(pat)
865            .iter()
866            .take_while(|c| c.is_some())
867            .count()
868    }
869}
870
871/// Build a [`GraphIndex`] from canonical `(s, p, o)` integer triples.
872pub struct GraphIndexBuilder {
873    triples: Vec<Triple>,
874    tile_budget: usize,
875    perms: PermSet,
876    /// Report each finished permutation to [`crate::build_progress`].
877    progress: bool,
878}
879
880impl Default for GraphIndexBuilder {
881    fn default() -> Self {
882        Self {
883            triples: Vec::new(),
884            tile_budget: crate::write_opts::tile_budget(),
885            perms: PermSet::ALL,
886            progress: false,
887        }
888    }
889}
890
891impl GraphIndexBuilder {
892    pub fn new() -> Self {
893        Self::default()
894    }
895
896    /// Build directly from an owned triple vector — moves it in, avoiding the
897    /// per-triple `push` copy when the caller already has the id-triples in a
898    /// `Vec` (the assembly path, where they double as the pyramid input).
899    pub fn from_triples(triples: Vec<Triple>) -> Self {
900        Self {
901            triples,
902            ..Self::default()
903        }
904    }
905
906    /// Override the per-tile byte budget (tests use tiny budgets to force
907    /// multi-tile sections on small data).
908    pub fn with_tile_budget(mut self, bytes: usize) -> Self {
909        self.tile_budget = bytes.max(1);
910        self
911    }
912
913    /// Build only the permutations in `perms` (default [`PermSet::ALL`]).
914    /// Sections outside the set come out empty and are not written to the file.
915    pub fn with_perms(mut self, perms: PermSet) -> Self {
916        self.perms = perms;
917        self
918    }
919
920    /// Advance [`crate::build_progress`] by the triple count as each
921    /// permutation is built (the in-RAM build's `index_permutations` phase).
922    pub(crate) fn with_progress(mut self) -> Self {
923        self.progress = true;
924        self
925    }
926
927    pub fn push(&mut self, t: Triple) {
928        self.triples.push(t);
929    }
930
931    /// Build the six permutations **one at a time** (each permutation's internal
932    /// sort still uses every core), so only a single permuted copy of the triples
933    /// is resident at once instead of all six concurrently. Trades the cross-
934    /// permutation parallelism of [`Self::build`] for a much lower peak RAM — the
935    /// large-graph low-memory path. Output is byte-identical to [`Self::build`].
936    pub fn build_seq(self) -> GraphIndex {
937        let triples = &self.triples;
938        let budget = self.tile_budget;
939        let progress = self.progress;
940        let wanted: Vec<IndexPermutation> = self.perms.iter().collect();
941        // Permute (parallel) + sort (parallel inside `build_tiles`) one permutation
942        // at a time, so only a single permuted copy of the triples is resident — but
943        // every core is still busy. Process permutations in batches of two to
944        // overlap one permutation's tile-building with the next one's sort while
945        // keeping the resident permuted copies to two. Byte-identical to `build`.
946        let build_one = |perm: IndexPermutation| -> Vec<Tile> {
947            #[cfg(feature = "parallel")]
948            let permuted: Vec<Triple> = {
949                use rayon::prelude::*;
950                triples.par_iter().map(|&t| perm.forward(t)).collect()
951            };
952            #[cfg(not(feature = "parallel"))]
953            let permuted: Vec<Triple> = triples.iter().map(|&t| perm.forward(t)).collect();
954            let tiles = build_tiles(permuted, budget);
955            if progress {
956                crate::build_progress::advance(triples.len() as u64);
957            }
958            tiles
959        };
960        #[cfg(feature = "parallel")]
961        let built: Vec<Vec<Tile>> = {
962            use rayon::prelude::*;
963            let mut built: Vec<Vec<Tile>> = Vec::with_capacity(wanted.len());
964            for chunk in wanted.chunks(2) {
965                built.extend(
966                    chunk
967                        .to_vec()
968                        .into_par_iter()
969                        .map(build_one)
970                        .collect::<Vec<_>>(),
971                );
972            }
973            built
974        };
975        #[cfg(not(feature = "parallel"))]
976        let built: Vec<Vec<Tile>> = wanted.iter().copied().map(build_one).collect();
977        GraphIndex::from_sections(scatter(&wanted, built), self.perms)
978    }
979
980    pub fn build(self) -> GraphIndex {
981        let wanted: Vec<IndexPermutation> = self.perms.iter().collect();
982        let triples = &self.triples;
983        let budget = self.tile_budget;
984        let progress = self.progress;
985        // The permutations are independent — permute + sort + tile each.
986        let build_one = move |perm: IndexPermutation| -> Vec<Tile> {
987            let permuted: Vec<Triple> = triples.iter().map(|&t| perm.forward(t)).collect();
988            let tiles = build_tiles(permuted, budget);
989            if progress {
990                crate::build_progress::advance(triples.len() as u64);
991            }
992            tiles
993        };
994        // Build the permutations concurrently when the `parallel` feature is on
995        // (they share no state); the per-permutation sort inside is also
996        // parallel. Output is byte-identical to the serial path.
997        #[cfg(feature = "parallel")]
998        let built: Vec<Vec<Tile>> = {
999            use rayon::iter::{IntoParallelIterator, ParallelIterator};
1000            wanted.clone().into_par_iter().map(build_one).collect()
1001        };
1002        #[cfg(not(feature = "parallel"))]
1003        let built: Vec<Vec<Tile>> = wanted.iter().copied().map(build_one).collect();
1004        GraphIndex::from_sections(scatter(&wanted, built), self.perms)
1005    }
1006}
1007
1008/// Place each built section at its permutation's fixed [`IndexPermutation::section_index`]
1009/// slot, leaving unbuilt permutations empty. The array stays six wide however
1010/// few permutations a file carries, so every `sections[perm.section_index()]`
1011/// in the engine keeps working unchanged.
1012fn scatter(wanted: &[IndexPermutation], built: Vec<Vec<Tile>>) -> [Vec<Tile>; NUM_PERMS] {
1013    let mut sections: [Vec<Tile>; NUM_PERMS] = Default::default();
1014    for (perm, tiles) in wanted.iter().zip(built) {
1015        sections[perm.section_index()] = tiles;
1016    }
1017    sections
1018}
1019
1020/// The encoded varint length of `v` (LEB128).
1021fn varint_len(mut v: u64) -> usize {
1022    let mut n = 1;
1023    while v >= 0x80 {
1024        v >>= 7;
1025        n += 1;
1026    }
1027    n
1028}
1029
1030/// Incremental encoded-size accounting for one a-group of a tiled section —
1031/// the running-delta chain of [`build_tiles`], streamable triple by triple.
1032/// Shared by the in-memory tiler here and the external build's streaming
1033/// tiler ([`crate::extbuild`]) so both choose IDENTICAL tile boundaries and
1034/// their outputs stay byte-identical.
1035pub(crate) struct GroupSizer {
1036    /// finalized contributions: the a-delta plus every closed b-run
1037    size: usize,
1038    num_b: u64,
1039    cur_b: u32,
1040    /// the open b-run's c count / last c
1041    num_c: u64,
1042    prev_c: u32,
1043    empty: bool,
1044}
1045
1046impl GroupSizer {
1047    /// Start a group for leading id `a` whose delta base is `prev_a` (the
1048    /// previous group's leading id; `a` again for a mid-group continuation).
1049    pub(crate) fn start(a: u32, prev_a: u32) -> Self {
1050        GroupSizer {
1051            size: varint_len((a - prev_a) as u64),
1052            num_b: 0,
1053            cur_b: 0,
1054            num_c: 0,
1055            prev_c: 0,
1056            empty: true,
1057        }
1058    }
1059
1060    /// Account one `(b, c)` of this group; returns the group's total encoded
1061    /// size so far (open count varints included).
1062    pub(crate) fn push(&mut self, b: u32, c: u32) -> usize {
1063        if self.empty || b != self.cur_b {
1064            if self.empty {
1065                self.size += varint_len(b as u64); // first b-run: delta from 0
1066            } else {
1067                self.size += varint_len(self.num_c); // close the previous b-run
1068                self.size += varint_len((b - self.cur_b) as u64);
1069            }
1070            self.cur_b = b;
1071            self.num_c = 0;
1072            self.prev_c = 0;
1073            self.num_b += 1;
1074            self.empty = false;
1075        }
1076        self.size += varint_len((c - self.prev_c) as u64);
1077        self.prev_c = c;
1078        self.num_c += 1;
1079        self.total()
1080    }
1081
1082    /// The group's total encoded size so far.
1083    pub(crate) fn total(&self) -> usize {
1084        self.size
1085            + if self.empty {
1086                0
1087            } else {
1088                varint_len(self.num_c)
1089            }
1090            + varint_len(self.num_b)
1091    }
1092}
1093
1094/// Split sorted, deduped permuted triples into size-targeted tiles: append
1095/// whole a-groups until the (estimated, near-exact) encoded size would exceed
1096/// `budget`, then flush — and additionally cut WITHIN an a-group whose own
1097/// running size exceeds the budget (a mega-group: one predicate or class
1098/// carrying a large share of the graph, e.g. a 2B-triple `cites` predicate in
1099/// POS). Split slices become consecutive tiles sharing the leading id — a
1100/// bound leading id routes to the whole run of covering tiles (see
1101/// [`GraphIndex::tile_span`]) — which bounds both builder memory and the
1102/// bytes a remote reader faults for one lookup. Boundary accounting lives in
1103/// [`GroupSizer`], shared with the external build for byte-identity.
1104fn build_tiles(mut triples: Vec<Triple>, budget: usize) -> Vec<Tile> {
1105    #[cfg(feature = "parallel")]
1106    {
1107        use rayon::slice::ParallelSliceMut;
1108        triples.par_sort_unstable();
1109    }
1110    #[cfg(not(feature = "parallel"))]
1111    triples.sort_unstable();
1112    triples.dedup();
1113    if triples.is_empty() {
1114        return Vec::new();
1115    }
1116
1117    let make_tile = |run: &[Triple]| -> Tile {
1118        let mut b = TripleBlockBuilder::new();
1119        for &t in run {
1120            b.push(t);
1121        }
1122        Tile::local(run[0].0, run[run.len() - 1].0, b.build())
1123    };
1124
1125    let mut tiles = Vec::new();
1126    let mut tile_start = 0usize;
1127    let mut tile_size = 0usize; // completed groups in the current tile
1128    let mut prev_a = 0u32;
1129    let mut i = 0usize;
1130    while i < triples.len() {
1131        let a = triples[i].0;
1132        let mut slice_start = i; // group start, or the last mid-group cut
1133        let mut sizer = GroupSizer::start(a, prev_a);
1134        let mut gtotal = 0usize;
1135        while i < triples.len() && triples[i].0 == a {
1136            gtotal = sizer.push(triples[i].1, triples[i].2);
1137            i += 1;
1138            if gtotal > budget {
1139                // Mega-group cut: everything buffered — completed groups plus
1140                // the slice measured so far — becomes one tile; the group
1141                // continues in a fresh chain with `a` as its own delta base.
1142                tiles.push(make_tile(&triples[tile_start..i]));
1143                tile_start = i;
1144                tile_size = 0;
1145                prev_a = a;
1146                slice_start = i;
1147                sizer = GroupSizer::start(a, a);
1148                gtotal = 0;
1149            }
1150        }
1151        if slice_start == i {
1152            continue; // the cut landed exactly on the group's end
1153        }
1154        if slice_start > tile_start && tile_size + gtotal > budget {
1155            tiles.push(make_tile(&triples[tile_start..slice_start]));
1156            tile_start = slice_start;
1157            tile_size = 0;
1158        }
1159        tile_size += gtotal;
1160        prev_a = a;
1161    }
1162    if tile_start < triples.len() {
1163        tiles.push(make_tile(&triples[tile_start..]));
1164    }
1165    tiles
1166}
1167
1168/// The six tiled permutation sections, queryable by triple pattern.
1169pub struct GraphIndex {
1170    /// Tiles per permutation (SPO, SOP, PSO, POS, OSP, OPS — see [`ALL_PERMS`]),
1171    /// ascending in their leading-id ranges; consecutive tiles may share a
1172    /// leading id when a mega-group was split (see [`build_tiles`]).
1173    pub(crate) sections: [TileList; NUM_PERMS],
1174    /// Which of those six slots this file actually carries. Sections outside
1175    /// the set are empty and must never be consulted — an empty section is
1176    /// indistinguishable from an empty *graph*, which is why the planner asks
1177    /// this rather than the section lengths.
1178    perms: PermSet,
1179    /// Faults in remote tiles on first scan (`None` for local indexes).
1180    loader: Option<TileLoader>,
1181    /// Optional batched fetch for multi-tile scans (`None` falls back to
1182    /// one [`TileLoader`] call per tile).
1183    bulk: Option<TileBulkLoader>,
1184    /// Set when the loader failed for some tile: results may be incomplete and
1185    /// the caller must surface an error rather than the partial answer.
1186    load_failed: std::sync::atomic::AtomicBool,
1187    /// The reader's concurrent-range fan-out (1 = strictly sequential) — see
1188    /// [`set_read_concurrency`](Self::set_read_concurrency).
1189    read_concurrency: usize,
1190    /// Decompressed tile bodies live here, keyed by `(section, tile)`, faulted
1191    /// in on first scan and evictable. Phase 0 caps it at unlimited, so a
1192    /// scanned permutation stays resident exactly as it did in the old per-tile
1193    /// `OnceLock`s.
1194    cache: Arc<ChunkCache>,
1195    /// The number of triples this index holds, when known without reading a
1196    /// tile (0 = unknown). A named graph's view takes it from the quad
1197    /// directory, so the planner sizes patterns inside `GRAPH <g>` against the
1198    /// graph rather than the whole file — see [`Self::size_hint`].
1199    size_hint: std::sync::atomic::AtomicU64,
1200}
1201
1202impl GraphIndex {
1203    fn from_sections(mut sections: [Vec<Tile>; NUM_PERMS], perms: PermSet) -> Self {
1204        // Move every locally-built/decoded tile body into the shared cache and
1205        // clear its staging slot, so the descriptors hold routing metadata only.
1206        let cache = ChunkCache::unlimited_arc();
1207        for (si, tiles) in sections.iter_mut().enumerate() {
1208            for (ti, tile) in tiles.iter_mut().enumerate() {
1209                if let Some(body) = tile.staged.take() {
1210                    cache.insert((si as u8, ti as u32), Arc::from(body));
1211                }
1212            }
1213        }
1214        GraphIndex {
1215            sections: sections.map(TileList::Full),
1216            perms,
1217            loader: None,
1218            bulk: None,
1219            load_failed: std::sync::atomic::AtomicBool::new(false),
1220            read_concurrency: 1,
1221            cache,
1222            size_hint: std::sync::atomic::AtomicU64::new(0),
1223        }
1224    }
1225
1226    /// Rebuild from tiled sections: per permutation, `(min_a, max_a, block
1227    /// bytes)` per tile in ascending leading-id order — the v0.2 layout.
1228    /// Slots outside `perms` must be empty.
1229    pub fn from_tiles(sections: [Vec<(u32, u32, Vec<u8>)>; NUM_PERMS], perms: PermSet) -> Self {
1230        let sections = sections.map(|tiles| {
1231            tiles
1232                .into_iter()
1233                .map(|(min_a, max_a, bytes)| Tile::local(min_a, max_a, bytes))
1234                .collect()
1235        });
1236        Self::from_sections(sections, perms)
1237    }
1238
1239    /// The permutations this index carries.
1240    pub fn perms(&self) -> PermSet {
1241        self.perms
1242    }
1243
1244    /// Set the byte cap of this index's tile cache (bounded-export phase 2).
1245    /// `u64::MAX` (the open-time default) is unlimited — a scanned permutation
1246    /// stays resident exactly as before. A finite cap evicts least-recently-used
1247    /// tiles, so a full scan's tile residency is bounded.
1248    ///
1249    /// Only safe on an index whose tiles can be **re-faulted** (a remote/ranged
1250    /// index with a loader). A locally-built index has no loader — an evicted
1251    /// tile could never be reloaded — so the caller must not cap those; the
1252    /// bounded-export path caps only the ranged default-graph index and leaves
1253    /// resident named-graph indexes unlimited (released per slot instead).
1254    pub fn set_cache_cap(&self, cap: u64) {
1255        self.cache.set_cap(cap);
1256    }
1257
1258    /// The current tile-cache byte cap (`u64::MAX` = unlimited).
1259    pub fn cache_cap(&self) -> u64 {
1260        self.cache.cap()
1261    }
1262
1263    /// Does this index fault its tiles through a loader (a remote/ranged open)?
1264    /// Only such an index is safe to give a finite tile-cache cap: a locally
1265    /// decoded index has no loader, so an evicted tile could never be reloaded.
1266    pub fn is_lazy(&self) -> bool {
1267        self.loader.is_some()
1268    }
1269
1270    /// A snapshot of the tile cache's observability counters.
1271    pub fn cache_stats(&self) -> crate::chunk_cache::CacheStats {
1272        self.cache.stats()
1273    }
1274
1275    /// Heap bytes this index holds for as long as it is open, outside its tile
1276    /// cache: one descriptor per tile of every permutation the file carries
1277    /// (plus a tile's group directory, if a bound-leading lookup built one),
1278    /// and — on a ranged open — the two copies of the per-tile byte ranges the
1279    /// loaders capture. Grows with the tile count, never with what a scan
1280    /// touches.
1281    pub fn directory_bytes(&self) -> u64 {
1282        let tiles: u64 = self.sections.iter().map(|s| s.len() as u64).sum();
1283        let caps: u64 = self.sections.iter().map(TileList::directory_bytes).sum();
1284        let ranges = if self.loader.is_some() {
1285            2 * 24 * tiles
1286        } else {
1287            0
1288        };
1289        caps + ranges
1290    }
1291
1292    /// A **remote** index: only the tile directories (leading-id ranges per
1293    /// permutation, ascending) are known; tile payloads fault in through
1294    /// `loader` on first scan. Check [`load_incomplete`](Self::load_incomplete)
1295    /// after evaluating — a failed fetch must become an error, never a
1296    /// silently smaller result.
1297    #[allow(clippy::type_complexity)]
1298    pub fn from_remote_directories(
1299        directories: [Vec<(u32, u32, Option<(u32, u32, u32, u32)>)>; NUM_PERMS],
1300        perms: PermSet,
1301        loader: TileLoader,
1302    ) -> Self {
1303        let sections = directories.map(|dir| {
1304            TileList::Full(
1305                dir.into_iter()
1306                    .map(|(min_a, max_a, syn)| Tile::remote(min_a, max_a, syn))
1307                    .collect(),
1308            )
1309        });
1310        GraphIndex {
1311            sections,
1312            perms,
1313            loader: Some(loader),
1314            bulk: None,
1315            load_failed: std::sync::atomic::AtomicBool::new(false),
1316            read_concurrency: 1,
1317            cache: ChunkCache::unlimited_arc(),
1318            size_hint: std::sync::atomic::AtomicU64::new(0),
1319        }
1320    }
1321
1322    /// Attach a batched tile fetcher (see [`TileBulkLoader`]): multi-tile
1323    /// scans prefetch their span through it instead of faulting tile by tile.
1324    pub fn with_bulk_loader(mut self, bulk: TileBulkLoader) -> Self {
1325        self.bulk = Some(bulk);
1326        self
1327    }
1328
1329    /// Record each remote tile's encoded (on-disk) byte length, per section —
1330    /// known to the ranged opener from the tile directory. Powers the join
1331    /// planner's fatness gates; never triggers a fetch.
1332    pub(crate) fn set_tile_lens(&mut self, lens: [Vec<u32>; NUM_PERMS]) {
1333        for (section, ls) in self.sections.iter_mut().zip(lens) {
1334            if let TileList::Full(section) = section {
1335                for (tile, l) in section.iter_mut().zip(ls) {
1336                    tile.len = l;
1337                }
1338            }
1339        }
1340    }
1341
1342    /// Record the reader's concurrent-range fan-out (see
1343    /// [`RangeReader::concurrency`](crate::reader::RangeReader::concurrency)) —
1344    /// the join planner widens its remote probe budget when round trips
1345    /// overlap instead of serializing.
1346    pub(crate) fn set_read_concurrency(&mut self, c: usize) {
1347        self.read_concurrency = c.max(1);
1348    }
1349
1350    /// Record how many triples this index holds (see [`Self::size_hint`]).
1351    pub(crate) fn set_size_hint(&self, triples: u64) {
1352        self.size_hint
1353            .store(triples, std::sync::atomic::Ordering::Relaxed);
1354    }
1355
1356    /// The triples this index holds, when it was told without a tile read
1357    /// (a quad-index view's graph count, a resident per-graph index).
1358    pub(crate) fn size_hint(&self) -> Option<u64> {
1359        match self.size_hint.load(std::sync::atomic::Ordering::Relaxed) {
1360            0 => None,
1361            n => Some(n),
1362        }
1363    }
1364
1365    /// The reader's concurrent-range fan-out (1 = strictly sequential).
1366    pub(crate) fn read_concurrency(&self) -> usize {
1367        self.read_concurrency
1368    }
1369
1370    /// Did any tile fetch fail since this index was opened — or since the last
1371    /// [`reset_load_failure`](Self::reset_load_failure)?
1372    pub fn load_incomplete(&self) -> bool {
1373        self.load_failed.load(std::sync::atomic::Ordering::Relaxed)
1374            || self.sections.iter().any(TileList::load_failed)
1375    }
1376
1377    /// Forget recorded fetch failures — the start-of-evaluation reset for a
1378    /// RESIDENT session, making the incompleteness verdict per-query instead of
1379    /// per-open (one transient network blip used to fail every later query on
1380    /// the session). Safe because failed tiles are never cached: the next scan
1381    /// simply retries them.
1382    pub fn reset_load_failure(&self) {
1383        self.load_failed
1384            .store(false, std::sync::atomic::Ordering::Relaxed);
1385        for s in &self.sections {
1386            s.reset_load_failure();
1387        }
1388    }
1389
1390    /// True for a remote/lazy index (tiles fault in over a `RangeReader`). The
1391    /// join planner uses this to pick read-aware strategies: here a per-row index
1392    /// probe is a network round-trip, not a memory lookup, so a left-deep scan +
1393    /// hash join often beats probing a moderately-sized prefix.
1394    pub fn is_remote(&self) -> bool {
1395        self.loader.is_some()
1396    }
1397
1398    /// Batch-fault the unloaded tiles in `[start, end)` of `section` through
1399    /// the bulk loader, if one is attached and at least two tiles are missing
1400    /// (a single missing tile costs the same either way). A failed batch is
1401    /// not an error here: the tiles stay unloaded and the per-tile loader
1402    /// retries each one (recording failures) when the scan reaches it.
1403    /// Fault, in one batch per permutation, the tile-directory leaves the
1404    /// bound leading ids of `patterns` route to (format `0x06`; a no-op on a
1405    /// resident directory).
1406    pub(crate) fn prefault_leaves(&self, patterns: &[Pattern]) {
1407        let mut want: [std::collections::BTreeSet<usize>; NUM_PERMS] = Default::default();
1408        for &pat in patterns {
1409            let perm = Self::best_permutation_in(self.perms, pat);
1410            let [pa, _, _] = perm.order_pattern(pat);
1411            let si = perm.section_index();
1412            if let (TileList::Leafed(l), Some(a)) = (&self.sections[si], pa) {
1413                // Leaves ascend by leading id: the run whose range holds `a`.
1414                let from = l.leaves.partition_point(|lf| lf.max_a < a);
1415                for (li, lf) in l.leaves.iter().enumerate().skip(from) {
1416                    if lf.min_a > a {
1417                        break;
1418                    }
1419                    want[si].insert(li);
1420                }
1421            }
1422        }
1423        for (si, lis) in want.iter().enumerate() {
1424            if let TileList::Leafed(l) = &self.sections[si] {
1425                if !lis.is_empty() {
1426                    l.ensure_leaves(&lis.iter().copied().collect::<Vec<_>>());
1427                }
1428            }
1429        }
1430    }
1431
1432    /// The absolute stored range of tile `(section, tile)` on a leafed
1433    /// (`0x06`) index, faulting its directory leaf.
1434    pub fn tile_range(&self, section: usize, tile: usize) -> Option<(u64, u64)> {
1435        self.sections.get(section)?.range(tile)
1436    }
1437
1438    /// A remote `0x06` index: per stored permutation a two-level tile
1439    /// directory, tiles faulted through `loader`/`bulk`.
1440    pub fn from_leafed(
1441        lists: [Option<Arc<LeafedTiles>>; NUM_PERMS],
1442        perms: PermSet,
1443        loader: TileLoader,
1444    ) -> Self {
1445        GraphIndex {
1446            sections: lists.map(|l| match l {
1447                Some(l) => TileList::Leafed(l),
1448                None => TileList::Full(Vec::new()),
1449            }),
1450            perms,
1451            loader: Some(loader),
1452            bulk: None,
1453            load_failed: std::sync::atomic::AtomicBool::new(false),
1454            read_concurrency: 1,
1455            cache: ChunkCache::unlimited_arc(),
1456            size_hint: std::sync::atomic::AtomicU64::new(0),
1457        }
1458    }
1459
1460    fn prefetch_span(&self, section: usize, start: usize, end: usize) {
1461        let missing: Vec<usize> = (start..end)
1462            .filter(|&ti| !self.tile_resident(section, ti))
1463            .collect();
1464        self.bulk_fault(section, &missing);
1465    }
1466
1467    /// Is tile `(section, tile)`'s body currently resident in the cache?
1468    fn tile_resident(&self, section: usize, tile: usize) -> bool {
1469        self.cache.contains((section as u8, tile as u32))
1470    }
1471
1472    /// Bulk-fault a set of (possibly scattered, ascending) missing tile indices in
1473    /// `section` through the bulk loader in one coalesced read, storing each image.
1474    /// No-op without a bulk loader or for fewer than two tiles (a single fault
1475    /// costs the same via the per-tile loader). A failed batch leaves the tiles
1476    /// unloaded for the per-tile loader to retry. Shared by the consecutive-span
1477    /// scan prefetch and the scattered batch-probe prefetch.
1478    fn bulk_fault(&self, section: usize, tiles: &[usize]) {
1479        if tiles.len() < 2 {
1480            return;
1481        }
1482        let Some(bulk) = &self.bulk else { return };
1483        if let Some(images) = bulk(section, tiles) {
1484            if images.len() == tiles.len() {
1485                for (&ti, img) in tiles.iter().zip(images) {
1486                    self.cache
1487                        .insert((section as u8, ti as u32), Arc::from(img));
1488                }
1489            }
1490        }
1491    }
1492
1493    /// Batch-fault the tiles a set of upcoming **probe** patterns will route to,
1494    /// before they are probed one at a time. Each probe (a bound leading id)
1495    /// routes to a single tile; gathered across the batch those tiles are
1496    /// scattered, so faulting them together turns N sequential remote round trips
1497    /// into a few coalesced parallel reads — the read-amplification win for a
1498    /// label-heavy join over a lazy reader. Honors the synopsis prune (never
1499    /// fetches a tile a bound secondary proves can't match). No-op for a local
1500    /// index or when fewer than two tiles per section need faulting.
1501    pub(crate) fn prefetch_probe_tiles(&self, patterns: &[Pattern]) {
1502        if self.bulk.is_none() {
1503            return;
1504        }
1505        let mut want: [std::collections::BTreeSet<usize>; NUM_PERMS] = Default::default();
1506        self.prefault_leaves(patterns);
1507        for &pat in patterns {
1508            let perm = Self::best_permutation(pat);
1509            let [pa, pb, pc] = perm.order_pattern(pat);
1510            let si = perm.section_index();
1511            let (start, end) = self.tile_span(si, pa);
1512            for ti in start..end {
1513                if self.sections[si][ti].syn_admits(pb, pc) && !self.tile_resident(si, ti) {
1514                    want[si].insert(ti);
1515                }
1516            }
1517        }
1518        for (si, set) in want.iter().enumerate() {
1519            let tiles: Vec<usize> = set.iter().copied().collect();
1520            self.bulk_fault(si, &tiles);
1521        }
1522    }
1523
1524    /// The tile's cache entry, faulting its block image in through the loader if
1525    /// remote. Returns an owned [`CacheEntry`] handle (an `Arc`) the caller
1526    /// holds for the duration of one parse/scan — valid even if a later eviction
1527    /// drops the map slot (the handout invariant).
1528    ///
1529    /// A FAILED fetch records the failure and caches nothing (returns `None`),
1530    /// so a later evaluation retries the tile — a transient network error must
1531    /// not permanently poison a long-lived (resident) session. A local tile
1532    /// with no body returns an empty entry (constructed empty on purpose).
1533    fn tile_entry(&self, section: usize, tile: usize) -> Option<Arc<CacheEntry>> {
1534        // A query run under a memory ceiling ([`crate::guard`]) stops here.
1535        crate::guard::check();
1536        let key = (section as u8, tile as u32);
1537        if let Some(e) = self.cache.get(key) {
1538            return Some(e);
1539        }
1540        match &self.loader {
1541            Some(load) => match load(section, tile) {
1542                Some(bytes) => Some(self.cache.insert(key, Arc::from(bytes))),
1543                None => {
1544                    self.load_failed
1545                        .store(true, std::sync::atomic::Ordering::Relaxed);
1546                    None
1547                }
1548            },
1549            None => Some(self.cache.insert(key, Arc::from(Vec::new()))),
1550        }
1551    }
1552
1553    /// Tile `(section, tile)`'s a-group directory, held in its cache entry and
1554    /// built from `block` (that entry's parsed body) on first use. Its bytes
1555    /// are charged to the tile cache when built, so a capped cache bounds the
1556    /// directories too, and they are evicted with their tile.
1557    fn tile_dir<'e>(
1558        &self,
1559        section: usize,
1560        tile: usize,
1561        entry: &'e Arc<CacheEntry>,
1562        block: &TripleBlock<'_>,
1563    ) -> &'e GroupDirectory {
1564        let (dir, built) = entry.group_dir(|| block.group_directory());
1565        if built {
1566            self.cache.charge_extra(
1567                (section as u8, tile as u32),
1568                entry,
1569                dir.heap_bytes() + std::mem::size_of::<GroupDirectory>() as u64,
1570            );
1571        }
1572        dir
1573    }
1574
1575    /// The tile's decompressed block image as a shared handle, for the file
1576    /// writer (local indexes, all tiles resident). Empty on a miss.
1577    pub(crate) fn tile_body(&self, section: usize, tile: usize) -> Arc<[u8]> {
1578        self.tile_entry(section, tile)
1579            .map(|e| e.body_arc())
1580            .unwrap_or_else(|| Arc::from(Vec::new()))
1581    }
1582
1583    /// Total triple count (sum of the SPO tiles' zone counts). For a remote
1584    /// index this faults in the SPO tiles — prefer the header's quad count.
1585    pub fn triple_count(&self) -> u32 {
1586        self.prefetch_span(0, 0, self.sections[0].len());
1587        (0..self.sections[0].len())
1588            .filter_map(|ti| {
1589                let entry = self.tile_entry(0, ti)?;
1590                TripleBlock::parse(entry.body())
1591                    .ok()
1592                    .map(|b| b.zone().count)
1593            })
1594            // A hostile tile can claim any count: saturate, never overflow.
1595            .fold(0u32, u32::saturating_add)
1596    }
1597
1598    /// Stream **every** triple of one stored permutation in that section's own
1599    /// order, as `(a, b, c)` in the permutation's column order (SPO gives
1600    /// `(s, p, o)`, OSP gives `(o, s, p)`, …).
1601    ///
1602    /// The access a whole-graph pass needs — `rete recard`'s streaming card
1603    /// derivation reads SPO for out-degrees and OSP for in-degrees and literal
1604    /// objects — and one [`scan_iter`](Self::scan_iter) cannot give, because a
1605    /// fully unbound pattern always routes to SPO. Tiles fault with the same
1606    /// geometric prefetch as a scan and are decoded one at a time; nothing is
1607    /// retained beyond the tile cache, so memory follows its cap, not the
1608    /// section. A failed fault is recorded in
1609    /// [`load_incomplete`](Self::load_incomplete) and the tile is skipped, as
1610    /// everywhere else. Returns `false` when the file does not store `perm`.
1611    pub fn for_each_in_order(
1612        &self,
1613        perm: IndexPermutation,
1614        f: &mut dyn FnMut(u32, u32, u32),
1615    ) -> bool {
1616        if !self.perms.contains(perm) {
1617            return false;
1618        }
1619        let si = perm.section_index();
1620        let end = self.sections[si].len();
1621        let mut window = PREFETCH_WINDOW_START;
1622        for ti in 0..end {
1623            if !self.tile_resident(si, ti) {
1624                self.prefetch_span(si, ti, (ti + window).min(end));
1625                window = window.saturating_mul(2).min(PREFETCH_WINDOW_MAX);
1626            }
1627            if let Some(entry) = self.tile_entry(si, ti) {
1628                if let Ok(block) = TripleBlock::parse(entry.body()) {
1629                    for (a, b, c) in block.scan(None, None, None) {
1630                        f(a, b, c);
1631                    }
1632                }
1633            }
1634        }
1635        true
1636    }
1637
1638    /// The tiles of each permutation section (in `ALL_PERMS` order), for the
1639    /// file writer.
1640    pub fn tile_sections(&self) -> [&TileList; NUM_PERMS] {
1641        [
1642            &self.sections[0],
1643            &self.sections[1],
1644            &self.sections[2],
1645            &self.sections[3],
1646            &self.sections[4],
1647            &self.sections[5],
1648        ]
1649    }
1650
1651    /// The permutation selected for a pattern: the one with the longest bound
1652    /// prefix. Ties keep the canonical SPO order (then POS, then OSP), which
1653    /// makes provenance stable for unbound or equally selective shapes and routes
1654    /// a fully unbound pattern to the SPO block rather than fetching all three.
1655    ///
1656    /// Six-permutation shorthand for [`best_permutation_in`]. Its answer is
1657    /// always inside [`PermSet::CORE`] (see `perm_routing_never_leaves_core`),
1658    /// so a file with any legal [`PermSet`] routes identically — but call the
1659    /// `_in` form wherever the file's set is at hand, so that stays true by
1660    /// construction rather than by argument.
1661    ///
1662    /// [`best_permutation_in`]: GraphIndex::best_permutation_in
1663    pub fn best_permutation(pattern: Pattern) -> IndexPermutation {
1664        Self::best_permutation_in(PermSet::ALL, pattern)
1665    }
1666
1667    /// [`best_permutation`](GraphIndex::best_permutation) restricted to the
1668    /// permutations a file carries.
1669    pub fn best_permutation_in(perms: PermSet, pattern: Pattern) -> IndexPermutation {
1670        let mut best = IndexPermutation::Spo;
1671        let mut best_score = best.leading_bound(pattern);
1672        for perm in perms.iter() {
1673            let score = perm.leading_bound(pattern);
1674            if score > best_score {
1675                best = perm;
1676                best_score = score;
1677            }
1678        }
1679        best
1680    }
1681
1682    /// Choose a permutation that **routes** on `pattern`'s bound prefix *and*
1683    /// streams sorted on the canonical column `sort_col` (0=subject, 1=predicate,
1684    /// 2=object) — i.e. `sort_col` is the leading *free* component after the bound
1685    /// prefix. Returns `None` if `sort_col` is itself bound (then every row shares
1686    /// that value — any permutation is "sorted" on it) or no permutation qualifies.
1687    /// This is the precondition a merge join needs: both inputs sorted on the join
1688    /// key. Among qualifiers, the longest bound prefix (best routing) wins.
1689    ///
1690    /// Six-permutation shorthand for [`permutation_sorted_on_in`].
1691    ///
1692    /// [`permutation_sorted_on_in`]: GraphIndex::permutation_sorted_on_in
1693    pub fn permutation_sorted_on(pattern: Pattern, sort_col: usize) -> Option<IndexPermutation> {
1694        Self::permutation_sorted_on_in(PermSet::ALL, pattern, sort_col)
1695    }
1696
1697    /// [`permutation_sorted_on`](GraphIndex::permutation_sorted_on) restricted
1698    /// to the permutations a file carries — **and it will not trade routing for
1699    /// sort order**.
1700    ///
1701    /// A qualifier must match the bound prefix that
1702    /// [`best_permutation_in`](GraphIndex::best_permutation_in) achieves.
1703    /// With all six that costs nothing: every ordering of `(s, p, o)` exists, so
1704    /// "the bound components first, then `sort_col`" is always available at the
1705    /// maximal prefix. With [`PermSet::CORE`] it is what stops a merge join from
1706    /// buying its co-sorted stream with a **whole-section scan** — e.g.
1707    /// `?s :p ?o` sorted on `?s` has no `P*S` order left, and the only remaining
1708    /// sorted-on-subject stream is an unrouted SPO scan of the entire graph. The
1709    /// merge seed then declines and the hash/probe path answers, which is
1710    /// strictly the cheaper plan.
1711    pub fn permutation_sorted_on_in(
1712        perms: PermSet,
1713        pattern: Pattern,
1714        sort_col: usize,
1715    ) -> Option<IndexPermutation> {
1716        let bound = [
1717            pattern.0.is_some(),
1718            pattern.1.is_some(),
1719            pattern.2.is_some(),
1720        ];
1721        if bound[sort_col] {
1722            return None;
1723        }
1724        let routed = Self::best_permutation_in(perms, pattern).leading_bound(pattern);
1725        let mut best: Option<(IndexPermutation, usize)> = None;
1726        for perm in perms.iter() {
1727            let roles = perm.roles();
1728            let lead = perm.leading_bound(pattern);
1729            // The component at slot `lead` is the first free one; it must be the
1730            // sort column (so the stream is sorted on it after the bound prefix).
1731            if lead < 3 && roles[lead] == sort_col && best.map(|(_, s)| lead > s).unwrap_or(true) {
1732                best = Some((perm, lead));
1733            }
1734        }
1735        best.filter(|&(_, lead)| lead >= routed).map(|(p, _)| p)
1736    }
1737
1738    /// Match one already-decoded serialized permutation block (a single v0.1
1739    /// section or one v0.2 tile). This is the core primitive for range-routed
1740    /// readers: the caller fetches only the selected payload, then this scans
1741    /// it as if it came from a full [`GraphIndex`].
1742    pub fn match_serialized_block(
1743        bytes: &[u8],
1744        permutation: IndexPermutation,
1745        pattern: Pattern,
1746    ) -> Vec<Triple> {
1747        let [pa, pb, pc] = permutation.order_pattern(pattern);
1748        let mut out: Vec<Triple> = TripleBlock::parse(bytes)
1749            .ok()
1750            .filter(|b| b.zone().may_contain(pa, pb, pc))
1751            .map(|b| b.scan(pa, pb, pc))
1752            .into_iter()
1753            .flatten()
1754            .map(move |abc| permutation.back(abc))
1755            .collect();
1756        out.sort_unstable();
1757        out
1758    }
1759
1760    /// All triples matching `pattern`, returned in canonical `(s, p, o)` order.
1761    ///
1762    /// Thin eager wrapper over [`scan_iter`](Self::scan_iter): collect the lazy
1763    /// stream and restore the canonical sort (the stream is sorted in the chosen
1764    /// permutation's order, which differs from canonical once `perm.back`
1765    /// permutes the free components).
1766    pub fn match_pattern(&self, pattern: Pattern) -> Vec<Triple> {
1767        let mut out: Vec<Triple> = self.scan_iter(pattern).collect();
1768        out.sort_unstable();
1769        out
1770    }
1771
1772    /// Lazily stream the triples matching `pattern` in canonical `(s, p, o)`
1773    /// order *within the chosen permutation* — the streaming entry point for
1774    /// callers that can stop early (ASK, `LIMIT`, BGP probes) or that don't need
1775    /// the canonical re-sort. Decodes only the matching groups (see
1776    /// [`TripleBlock::scan`]); a malformed/absent block yields nothing rather
1777    /// than panicking. The permutation is chosen for the longest bound prefix.
1778    pub fn scan_iter(&self, pattern: Pattern) -> impl Iterator<Item = Triple> + '_ {
1779        self.scan_iter_with(pattern, Self::best_permutation(pattern))
1780    }
1781
1782    /// What [`scan_iter`](Self::scan_iter) *would* fetch for `pattern`, computed
1783    /// from the tile directories alone — **no tile is fetched, decoded, or even
1784    /// looked at**. This is the cost-preview half of the routed scan: it walks
1785    /// the same three decisions the scan makes (permutation, routed tile span,
1786    /// synopsis prune) and reports what survives, so a caller can say "this dump
1787    /// will pull 16 MB of index, not 376 MB" before starting it.
1788    ///
1789    /// It is a preview, not a promise: `tile_bytes` is an **upper bound** on the
1790    /// index bytes the scan reads (an admitted tile can still be rejected by its
1791    /// in-tile zone map once in hand, and a re-scan of an already-faulted tile
1792    /// reads nothing), and it says nothing about the dictionary, which is where
1793    /// a term-resolving dump's real cost usually lives.
1794    pub fn scan_plan(&self, pattern: Pattern) -> ScanPlan {
1795        let perm = Self::best_permutation_in(self.perms, pattern);
1796        let [_, pb, pc] = perm.order_pattern(pattern);
1797        let si = perm.section_index();
1798        let tiles = &self.sections[si];
1799        let (start, end) = self.tile_span(si, perm.order_pattern(pattern)[0]);
1800        let mut admitted = 0usize;
1801        let mut tile_bytes = 0u64;
1802        for ti in start..end {
1803            let t = &tiles[ti];
1804            if t.syn_admits(pb, pc) {
1805                admitted += 1;
1806                tile_bytes += t.encoded_len();
1807            }
1808        }
1809        ScanPlan {
1810            permutation: perm,
1811            tiles_total: tiles.len(),
1812            tiles_routed: end - start,
1813            tiles_admitted: admitted,
1814            tile_bytes,
1815            section_bytes: tiles.total_bytes(),
1816        }
1817    }
1818
1819    /// One bounded, **resumable** slice of `pattern`'s matches: the pull form of
1820    /// [`scan_iter`](Self::scan_iter), for a caller that cannot hold a borrow
1821    /// across calls.
1822    ///
1823    /// Returns `(triples, next_cursor, done)`; start at `cursor = 0` and feed the
1824    /// returned cursor back until `done`. The whole resume state is one opaque
1825    /// `u64` — `(tile index, next a-group id)` — so no iterator, no thread and no
1826    /// self-referential struct has to survive between calls. That is the shape a
1827    /// foreign-function boundary needs: a wasm module cannot suspend a Rust
1828    /// iterator mid-scan and hand control back to its host, but it can be handed
1829    /// a `u64` and told to carry on.
1830    ///
1831    /// `max_rows` is a **floor**, not a hard cut: a batch always ends on an
1832    /// a-group boundary of the chosen permutation, so no group is ever split
1833    /// across two calls and the overshoot is one group's fanout. Nothing is ever
1834    /// rescanned ([`TripleBlock::scan_resume`] jumps straight to the group
1835    /// through the tile's directory), so a full drain is O(n) overall — unlike
1836    /// `(offset, limit)`, which is O(n²/limit).
1837    ///
1838    /// Every call either yields at least one row or reports `done`, so a caller
1839    /// never has to guard against an empty non-final batch.
1840    ///
1841    /// **Order.** Rows arrive in the chosen permutation's order and each batch is
1842    /// re-sorted canonically, exactly as [`match_pattern`](Self::match_pattern)
1843    /// does for a whole result. For a fully unbound pattern (routed to SPO) that
1844    /// makes the concatenation of every batch identical to `match_pattern`'s
1845    /// output; for a bound pattern the *set* is identical but the canonical
1846    /// re-sort is per batch rather than global.
1847    ///
1848    /// **Memory** is O(`max_rows` + one a-group + the tiles faulted so far) —
1849    /// never O(matches). That is the whole point: an unbounded `(?s ?p ?o)` over
1850    /// a 26-million-quad graph used to materialize every match before the caller
1851    /// saw one row.
1852    pub fn scan_batch(
1853        &self,
1854        pattern: Pattern,
1855        cursor: u64,
1856        max_rows: usize,
1857    ) -> (Vec<Triple>, u64, bool) {
1858        let max_rows = max_rows.max(1);
1859        let perm = Self::best_permutation(pattern);
1860        let [pa, pb, pc] = perm.order_pattern(pattern);
1861        let si = perm.section_index();
1862        // Same routing as `scan_iter_with`: the tile directory alone decides the
1863        // span, so this is free on a lazy/remote open.
1864        let (start, end) = self.tile_span(si, pa);
1865        let mut ti = ((cursor >> 32) as usize).max(start);
1866        let mut from_a = cursor as u32;
1867        let mut out: Vec<Triple> = Vec::new();
1868        // Same geometric prefetch ramp as the streaming scan: a batch that stops
1869        // after one tile never fetches the ones behind it.
1870        let mut window = PREFETCH_WINDOW_START;
1871        while ti < end {
1872            // Synopsis pre-fault prune — no tile is fetched to reject it.
1873            if !self.sections[si][ti].syn_admits(pb, pc) {
1874                ti += 1;
1875                from_a = 0;
1876                continue;
1877            }
1878            if !self.tile_resident(si, ti) {
1879                self.prefetch_span(si, ti, (ti + window).min(end));
1880                window = window.saturating_mul(2).min(PREFETCH_WINDOW_MAX);
1881            }
1882            let entry = self.tile_entry(si, ti);
1883            let mut resume_at: Option<u32> = None;
1884            if let Some(block) = entry
1885                .as_deref()
1886                .and_then(|e| TripleBlock::parse(e.body()).ok())
1887                .filter(|b| b.zone().may_contain(pa, pb, pc))
1888            {
1889                let rows = match pa {
1890                    // A bound leading id is one a-group; it cannot be split
1891                    // inside a tile, only across tiles (a mega-group split), so
1892                    // there is nothing finer to resume at than the tile.
1893                    Some(a) => {
1894                        let dir = self.tile_dir(si, ti, entry.as_ref().unwrap(), &block);
1895                        block.scan_from(dir, a, pb, pc)
1896                    }
1897                    // Fresh tile: no directory needed, so a straight drain never
1898                    // pays for one.
1899                    None if from_a == 0 => block.scan(None, pb, pc),
1900                    None => {
1901                        let dir = self.tile_dir(si, ti, entry.as_ref().unwrap(), &block);
1902                        block.scan_resume(dir, from_a, pb, pc)
1903                    }
1904                };
1905                // Only a leading-unbound scan can be resumed mid-tile: with `pa`
1906                // bound there is one a-group and `scan_from` always restarts it
1907                // from the top, so cutting inside it would duplicate rows.
1908                let splittable = pa.is_none();
1909                let mut last_a: Option<u32> = None;
1910                for abc in rows {
1911                    // Full, and this row opens a new a-group ⇒ cut here and
1912                    // resume at this group. The row is dropped, not lost: the
1913                    // next call re-enters at exactly this id.
1914                    if splittable && out.len() >= max_rows && last_a != Some(abc.0) {
1915                        resume_at = Some(abc.0);
1916                        break;
1917                    }
1918                    last_a = Some(abc.0);
1919                    out.push(perm.back(abc));
1920                }
1921            }
1922            if let Some(a) = resume_at {
1923                out.sort_unstable();
1924                return (out, ((ti as u64) << 32) | a as u64, false);
1925            }
1926            ti += 1;
1927            from_a = 0;
1928            if out.len() >= max_rows {
1929                break;
1930            }
1931        }
1932        let done = ti >= end;
1933        out.sort_unstable();
1934        (out, (ti as u64) << 32, done)
1935    }
1936
1937    /// Stream `pattern`'s matches **sorted on the canonical column `sort_col`**
1938    /// (0=subject, 1=predicate, 2=object): the stream's `sort_col` values are
1939    /// ascending. Chooses a permutation that routes on the bound prefix *and*
1940    /// orders by `sort_col` ([`permutation_sorted_on`](Self::permutation_sorted_on));
1941    /// `None` when none qualifies (e.g. `sort_col` is itself bound). The
1942    /// precondition for feeding a [merge join](crate::bgp): both inputs sorted on
1943    /// the shared join key.
1944    pub(crate) fn scan_iter_sorted_on(
1945        &self,
1946        pattern: Pattern,
1947        sort_col: usize,
1948    ) -> Option<impl Iterator<Item = Triple> + '_> {
1949        Some(self.scan_iter_with(
1950            pattern,
1951            Self::permutation_sorted_on_in(self.perms, pattern, sort_col)?,
1952        ))
1953    }
1954
1955    /// Stream `pattern`'s matches using a **specific** permutation `perm`; the
1956    /// stream is sorted in `perm`'s `(a, b, c)` order. Shared core of `scan_iter`
1957    /// and `scan_iter_sorted_on`.
1958    fn scan_iter_with(
1959        &self,
1960        pattern: Pattern,
1961        perm: IndexPermutation,
1962    ) -> impl Iterator<Item = Triple> + '_ {
1963        // Route: a bound leading id binary-searches the tile directory to exactly
1964        // one tile (groups are never split across tiles); an unbound one chains
1965        // every tile's cursor. Within a tile, a bound leading scan jumps to its
1966        // a-group via the tile's lazily-built group directory — built on first
1967        // use, costing one walk of that (budget-sized) tile.
1968        let [pa, pb, pc] = perm.order_pattern(pattern);
1969        let si = perm.section_index();
1970        let (start, end) = self.tile_span(si, pa);
1971        // Coalesce tile faults, but ramp the prefetch window geometrically as
1972        // the scan advances rather than fetching the whole span up front: a
1973        // small `LIMIT` (which stops pulling early) then never faults tiles
1974        // past the rows it needs, while a full scan still batches into a
1975        // handful of coalesced reads (4, 8, 16, … tiles). A bound leading scan
1976        // spans a single tile, so the prefetch no-ops and it faults just that
1977        // one tile, unchanged.
1978        let window = std::cell::Cell::new(PREFETCH_WINDOW_START);
1979        (start..end)
1980            // Synopsis pre-fault prune: drop a routed tile the directory proves
1981            // can't match a bound secondary component, **without fetching it**
1982            // (the remote win — a negative/sparse lookup costs zero tile reads).
1983            // A bound leading id routes to a single tile, so this is where it
1984            // bites; a fully-unbound leading scan leaves `pb`/`pc` unbound, so
1985            // `syn_admits` keeps every tile, unchanged.
1986            .filter(move |&ti| self.sections[si][ti].syn_admits(pb, pc))
1987            .flat_map(move |ti| {
1988                // Fault in (if remote), parse (untrusted bytes ⇒ `None` on
1989                // malformed), and zone-prune per tile, then stream the
1990                // matching groups.
1991                if !self.tile_resident(si, ti) {
1992                    let w = window.get();
1993                    self.prefetch_span(si, ti, (ti + w).min(end));
1994                    window.set(w.saturating_mul(2).min(PREFETCH_WINDOW_MAX));
1995                }
1996                // A failed fetch is recorded by the index and scans as empty.
1997                let rows: Vec<Triple> = self.tile_rows(si, ti, pa, pb, pc).unwrap_or_default();
1998                rows.into_iter()
1999            })
2000            .map(move |abc| perm.back(abc))
2001    }
2002
2003    /// The rows of tile `ti` of section `si` matching `(pa, pb, pc)` (in that
2004    /// section's `a, b, c` order). The tile is decoded once and its matching
2005    /// rows materialized into an owned `Vec`, so the borrowed block/cursor never
2006    /// outlives the tile's cache handle (an `Arc` dropped on return); the buffer
2007    /// is bounded to one tile's rows. `None` = the tile could not be fetched
2008    /// (the failure is recorded, see [`Self::load_incomplete`]); a malformed or
2009    /// zone-pruned tile yields no rows.
2010    fn tile_rows(
2011        &self,
2012        si: usize,
2013        ti: usize,
2014        pa: Option<u32>,
2015        pb: Option<u32>,
2016        pc: Option<u32>,
2017    ) -> Option<Vec<Triple>> {
2018        let entry = self.tile_entry(si, ti)?;
2019        Some(
2020            TripleBlock::parse(entry.body())
2021                .ok()
2022                .filter(|b| b.zone().may_contain(pa, pb, pc))
2023                .map(|b| match pa {
2024                    Some(a) => {
2025                        let dir = self.tile_dir(si, ti, &entry, &b);
2026                        b.scan_from(dir, a, pb, pc).collect()
2027                    }
2028                    None => b.scan(pa, pb, pc).collect(),
2029                })
2030                .unwrap_or_default(),
2031        )
2032    }
2033
2034    /// Is the object of every triple with predicate `p` inside an object-id
2035    /// range `covers` accepts? Read off the predicate-leading tiles (POS, or PSO):
2036    /// a tile whose synopsis object range `covers` accepts is proven without
2037    /// being read — the synopsis bounds every row of the tile, whatever its
2038    /// predicate, so a tile shared with other predicates only makes the test
2039    /// stricter. The tiles left (no synopsis, or a range it does not accept) are
2040    /// read and their rows of `p` checked one by one: on a lazy index only when
2041    /// there are at most `max_read` of them, in one batch, else the answer is
2042    /// `false`; on a resident one always, in memory. `false` also when a tile
2043    /// cannot be read. An index without `p` proves it trivially.
2044    pub(crate) fn predicate_objects_within(
2045        &self,
2046        p: u32,
2047        covers: impl Fn(u32, u32) -> bool,
2048        max_read: usize,
2049    ) -> bool {
2050        // (perm, which synopsis/row column holds the object)
2051        let (perm, object_is_b) = if self.perms.contains(IndexPermutation::Pos) {
2052            (IndexPermutation::Pos, true)
2053        } else if self.perms.contains(IndexPermutation::Pso) {
2054            (IndexPermutation::Pso, false)
2055        } else {
2056            return false;
2057        };
2058        let si = perm.section_index();
2059        let (start, end) = self.tile_span(si, Some(p));
2060        let unproven: Vec<usize> = (start..end)
2061            .filter(|&ti| match self.sections[si][ti].syn {
2062                Some((min_b, max_b, min_c, max_c)) => {
2063                    let (lo, hi) = if object_is_b {
2064                        (min_b, max_b)
2065                    } else {
2066                        (min_c, max_c)
2067                    };
2068                    !covers(lo, hi)
2069                }
2070                None => true,
2071            })
2072            .collect();
2073        if unproven.is_empty() {
2074            return true;
2075        }
2076        if self.is_lazy() {
2077            if unproven.len() > max_read {
2078                return false;
2079            }
2080            let missing: Vec<usize> = unproven
2081                .iter()
2082                .copied()
2083                .filter(|&ti| !self.tile_resident(si, ti))
2084                .collect();
2085            self.bulk_fault(si, &missing);
2086        }
2087        for ti in unproven {
2088            let Some(rows) = self.tile_rows(si, ti, Some(p), None, None) else {
2089                return false;
2090            };
2091            let all = rows.iter().all(|&(_, b, c)| {
2092                let o = if object_is_b { b } else { c };
2093                covers(o, o)
2094            });
2095            if !all {
2096                return false;
2097            }
2098        }
2099        true
2100    }
2101
2102    /// A planner estimate of how many triples match `pattern`, read off the
2103    /// tile directories alone — no tile is fetched or decoded, so it costs the
2104    /// same on a lazy 50 GB file as on a resident one. `total` is the number
2105    /// of triples the index holds (an upper bound is fine).
2106    ///
2107    /// For every carried permutation that leads with a bound component, the
2108    /// routed tiles the synopsis admits are counted; a tile holds about
2109    /// `total / tiles` triples, and the routed run is charged half a tile less
2110    /// than its length, since a key rarely starts and ends on tile boundaries.
2111    /// The smallest figure wins, so `?s a :Rare` (a few tiles of OPS) beats
2112    /// `?s :p ?o` (every tile of `:p` in PSO). Its resolution is one tile:
2113    /// patterns inside one tile tie, which is all a join order needs. A
2114    /// pattern with no bound component estimates `total`; one the synopsis
2115    /// rules out estimates 0.
2116    pub(crate) fn estimate_matches(&self, pattern: Pattern, total: u64) -> f64 {
2117        let mut best = total as f64;
2118        for perm in self.perms.iter() {
2119            let [pa, pb, pc] = perm.order_pattern(pattern);
2120            if pa.is_none() {
2121                continue;
2122            }
2123            let si = perm.section_index();
2124            let tiles = &self.sections[si];
2125            if tiles.is_empty() {
2126                return 0.0;
2127            }
2128            let per_tile = total as f64 / tiles.len() as f64;
2129            // A leafed list whose leaf is not resident must still be estimated
2130            // from the root alone (a cost estimate must not cost a read), and
2131            // must give the figure the same directory gives when laid out flat
2132            // (format 0x05) to within the unavoidable: a predicate that fills
2133            // thousands of tiles is thousands of tiles whichever layout stores
2134            // them. Counting "one tile" here made the planner see a 2 M-row
2135            // predicate as 7 k rows and hash-join it (see [`LeafedTiles::admitted_estimate`]).
2136            let admitted = match (tiles, pa) {
2137                (TileList::Leafed(l), Some(a))
2138                    if !l
2139                        .leaves
2140                        .iter()
2141                        .enumerate()
2142                        .filter(|(_, lf)| lf.min_a <= a && a <= lf.max_a)
2143                        .all(|(li, _)| l.cells[li].get().is_some()) =>
2144                {
2145                    l.admitted_estimate(a, pb, pc)
2146                }
2147                _ => {
2148                    let (start, end) = self.tile_span(si, pa);
2149                    (start..end)
2150                        .filter(|&ti| tiles[ti].syn_admits(pb, pc))
2151                        .count() as f64
2152                }
2153            };
2154            let est = if admitted <= 0.0 {
2155                0.0
2156            } else {
2157                (admitted - 0.5).max(0.5) * per_tile
2158            };
2159            best = best.min(est);
2160        }
2161        best
2162    }
2163
2164    /// The smallest leading id `>= a` present in permutation section `section`,
2165    /// or `None` past the last one. Routes by the tile directory, then reads the
2166    /// tile's own group directory — so it skips the ids a tile's *range* covers
2167    /// but its body does not hold. That matters where a range is conservative:
2168    /// a quad-index view's tile shared with other graphs is `0..=u32::MAX`, and
2169    /// stepping through it id by id would never finish. Faults the tiles it
2170    /// reads; a failed fault ends the search (`load_incomplete` says so).
2171    pub(crate) fn next_leading(&self, section: usize, a: u32) -> Option<u32> {
2172        let tiles = &self.sections[section];
2173        let mut ti = tiles.first_max_ge(a);
2174        while ti < tiles.len() {
2175            let from = a.max(tiles[ti].min_a);
2176            let entry = self.tile_entry(section, ti)?;
2177            if let Ok(block) = TripleBlock::parse(entry.body()) {
2178                let dir = self.tile_dir(section, ti, &entry, &block);
2179                if let Some(x) = dir.first_at_or_after(from) {
2180                    return Some(x);
2181                }
2182            }
2183            ti += 1;
2184        }
2185        None
2186    }
2187
2188    /// The tile index span a scan must visit: every tile when the leading
2189    /// component is unbound, else the run of tiles whose leading-id ranges
2190    /// cover it. Tile ranges are ascending; consecutive tiles may SHARE a
2191    /// leading id when a mega-group was split across tiles (see
2192    /// [`build_tiles`]), so the span is a range scan, not a single hit —
2193    /// files without splits still yield a span of at most one tile.
2194    pub(crate) fn tile_span(&self, section: usize, pa: Option<u32>) -> (usize, usize) {
2195        let tiles = &self.sections[section];
2196        match pa {
2197            None => {
2198                // An unbound scan visits every tile: fault every leaf at once.
2199                tiles.ensure_span(0, tiles.len());
2200                (0, tiles.len())
2201            }
2202            Some(a) => {
2203                let i = tiles.first_max_ge(a);
2204                let j = tiles.first_min_gt(a).max(i);
2205                // The callers walk the run: fault its leaves in one batch, as
2206                // the tile-by-tile walk this replaced did one by one.
2207                tiles.ensure_span(i, j);
2208                (i, j)
2209            }
2210        }
2211    }
2212}
2213
2214#[cfg(test)]
2215mod tests {
2216    use super::*;
2217
2218    fn graph() -> (GraphIndex, Vec<Triple>) {
2219        let data = vec![
2220            (1, 10, 100),
2221            (1, 10, 101),
2222            (1, 11, 100),
2223            (2, 10, 100),
2224            (2, 12, 200),
2225            (3, 11, 300),
2226        ];
2227        let mut b = GraphIndexBuilder::new();
2228        for &t in &data {
2229            b.push(t);
2230        }
2231        (b.build(), data)
2232    }
2233
2234    /// Brute-force reference for a pattern.
2235    fn reference(data: &[Triple], (s, p, o): Pattern) -> Vec<Triple> {
2236        let mut v: Vec<Triple> = data
2237            .iter()
2238            .copied()
2239            .filter(|&(a, b, c)| {
2240                s.is_none_or(|x| x == a) && p.is_none_or(|x| x == b) && o.is_none_or(|x| x == c)
2241            })
2242            .collect();
2243        v.sort_unstable();
2244        v
2245    }
2246
2247    /// The tile-directory estimate orders patterns the way their true sizes
2248    /// do, never fetches anything, and is exact at its extremes: unbound is the
2249    /// total, a key no tile can hold is zero.
2250    #[test]
2251    fn tile_estimates_rank_patterns_by_size() {
2252        // Predicate 1 on every subject, predicate 2 on a few; object 50 is a
2253        // "class" with 4000 instances, object 60 one with 3.
2254        let mut b = GraphIndexBuilder::new().with_tile_budget(256);
2255        let mut n = 0u64;
2256        for s in 0..4000u32 {
2257            b.push((s, 1, 50));
2258            b.push((s, 3, 1000 + s));
2259            n += 2;
2260            if s % 1000 == 0 {
2261                b.push((s, 2, 60));
2262                n += 1;
2263            }
2264        }
2265        let idx = b.build();
2266        let est = |p: Pattern| idx.estimate_matches(p, n);
2267        assert_eq!(est((None, None, None)), n as f64);
2268        assert!(est((None, Some(2), None)) < est((None, Some(1), None)));
2269        assert!(est((None, None, Some(60))) < est((None, None, Some(50))));
2270        assert!(est((Some(7), None, None)) < est((None, Some(3), None)));
2271        // A big predicate spans many tiles; the rare one fits in at most one.
2272        let per_tile = n as f64 / idx.tile_sections()[0].len() as f64;
2273        assert!(est((None, Some(2), None)) <= 1.5 * per_tile);
2274        assert!(est((None, Some(1), None)) > 10.0 * per_tile);
2275        // A leading id beyond every tile's range is provably absent.
2276        assert_eq!(est((Some(99_999), None, None)), 0.0);
2277        // An empty index estimates nothing for a bound pattern.
2278        let empty = GraphIndexBuilder::new().build();
2279        assert_eq!(empty.estimate_matches((Some(1), None, None), 0), 0.0);
2280    }
2281
2282    #[test]
2283    fn every_pattern_shape_matches_reference() {
2284        let (idx, data) = graph();
2285        let vals = |opts: &[u32]| {
2286            let mut v: Vec<Option<u32>> = opts.iter().map(|&x| Some(x)).collect();
2287            v.push(None);
2288            v
2289        };
2290        // Exercise all 8 bound/unbound shapes across representative values.
2291        for s in vals(&[1, 2, 9]) {
2292            for p in vals(&[10, 11, 99]) {
2293                for o in vals(&[100, 300, 999]) {
2294                    let pat = (s, p, o);
2295                    assert_eq!(
2296                        idx.match_pattern(pat),
2297                        reference(&data, pat),
2298                        "pattern {pat:?}"
2299                    );
2300                    // The lazy stream must match the eager result once sorted —
2301                    // same triples, just without the up-front canonical re-sort.
2302                    let mut streamed: Vec<Triple> = idx.scan_iter(pat).collect();
2303                    streamed.sort_unstable();
2304                    assert_eq!(streamed, reference(&data, pat), "scan_iter {pat:?}");
2305                }
2306            }
2307        }
2308    }
2309
2310    #[test]
2311    fn unbound_returns_everything_sorted() {
2312        let (idx, data) = graph();
2313        let mut sorted = data.clone();
2314        sorted.sort_unstable();
2315        assert_eq!(idx.match_pattern((None, None, None)), sorted);
2316    }
2317
2318    /// A tiny tile budget must split sections into many tiles — including
2319    /// MID-GROUP cuts (budget 1/16 makes every a-group oversized) — and every
2320    /// pattern shape must still match the brute-force reference: bound
2321    /// leading ids route to the run of covering tiles, unbound scans chain
2322    /// all tiles.
2323    #[test]
2324    fn multi_tile_sections_match_reference_every_shape() {
2325        // Enough distinct leading ids to split under a tiny budget.
2326        let mut data: Vec<Triple> = Vec::new();
2327        for s in 1..=40u32 {
2328            for p in [10u32, 11] {
2329                for o in [100u32, 100 + s] {
2330                    data.push((s, p, o));
2331                }
2332            }
2333        }
2334        for budget in [1usize, 16, 64, 1 << 20] {
2335            let mut b = GraphIndexBuilder::new().with_tile_budget(budget);
2336            for &t in &data {
2337                b.push(t);
2338            }
2339            let idx = b.build();
2340            let spo_tiles = idx.tile_sections()[0].len();
2341            if budget <= 16 {
2342                assert!(spo_tiles > 1, "budget {budget} should force tiling");
2343            }
2344            // Tile ranges must be ascending; a split mega-group may leave
2345            // consecutive tiles SHARING a leading id (never overlapping past it).
2346            for w in idx.tile_sections()[0].as_full().unwrap().windows(2) {
2347                assert!(w[0].leading_range().1 <= w[1].leading_range().0);
2348            }
2349            assert_eq!(idx.triple_count() as usize, data.len(), "budget {budget}");
2350
2351            let vals = |opts: &[u32]| {
2352                let mut v: Vec<Option<u32>> = opts.iter().map(|&x| Some(x)).collect();
2353                v.push(None);
2354                v
2355            };
2356            for _round in 0..2 {
2357                for s in vals(&[1, 20, 40, 99]) {
2358                    for p in vals(&[10, 11, 99]) {
2359                        for o in vals(&[100, 120, 999]) {
2360                            let pat = (s, p, o);
2361                            assert_eq!(
2362                                idx.match_pattern(pat),
2363                                reference(&data, pat),
2364                                "budget {budget} pattern {pat:?}"
2365                            );
2366                        }
2367                    }
2368                }
2369            }
2370        }
2371    }
2372
2373    /// The Crossref shape: ONE mega-predicate carrying most of the graph. At
2374    /// the real 64 KiB budget its P-leading a-group must be split across
2375    /// multiple tiles (bounding builder memory), and a bound leading id must
2376    /// route to the whole run of covering tiles with complete results.
2377    #[test]
2378    fn mega_group_splits_across_tiles_and_lookups_stay_complete() {
2379        let hot_p = 7u32;
2380        let mut data: Vec<Triple> = Vec::new();
2381        for i in 0..40_000u32 {
2382            data.push((1_000 + i % 200, hot_p, 50_000 + i));
2383        }
2384        data.push((1, 1, 1));
2385        data.push((2, 2, 2));
2386        let mut b = GraphIndexBuilder::new(); // default INDEX_TILE_BUDGET
2387        for &t in &data {
2388            b.push(t);
2389        }
2390        let idx = b.build();
2391
2392        // PSO (a = predicate): the hot_p group must span several tiles.
2393        let pso = ALL_PERMS
2394            .iter()
2395            .position(|p| matches!(p, IndexPermutation::Pso))
2396            .unwrap();
2397        let covering = idx.tile_sections()[pso]
2398            .iter()
2399            .filter(|t| {
2400                let (lo, hi) = t.leading_range();
2401                lo <= hot_p && hot_p <= hi
2402            })
2403            .count();
2404        assert!(
2405            covering > 1,
2406            "expected the hot predicate split across tiles, got {covering}"
2407        );
2408
2409        // Bound-p lookup must still return every triple of the mega-group —
2410        // probing bound objects in the FIRST, MIDDLE and LAST slices of the
2411        // split run (a mid-run object once returned 0 on the first split file).
2412        for pat in [
2413            (None, Some(hot_p), None),
2414            (Some(1_050), Some(hot_p), None),
2415            (None, Some(hot_p), Some(50_123)), // first slice
2416            (None, Some(hot_p), Some(70_000)), // middle of the run
2417            (None, Some(hot_p), Some(89_999)), // last slice
2418            (None, Some(hot_p), Some(49_000)), // below the range → empty
2419            (None, Some(hot_p), Some(95_000)), // above the range → empty
2420        ] {
2421            assert_eq!(idx.match_pattern(pat), reference(&data, pat), "{pat:?}");
2422        }
2423    }
2424
2425    /// Scans must yield identical results across the directory lifecycle: the
2426    /// first leading-bound scan walks linearly, the second builds the
2427    /// directory, and later ones jump through it — including absent ids
2428    /// (between, below, and above the stored groups).
2429    #[test]
2430    fn directory_backed_scans_match_reference_every_shape() {
2431        let (idx, data) = graph();
2432        let vals = |opts: &[u32]| {
2433            let mut v: Vec<Option<u32>> = opts.iter().map(|&x| Some(x)).collect();
2434            v.push(None);
2435            v
2436        };
2437        for round in 0..3 {
2438            for s in vals(&[1, 2, 3, 0, 9]) {
2439                for p in vals(&[10, 11, 12, 99]) {
2440                    for o in vals(&[100, 200, 300, 999]) {
2441                        let pat = (s, p, o);
2442                        let mut got: Vec<Triple> = idx.scan_iter(pat).collect();
2443                        got.sort_unstable();
2444                        assert_eq!(got, reference(&data, pat), "round {round} scan {pat:?}");
2445                    }
2446                }
2447            }
2448        }
2449    }
2450
2451    /// A tile synopsis must let a routed scan **skip the fetch** of a tile its
2452    /// secondary-column range rules out — and never skip one that could match.
2453    #[test]
2454    fn synopsis_prunes_the_routed_tile_before_fetch() {
2455        use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
2456        use std::sync::Arc;
2457
2458        // One real SPO tile: subject a=5, predicates b∈{10,11}, object c=100.
2459        let block = {
2460            let mut b = TripleBlockBuilder::new();
2461            b.push((5, 10, 100));
2462            b.push((5, 11, 100));
2463            b.build()
2464        };
2465        let fetches = Arc::new(AtomicUsize::new(0));
2466        let (blk, fc) = (block.clone(), fetches.clone());
2467        // The loader returns the (already-decompressed) tile image and counts calls.
2468        let loader: TileLoader = Box::new(move |_si, _ti| {
2469            fc.fetch_add(1, SeqCst);
2470            Some(blk.clone())
2471        });
2472        // Remote SPO directory: one tile, leading a∈[5,5], synopsis b∈[10,11], c∈[100,100].
2473        let dirs = [
2474            vec![(5u32, 5u32, Some((10u32, 11u32, 100u32, 100u32)))],
2475            Vec::new(),
2476            Vec::new(),
2477            Vec::new(),
2478            Vec::new(),
2479            Vec::new(),
2480        ];
2481        let idx = GraphIndex::from_remote_directories(dirs, PermSet::ALL, loader);
2482
2483        // Predicate 99 is outside the tile's b-range → prune, zero fetches, empty.
2484        assert!(idx.match_pattern((Some(5), Some(99), None)).is_empty());
2485        assert_eq!(fetches.load(SeqCst), 0, "synopsis must skip the fetch");
2486
2487        // Object 999 outside the tile's c-range → prune, still zero fetches.
2488        assert!(idx.match_pattern((Some(5), None, Some(999))).is_empty());
2489        assert_eq!(
2490            fetches.load(SeqCst),
2491            0,
2492            "secondary-c prune also skips the fetch"
2493        );
2494
2495        // Predicate 10 is in range → the tile is fetched and the match returned.
2496        assert_eq!(
2497            idx.match_pattern((Some(5), Some(10), None)),
2498            vec![(5, 10, 100)]
2499        );
2500        assert_eq!(
2501            fetches.load(SeqCst),
2502            1,
2503            "an admissible secondary still fetches"
2504        );
2505    }
2506
2507    /// Without a synopsis (`None`), nothing is pruned early — the tile is always
2508    /// fetched and the in-tile zone map does the (correct) pruning, as before.
2509    #[test]
2510    fn absent_synopsis_never_prunes() {
2511        use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
2512        use std::sync::Arc;
2513        let block = {
2514            let mut b = TripleBlockBuilder::new();
2515            b.push((5, 10, 100));
2516            b.build()
2517        };
2518        let fetches = Arc::new(AtomicUsize::new(0));
2519        let (blk, fc) = (block.clone(), fetches.clone());
2520        let loader: TileLoader = Box::new(move |_si, _ti| {
2521            fc.fetch_add(1, SeqCst);
2522            Some(blk.clone())
2523        });
2524        let dirs = [
2525            vec![(5u32, 5u32, None)],
2526            Vec::new(),
2527            Vec::new(),
2528            Vec::new(),
2529            Vec::new(),
2530            Vec::new(),
2531        ];
2532        let idx = GraphIndex::from_remote_directories(dirs, PermSet::ALL, loader);
2533        // Predicate 99 absent, but with no synopsis the tile is fetched (then the
2534        // zone map yields no match) — correctness preserved, just no fetch saved.
2535        assert!(idx.match_pattern((Some(5), Some(99), None)).is_empty());
2536        assert_eq!(fetches.load(SeqCst), 1, "no synopsis ⇒ no early prune");
2537    }
2538
2539    /// The eight triple-pattern shapes, as `(s, p, o)` boundness.
2540    fn all_patterns() -> Vec<Pattern> {
2541        (0..8u8)
2542            .map(|mask| {
2543                (
2544                    (mask & 1 != 0).then_some(1u32),
2545                    (mask & 2 != 0).then_some(2u32),
2546                    (mask & 4 != 0).then_some(3u32),
2547                )
2548            })
2549            .collect()
2550    }
2551
2552    /// The claim the whole option rests on: **{SPO, POS, OSP} ties the longest
2553    /// bound prefix on all eight pattern shapes**, so the other three never win
2554    /// routing — only sort order. Enumerated and printed rather than asserted in
2555    /// prose, because if it were ever false a lean file would answer a pattern
2556    /// from a worse index without anything failing.
2557    #[test]
2558    fn perm_routing_never_leaves_core() {
2559        for pat in all_patterns() {
2560            let best_all = ALL_PERMS
2561                .iter()
2562                .map(|p| p.leading_bound(pat))
2563                .max()
2564                .unwrap();
2565            let best_core = PermSet::CORE
2566                .iter()
2567                .map(|p| p.leading_bound(pat))
2568                .max()
2569                .unwrap();
2570            let scores: Vec<String> = ALL_PERMS
2571                .iter()
2572                .map(|p| format!("{}={}", p.name(), p.leading_bound(pat)))
2573                .collect();
2574            eprintln!(
2575                "s={} p={} o={} | {} | best6={best_all} best3={best_core} chosen={}",
2576                pat.0.is_some() as u8,
2577                pat.1.is_some() as u8,
2578                pat.2.is_some() as u8,
2579                scores.join(" "),
2580                GraphIndex::best_permutation(pat).name(),
2581            );
2582            assert_eq!(
2583                best_core, best_all,
2584                "the three routing permutations must tie the best of six on {pat:?}"
2585            );
2586            let chosen = GraphIndex::best_permutation(pat);
2587            assert!(
2588                PermSet::CORE.contains(chosen),
2589                "best_permutation chose {} for {pat:?}, outside the routing three",
2590                chosen.name()
2591            );
2592            assert_eq!(
2593                chosen,
2594                GraphIndex::best_permutation_in(PermSet::CORE, pat),
2595                "routing must be identical with three permutations and with six"
2596            );
2597        }
2598    }
2599
2600    /// Restricting the set may never make a merge join pick a *worse-routed*
2601    /// stream than the plain lookup would use: with six the co-sorted
2602    /// permutation always ties the best prefix, and with three the guard
2603    /// declines rather than buying sort order with a whole-section scan.
2604    #[test]
2605    fn sorted_permutation_never_sacrifices_routing() {
2606        let mut declined = Vec::new();
2607        for pat in all_patterns() {
2608            for col in 0..3usize {
2609                let routed_all = GraphIndex::best_permutation(pat).leading_bound(pat);
2610                if let Some(p) = GraphIndex::permutation_sorted_on(pat, col) {
2611                    assert_eq!(
2612                        p.leading_bound(pat),
2613                        routed_all,
2614                        "six permutations must always co-sort at the best prefix"
2615                    );
2616                    assert_eq!(p.roles()[p.leading_bound(pat)], col);
2617                }
2618                match GraphIndex::permutation_sorted_on_in(PermSet::CORE, pat, col) {
2619                    Some(p) => {
2620                        assert!(PermSet::CORE.contains(p));
2621                        assert_eq!(
2622                            p.leading_bound(pat),
2623                            GraphIndex::best_permutation_in(PermSet::CORE, pat).leading_bound(pat)
2624                        );
2625                    }
2626                    None => {
2627                        if GraphIndex::permutation_sorted_on(pat, col).is_some() {
2628                            declined.push((pat, col));
2629                        }
2630                    }
2631                }
2632            }
2633        }
2634        // Exactly the three shapes SOP / PSO / OPS exist for: one bound
2635        // component, sorted on a column no remaining order can lead with after
2636        // it — subject-bound sorted on object, predicate-bound sorted on
2637        // subject, object-bound sorted on predicate.
2638        for (pat, col) in &declined {
2639            let bound: Vec<usize> = (0..3)
2640                .filter(|&i| [pat.0, pat.1, pat.2][i].is_some())
2641                .collect();
2642            assert_eq!(
2643                bound.len(),
2644                1,
2645                "only single-bound shapes lose a merge order"
2646            );
2647            assert_ne!(bound[0], *col);
2648        }
2649        eprintln!("three-permutation merge-join declines: {declined:?}");
2650        assert_eq!(declined.len(), 3, "exactly three (bound, sort) shapes lost");
2651    }
2652
2653    #[test]
2654    fn perm_set_masks_and_positions() {
2655        assert_eq!(PermSet::ALL.len(), 6);
2656        assert_eq!(PermSet::CORE.len(), 3);
2657        assert_eq!(PermSet::CORE.bits(), 0b0000_0111);
2658        assert_eq!(PermSet::CORE.names(), vec!["SPO", "POS", "OSP"]);
2659        assert_eq!(
2660            PermSet::ALL.names(),
2661            vec!["SPO", "POS", "OSP", "SOP", "PSO", "OPS"]
2662        );
2663        // Container position is rank among the STORED permutations, which is
2664        // why the routing three had to lead ALL_PERMS.
2665        for (i, p) in PermSet::CORE.iter().enumerate() {
2666            assert_eq!(PermSet::CORE.position(p), Some(i));
2667            assert_eq!(p.section_index(), i);
2668        }
2669        assert_eq!(PermSet::CORE.position(IndexPermutation::Sop), None);
2670        assert_eq!(PermSet::ALL.position(IndexPermutation::Ops), Some(5));
2671        // Only supersets of the routing three are legal.
2672        assert!(PermSet::from_bits(0b0000_0111).is_ok());
2673        assert!(PermSet::from_bits(0b0011_1111).is_ok());
2674        assert!(PermSet::from_bits(0b0000_1111).is_ok());
2675        assert!(PermSet::from_bits(0b0000_0011).is_err());
2676        assert!(PermSet::from_bits(0b0011_1000).is_err());
2677        assert!(PermSet::from_bits(0b1000_0111).is_err());
2678    }
2679
2680    /// A lean index answers every pattern shape with exactly the rows a full
2681    /// one does — the property that makes the option safe at all.
2682    #[test]
2683    fn three_and_six_agree_on_every_pattern() {
2684        let triples: Vec<Triple> = (0..200u32)
2685            .map(|i| (i % 17 + 1, i % 5 + 1, i % 23 + 1))
2686            .collect();
2687        let six = GraphIndexBuilder::from_triples(triples.clone())
2688            .with_tile_budget(128)
2689            .build();
2690        let three = GraphIndexBuilder::from_triples(triples)
2691            .with_tile_budget(128)
2692            .with_perms(PermSet::CORE)
2693            .build();
2694        assert_eq!(six.perms(), PermSet::ALL);
2695        assert_eq!(three.perms(), PermSet::CORE);
2696        for perm in [
2697            IndexPermutation::Sop,
2698            IndexPermutation::Pso,
2699            IndexPermutation::Ops,
2700        ] {
2701            assert!(three.tile_sections()[perm.section_index()].is_empty());
2702        }
2703        for a in [None, Some(3u32)] {
2704            for b in [None, Some(2u32)] {
2705                for c in [None, Some(9u32)] {
2706                    let mut x = six.match_pattern((a, b, c));
2707                    let mut y = three.match_pattern((a, b, c));
2708                    x.sort_unstable();
2709                    y.sort_unstable();
2710                    assert_eq!(x, y, "pattern {:?} disagreed", (a, b, c));
2711                }
2712            }
2713        }
2714    }
2715}