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