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, OnceLock};
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.
27pub const INDEX_TILE_BUDGET: usize = 64 * 1024;
28
29/// Geometric tile-prefetch ramp for an unbound (multi-tile) scan (see
30/// [`GraphIndex::scan_iter`]). The first coalesced batch faults this many
31/// tiles; each subsequent batch doubles up to [`PREFETCH_WINDOW_MAX`]. Small
32/// enough that `LIMIT 1` fetches only a few tiles, large enough that a full
33/// scan still coalesces into a handful of range reads.
34const PREFETCH_WINDOW_START: usize = 4;
35const PREFETCH_WINDOW_MAX: usize = 512;
36
37/// Fetches one tile's (uncompressed) block image on demand: the bridge that
38/// lets a remote `GraphIndex` fault tiles in over a `RangeReader` without this
39/// module knowing about I/O. `None` = the fetch failed; the index records the
40/// failure ([`GraphIndex::load_incomplete`]) and the scan sees an empty tile —
41/// callers running over remote data MUST check the flag after evaluating.
42pub type TileLoader = Box<dyn Fn(usize, usize) -> Option<Vec<u8>> + Send + Sync>;
43
44/// Fetches **many** tiles of one section in one round trip: given the section
45/// and the (ascending) indices of the tiles wanted, returns each tile's
46/// uncompressed block image in the same order. The ranged reader implements
47/// this by coalescing byte-adjacent tile ranges into single range reads, so a
48/// full-section scan costs a handful of requests instead of one per tile.
49/// `None` = the batch failed as a whole; callers fall back to the per-tile
50/// [`TileLoader`] (which records per-tile failures).
51pub type TileBulkLoader = Box<dyn Fn(usize, &[usize]) -> Option<Vec<Vec<u8>>> + Send + Sync>;
52
53/// One tile of a permutation section: a fully self-contained [`TripleBlock`]
54/// over a consecutive run of whole a-groups, plus the leading-id range it
55/// covers (for routing without parsing or fetching) and its lazily-built group
56/// directory (for intra-tile point lookups). The block image itself may be
57/// local (built/decoded eagerly) or faulted in by the index's [`TileLoader`].
58pub struct Tile {
59    min_a: u32,
60    max_a: u32,
61    /// Optional **tile synopsis**: the inclusive min/max of the two non-leading
62    /// columns `(min_b, max_b, min_c, max_c)`, read from the directory trailer of
63    /// a synopsis-carrying file. Lets a scan prune this tile by a bound secondary
64    /// component *before* faulting it (remote reads). `None` = no synopsis (an
65    /// older file, or a locally-built/opened tile) — then nothing is pruned early
66    /// and the in-tile zone map prunes after the tile is in hand, as before.
67    pub(crate) syn: Option<(u32, u32, u32, u32)>,
68    /// Encoded byte length of this tile (compressed on-disk size for a remote
69    /// tile, in-memory image size for a local one; 0 = unknown). Feeds the join
70    /// planner's fatness gates without faulting any data.
71    len: u32,
72    /// Build-time staging only: a locally-built or locally-decoded tile's body
73    /// waits here until [`GraphIndex::from_sections`] moves it into the shared
74    /// [`ChunkCache`] (keyed by `(section, tile)`) and clears it. `None`
75    /// afterward, and for a remote tile whose body faults in through the loader.
76    /// The long-lived retention is the cache, not this field.
77    staged: Option<Vec<u8>>,
78    dir: OnceLock<GroupDirectory>,
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            dir: OnceLock::new(),
91        }
92    }
93
94    fn remote(min_a: u32, max_a: u32, syn: Option<(u32, u32, u32, u32)>) -> Self {
95        Tile {
96            min_a,
97            max_a,
98            syn,
99            len: 0,
100            staged: None,
101            dir: OnceLock::new(),
102        }
103    }
104
105    /// Encoded byte length (see the field doc); falls back to a still-staged
106    /// body's size when the directory didn't provide one.
107    pub(crate) fn encoded_len(&self) -> u64 {
108        if self.len > 0 {
109            self.len as u64
110        } else {
111            self.staged.as_ref().map_or(0, |d| d.len() as u64)
112        }
113    }
114
115    /// Leading-component (permuted `a`) range this tile covers, inclusive.
116    pub fn leading_range(&self) -> (u32, u32) {
117        (self.min_a, self.max_a)
118    }
119
120    /// Could this tile hold a triple with the given bound non-leading components,
121    /// per its synopsis? `true` when there is no synopsis (can't rule it out) or
122    /// the bound `b`/`c` fall inside the recorded ranges. Conservative: a `false`
123    /// is a *proven* miss (the in-tile zone map would reject the same tile), so
124    /// skipping the fetch never drops a result.
125    pub(crate) fn syn_admits(&self, pb: Option<u32>, pc: Option<u32>) -> bool {
126        match self.syn {
127            None => true,
128            Some((min_b, max_b, min_c, max_c)) => {
129                let ok = |v: Option<u32>, lo: u32, hi: u32| v.is_none_or(|x| lo <= x && x <= hi);
130                ok(pb, min_b, max_b) && ok(pc, min_c, max_c)
131            }
132        }
133    }
134}
135
136/// What a routed scan of one pattern will touch, from the tile directories
137/// alone — see [`GraphIndex::scan_plan`].
138#[derive(Debug, Clone, Copy, PartialEq, Eq)]
139pub struct ScanPlan {
140    /// The permutation the scan routes to (always inside [`PermSet::CORE`]).
141    pub permutation: IndexPermutation,
142    /// Tiles in that permutation's section — the denominator.
143    pub tiles_total: usize,
144    /// Tiles left after routing on the bound leading component.
145    pub tiles_routed: usize,
146    /// Tiles left after the synopsis prune — what the scan will actually fetch.
147    pub tiles_admitted: usize,
148    /// Encoded (on-disk) bytes of the admitted tiles: the index-side upper
149    /// bound on the scan's range reads.
150    pub tile_bytes: u64,
151    /// Encoded bytes of the whole section — what an unpruned scan would read.
152    pub section_bytes: u64,
153}
154
155/// Which stored permutation to scan. The full **six** orders (a SPARQL engine's
156/// classic set, as in QLever): together they sort the triples on *every* prefix
157/// of `(s, p, o)` columns, so for any bound-component prefix and any free
158/// component there is a permutation that routes on the prefix **and** yields the
159/// stream sorted on that free component — the precondition for a merge join.
160#[derive(Debug, Clone, Copy, PartialEq, Eq)]
161pub enum IndexPermutation {
162    Spo,
163    Sop,
164    Pso,
165    Pos,
166    Osp,
167    Ops,
168}
169
170/// The number of permutations the format can address (the width of every
171/// per-permutation array). How many a *given file* stores is its [`PermSet`].
172pub(crate) const NUM_PERMS: usize = 6;
173
174/// The six permutations in section order — the canonical iteration list. The
175/// original three (SPO, POS, OSP) lead, so a default scan's tie-break (and its
176/// provenance) is unchanged from the 3-permutation format; the three added orders
177/// (SOP, PSO, OPS) follow and exist to give a merge join a stream sorted on the
178/// join key.
179pub(crate) const ALL_PERMS: [IndexPermutation; NUM_PERMS] = [
180    IndexPermutation::Spo,
181    IndexPermutation::Pos,
182    IndexPermutation::Osp,
183    IndexPermutation::Sop,
184    IndexPermutation::Pso,
185    IndexPermutation::Ops,
186];
187
188/// **Which** permutations a file stores — a 6-bit set over
189/// [`IndexPermutation::section_index`], recorded in the header so a reader
190/// never has to assume six.
191///
192/// Every legal set is a superset of [`PermSet::CORE`] = {SPO, POS, OSP}. That
193/// is not a convention: those three **tie the longest bound prefix on all eight
194/// triple-pattern shapes**, so [`GraphIndex::best_permutation`] never selects
195/// outside them and a lean file routes every pattern exactly where a full one
196/// does — same tiles, same rows. What the other three (SOP, PSO, OPS) add is a
197/// *sort order*: they let [`GraphIndex::permutation_sorted_on`] hand a
198/// sort-merge join a co-sorted stream for three of the twelve
199/// (bound-set, join-column) shapes it otherwise has to hash-join. Absent, the
200/// merge seed declines and the hash path answers — the cost is a fast path,
201/// never correctness (`bgp::try_merge_join` already returns `Option`).
202///
203/// `CORE` is the smallest legal set, [`PermSet::ALL`] is the default and what
204/// every file written before this existed carries.
205#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
206pub struct PermSet(u8);
207
208impl Default for PermSet {
209    fn default() -> Self {
210        PermSet::ALL
211    }
212}
213
214impl PermSet {
215    /// All six permutations — the default, and what a `0` header byte means.
216    pub const ALL: PermSet = PermSet(0b0011_1111);
217    /// The three routing permutations {SPO, POS, OSP}: every pattern shape
218    /// routes as well as it does with six, no merge-join sort orders.
219    pub const CORE: PermSet = PermSet(0b0000_0111);
220
221    /// Parse a raw mask. `Err` when it is not a superset of [`PermSet::CORE`]
222    /// or addresses a permutation this build has no section index for — a
223    /// pattern would then route to a permutation the file does not carry, which
224    /// is the one failure mode that must never be silent.
225    pub fn from_bits(bits: u8) -> Result<PermSet, &'static str> {
226        if bits & !PermSet::ALL.0 != 0 {
227            return Err("permutation mask addresses more than six permutations");
228        }
229        if bits & PermSet::CORE.0 != PermSet::CORE.0 {
230            return Err("permutation mask must contain SPO, POS and OSP");
231        }
232        Ok(PermSet(bits))
233    }
234
235    /// The raw 6-bit mask.
236    pub fn bits(self) -> u8 {
237        self.0
238    }
239
240    /// Is `perm` stored in this file?
241    pub fn contains(self, perm: IndexPermutation) -> bool {
242        self.0 & (1 << perm.section_index()) != 0
243    }
244
245    /// How many permutation sections the file's index container holds.
246    pub fn len(self) -> usize {
247        self.0.count_ones() as usize
248    }
249
250    /// Never true: every legal set contains [`PermSet::CORE`]. Present because
251    /// a public `len` without it is a clippy error, and the honest answer is
252    /// more useful than an `allow`.
253    pub fn is_empty(self) -> bool {
254        false
255    }
256
257    /// The stored permutations in [`ALL_PERMS`] (= container) order.
258    pub fn iter(self) -> impl Iterator<Item = IndexPermutation> {
259        ALL_PERMS.into_iter().filter(move |p| self.contains(*p))
260    }
261
262    /// `perm`'s position **inside the index container**, which is its rank
263    /// among the stored permutations — not [`IndexPermutation::section_index`],
264    /// which is its slot in the format's fixed six-wide addressing. The two
265    /// coincide only for [`PermSet::ALL`].
266    pub fn position(self, perm: IndexPermutation) -> Option<usize> {
267        self.contains(perm)
268            .then(|| (self.0 & ((1u8 << perm.section_index()) - 1)).count_ones() as usize)
269    }
270
271    /// Names in container order, e.g. `["SPO", "POS", "OSP"]`.
272    pub fn names(self) -> Vec<&'static str> {
273        self.iter().map(|p| p.name()).collect()
274    }
275
276    /// Does this set carry the three merge-join orders?
277    pub fn has_merge_orders(self) -> bool {
278        self == PermSet::ALL
279    }
280}
281
282impl IndexPermutation {
283    /// Stable display name used in CLI diagnostics and provenance records.
284    pub fn name(self) -> &'static str {
285        match self {
286            IndexPermutation::Spo => "SPO",
287            IndexPermutation::Sop => "SOP",
288            IndexPermutation::Pso => "PSO",
289            IndexPermutation::Pos => "POS",
290            IndexPermutation::Osp => "OSP",
291            IndexPermutation::Ops => "OPS",
292        }
293    }
294
295    /// Section number inside the stored permutation container.
296    pub fn section_index(self) -> usize {
297        match self {
298            IndexPermutation::Spo => 0,
299            IndexPermutation::Pos => 1,
300            IndexPermutation::Osp => 2,
301            IndexPermutation::Sop => 3,
302            IndexPermutation::Pso => 4,
303            IndexPermutation::Ops => 5,
304        }
305    }
306
307    /// The canonical `(s, p, o)` roles in this permutation's `(a, b, c)` slots,
308    /// as indices (0=s, 1=p, 2=o). The single source of truth for `forward` /
309    /// `back` / `order_pattern`.
310    pub(crate) const fn roles(self) -> [usize; 3] {
311        match self {
312            IndexPermutation::Spo => [0, 1, 2],
313            IndexPermutation::Sop => [0, 2, 1],
314            IndexPermutation::Pso => [1, 0, 2],
315            IndexPermutation::Pos => [1, 2, 0],
316            IndexPermutation::Osp => [2, 0, 1],
317            IndexPermutation::Ops => [2, 1, 0],
318        }
319    }
320
321    /// Map a canonical `(s, p, o)` triple into this permutation's `(a, b, c)`.
322    pub(crate) fn forward(self, t: Triple) -> Triple {
323        let c = [t.0, t.1, t.2];
324        let r = self.roles();
325        (c[r[0]], c[r[1]], c[r[2]])
326    }
327
328    /// Map a stored `(a, b, c)` back to canonical `(s, p, o)`.
329    fn back(self, abc: Triple) -> Triple {
330        let a = [abc.0, abc.1, abc.2];
331        let r = self.roles();
332        let mut out = [0u32; 3];
333        out[r[0]] = a[0];
334        out[r[1]] = a[1];
335        out[r[2]] = a[2];
336        (out[0], out[1], out[2])
337    }
338
339    /// The pattern's bound components in this permutation's component order.
340    pub(crate) fn order_pattern(self, p: Pattern) -> [Option<u32>; 3] {
341        let c = [p.0, p.1, p.2];
342        let r = self.roles();
343        [c[r[0]], c[r[1]], c[r[2]]]
344    }
345
346    /// Length of the leading run of bound components — higher is more selective.
347    fn leading_bound(self, pat: Pattern) -> usize {
348        self.order_pattern(pat)
349            .iter()
350            .take_while(|c| c.is_some())
351            .count()
352    }
353}
354
355/// Build a [`GraphIndex`] from canonical `(s, p, o)` integer triples.
356pub struct GraphIndexBuilder {
357    triples: Vec<Triple>,
358    tile_budget: usize,
359    perms: PermSet,
360}
361
362impl Default for GraphIndexBuilder {
363    fn default() -> Self {
364        Self {
365            triples: Vec::new(),
366            tile_budget: INDEX_TILE_BUDGET,
367            perms: PermSet::ALL,
368        }
369    }
370}
371
372impl GraphIndexBuilder {
373    pub fn new() -> Self {
374        Self::default()
375    }
376
377    /// Build directly from an owned triple vector — moves it in, avoiding the
378    /// per-triple `push` copy when the caller already has the id-triples in a
379    /// `Vec` (the assembly path, where they double as the pyramid input).
380    pub fn from_triples(triples: Vec<Triple>) -> Self {
381        Self {
382            triples,
383            tile_budget: INDEX_TILE_BUDGET,
384            perms: PermSet::ALL,
385        }
386    }
387
388    /// Override the per-tile byte budget (tests use tiny budgets to force
389    /// multi-tile sections on small data).
390    pub fn with_tile_budget(mut self, bytes: usize) -> Self {
391        self.tile_budget = bytes.max(1);
392        self
393    }
394
395    /// Build only the permutations in `perms` (default [`PermSet::ALL`]).
396    /// Sections outside the set come out empty and are not written to the file.
397    pub fn with_perms(mut self, perms: PermSet) -> Self {
398        self.perms = perms;
399        self
400    }
401
402    pub fn push(&mut self, t: Triple) {
403        self.triples.push(t);
404    }
405
406    /// Build the six permutations **one at a time** (each permutation's internal
407    /// sort still uses every core), so only a single permuted copy of the triples
408    /// is resident at once instead of all six concurrently. Trades the cross-
409    /// permutation parallelism of [`Self::build`] for a much lower peak RAM — the
410    /// large-graph low-memory path. Output is byte-identical to [`Self::build`].
411    pub fn build_seq(self) -> GraphIndex {
412        let triples = &self.triples;
413        let budget = self.tile_budget;
414        let wanted: Vec<IndexPermutation> = self.perms.iter().collect();
415        // Permute (parallel) + sort (parallel inside `build_tiles`) one permutation
416        // at a time, so only a single permuted copy of the triples is resident — but
417        // every core is still busy. Process permutations in batches of two to
418        // overlap one permutation's tile-building with the next one's sort while
419        // keeping the resident permuted copies to two. Byte-identical to `build`.
420        let build_one = |perm: IndexPermutation| -> Vec<Tile> {
421            #[cfg(feature = "parallel")]
422            let permuted: Vec<Triple> = {
423                use rayon::prelude::*;
424                triples.par_iter().map(|&t| perm.forward(t)).collect()
425            };
426            #[cfg(not(feature = "parallel"))]
427            let permuted: Vec<Triple> = triples.iter().map(|&t| perm.forward(t)).collect();
428            build_tiles(permuted, budget)
429        };
430        #[cfg(feature = "parallel")]
431        let built: Vec<Vec<Tile>> = {
432            use rayon::prelude::*;
433            let mut built: Vec<Vec<Tile>> = Vec::with_capacity(wanted.len());
434            for chunk in wanted.chunks(2) {
435                built.extend(
436                    chunk
437                        .to_vec()
438                        .into_par_iter()
439                        .map(build_one)
440                        .collect::<Vec<_>>(),
441                );
442            }
443            built
444        };
445        #[cfg(not(feature = "parallel"))]
446        let built: Vec<Vec<Tile>> = wanted.iter().copied().map(build_one).collect();
447        GraphIndex::from_sections(scatter(&wanted, built), self.perms)
448    }
449
450    pub fn build(self) -> GraphIndex {
451        let wanted: Vec<IndexPermutation> = self.perms.iter().collect();
452        let triples = &self.triples;
453        let budget = self.tile_budget;
454        // The permutations are independent — permute + sort + tile each.
455        let build_one = move |perm: IndexPermutation| -> Vec<Tile> {
456            let permuted: Vec<Triple> = triples.iter().map(|&t| perm.forward(t)).collect();
457            build_tiles(permuted, budget)
458        };
459        // Build the permutations concurrently when the `parallel` feature is on
460        // (they share no state); the per-permutation sort inside is also
461        // parallel. Output is byte-identical to the serial path.
462        #[cfg(feature = "parallel")]
463        let built: Vec<Vec<Tile>> = {
464            use rayon::iter::{IntoParallelIterator, ParallelIterator};
465            wanted.clone().into_par_iter().map(build_one).collect()
466        };
467        #[cfg(not(feature = "parallel"))]
468        let built: Vec<Vec<Tile>> = wanted.iter().copied().map(build_one).collect();
469        GraphIndex::from_sections(scatter(&wanted, built), self.perms)
470    }
471}
472
473/// Place each built section at its permutation's fixed [`IndexPermutation::section_index`]
474/// slot, leaving unbuilt permutations empty. The array stays six wide however
475/// few permutations a file carries, so every `sections[perm.section_index()]`
476/// in the engine keeps working unchanged.
477fn scatter(wanted: &[IndexPermutation], built: Vec<Vec<Tile>>) -> [Vec<Tile>; NUM_PERMS] {
478    let mut sections: [Vec<Tile>; NUM_PERMS] = Default::default();
479    for (perm, tiles) in wanted.iter().zip(built) {
480        sections[perm.section_index()] = tiles;
481    }
482    sections
483}
484
485/// The encoded varint length of `v` (LEB128).
486fn varint_len(mut v: u64) -> usize {
487    let mut n = 1;
488    while v >= 0x80 {
489        v >>= 7;
490        n += 1;
491    }
492    n
493}
494
495/// Incremental encoded-size accounting for one a-group of a tiled section —
496/// the running-delta chain of [`build_tiles`], streamable triple by triple.
497/// Shared by the in-memory tiler here and the external build's streaming
498/// tiler ([`crate::extbuild`]) so both choose IDENTICAL tile boundaries and
499/// their outputs stay byte-identical.
500pub(crate) struct GroupSizer {
501    /// finalized contributions: the a-delta plus every closed b-run
502    size: usize,
503    num_b: u64,
504    cur_b: u32,
505    /// the open b-run's c count / last c
506    num_c: u64,
507    prev_c: u32,
508    empty: bool,
509}
510
511impl GroupSizer {
512    /// Start a group for leading id `a` whose delta base is `prev_a` (the
513    /// previous group's leading id; `a` again for a mid-group continuation).
514    pub(crate) fn start(a: u32, prev_a: u32) -> Self {
515        GroupSizer {
516            size: varint_len((a - prev_a) as u64),
517            num_b: 0,
518            cur_b: 0,
519            num_c: 0,
520            prev_c: 0,
521            empty: true,
522        }
523    }
524
525    /// Account one `(b, c)` of this group; returns the group's total encoded
526    /// size so far (open count varints included).
527    pub(crate) fn push(&mut self, b: u32, c: u32) -> usize {
528        if self.empty || b != self.cur_b {
529            if self.empty {
530                self.size += varint_len(b as u64); // first b-run: delta from 0
531            } else {
532                self.size += varint_len(self.num_c); // close the previous b-run
533                self.size += varint_len((b - self.cur_b) as u64);
534            }
535            self.cur_b = b;
536            self.num_c = 0;
537            self.prev_c = 0;
538            self.num_b += 1;
539            self.empty = false;
540        }
541        self.size += varint_len((c - self.prev_c) as u64);
542        self.prev_c = c;
543        self.num_c += 1;
544        self.total()
545    }
546
547    /// The group's total encoded size so far.
548    pub(crate) fn total(&self) -> usize {
549        self.size
550            + if self.empty {
551                0
552            } else {
553                varint_len(self.num_c)
554            }
555            + varint_len(self.num_b)
556    }
557}
558
559/// Split sorted, deduped permuted triples into size-targeted tiles: append
560/// whole a-groups until the (estimated, near-exact) encoded size would exceed
561/// `budget`, then flush — and additionally cut WITHIN an a-group whose own
562/// running size exceeds the budget (a mega-group: one predicate or class
563/// carrying a large share of the graph, e.g. a 2B-triple `cites` predicate in
564/// POS). Split slices become consecutive tiles sharing the leading id — a
565/// bound leading id routes to the whole run of covering tiles (see
566/// [`GraphIndex::tile_span`]) — which bounds both builder memory and the
567/// bytes a remote reader faults for one lookup. Boundary accounting lives in
568/// [`GroupSizer`], shared with the external build for byte-identity.
569fn build_tiles(mut triples: Vec<Triple>, budget: usize) -> Vec<Tile> {
570    #[cfg(feature = "parallel")]
571    {
572        use rayon::slice::ParallelSliceMut;
573        triples.par_sort_unstable();
574    }
575    #[cfg(not(feature = "parallel"))]
576    triples.sort_unstable();
577    triples.dedup();
578    if triples.is_empty() {
579        return Vec::new();
580    }
581
582    let make_tile = |run: &[Triple]| -> Tile {
583        let mut b = TripleBlockBuilder::new();
584        for &t in run {
585            b.push(t);
586        }
587        Tile::local(run[0].0, run[run.len() - 1].0, b.build())
588    };
589
590    let mut tiles = Vec::new();
591    let mut tile_start = 0usize;
592    let mut tile_size = 0usize; // completed groups in the current tile
593    let mut prev_a = 0u32;
594    let mut i = 0usize;
595    while i < triples.len() {
596        let a = triples[i].0;
597        let mut slice_start = i; // group start, or the last mid-group cut
598        let mut sizer = GroupSizer::start(a, prev_a);
599        let mut gtotal = 0usize;
600        while i < triples.len() && triples[i].0 == a {
601            gtotal = sizer.push(triples[i].1, triples[i].2);
602            i += 1;
603            if gtotal > budget {
604                // Mega-group cut: everything buffered — completed groups plus
605                // the slice measured so far — becomes one tile; the group
606                // continues in a fresh chain with `a` as its own delta base.
607                tiles.push(make_tile(&triples[tile_start..i]));
608                tile_start = i;
609                tile_size = 0;
610                prev_a = a;
611                slice_start = i;
612                sizer = GroupSizer::start(a, a);
613                gtotal = 0;
614            }
615        }
616        if slice_start == i {
617            continue; // the cut landed exactly on the group's end
618        }
619        if slice_start > tile_start && tile_size + gtotal > budget {
620            tiles.push(make_tile(&triples[tile_start..slice_start]));
621            tile_start = slice_start;
622            tile_size = 0;
623        }
624        tile_size += gtotal;
625        prev_a = a;
626    }
627    if tile_start < triples.len() {
628        tiles.push(make_tile(&triples[tile_start..]));
629    }
630    tiles
631}
632
633/// The six tiled permutation sections, queryable by triple pattern.
634pub struct GraphIndex {
635    /// Tiles per permutation (SPO, SOP, PSO, POS, OSP, OPS — see [`ALL_PERMS`]),
636    /// ascending in their leading-id ranges; consecutive tiles may share a
637    /// leading id when a mega-group was split (see [`build_tiles`]).
638    pub(crate) sections: [Vec<Tile>; NUM_PERMS],
639    /// Which of those six slots this file actually carries. Sections outside
640    /// the set are empty and must never be consulted — an empty section is
641    /// indistinguishable from an empty *graph*, which is why the planner asks
642    /// this rather than the section lengths.
643    perms: PermSet,
644    /// Faults in remote tiles on first scan (`None` for local indexes).
645    loader: Option<TileLoader>,
646    /// Optional batched fetch for multi-tile scans (`None` falls back to
647    /// one [`TileLoader`] call per tile).
648    bulk: Option<TileBulkLoader>,
649    /// Set when the loader failed for some tile: results may be incomplete and
650    /// the caller must surface an error rather than the partial answer.
651    load_failed: std::sync::atomic::AtomicBool,
652    /// The reader's concurrent-range fan-out (1 = strictly sequential) — see
653    /// [`set_read_concurrency`](Self::set_read_concurrency).
654    read_concurrency: usize,
655    /// Decompressed tile bodies live here, keyed by `(section, tile)`, faulted
656    /// in on first scan and evictable. Phase 0 caps it at unlimited, so a
657    /// scanned permutation stays resident exactly as it did in the old per-tile
658    /// `OnceLock`s.
659    cache: Arc<ChunkCache>,
660}
661
662impl GraphIndex {
663    fn from_sections(mut sections: [Vec<Tile>; NUM_PERMS], perms: PermSet) -> Self {
664        // Move every locally-built/decoded tile body into the shared cache and
665        // clear its staging slot, so the descriptors hold routing metadata only.
666        let cache = ChunkCache::unlimited_arc();
667        for (si, tiles) in sections.iter_mut().enumerate() {
668            for (ti, tile) in tiles.iter_mut().enumerate() {
669                if let Some(body) = tile.staged.take() {
670                    cache.insert((si as u8, ti as u32), Arc::from(body));
671                }
672            }
673        }
674        GraphIndex {
675            sections,
676            perms,
677            loader: None,
678            bulk: None,
679            load_failed: std::sync::atomic::AtomicBool::new(false),
680            read_concurrency: 1,
681            cache,
682        }
683    }
684
685    /// Rebuild from tiled sections: per permutation, `(min_a, max_a, block
686    /// bytes)` per tile in ascending leading-id order — the v0.2 layout.
687    /// Slots outside `perms` must be empty.
688    pub fn from_tiles(sections: [Vec<(u32, u32, Vec<u8>)>; NUM_PERMS], perms: PermSet) -> Self {
689        let sections = sections.map(|tiles| {
690            tiles
691                .into_iter()
692                .map(|(min_a, max_a, bytes)| Tile::local(min_a, max_a, bytes))
693                .collect()
694        });
695        Self::from_sections(sections, perms)
696    }
697
698    /// The permutations this index carries.
699    pub fn perms(&self) -> PermSet {
700        self.perms
701    }
702
703    /// Set the byte cap of this index's tile cache (bounded-export phase 2).
704    /// `u64::MAX` (the open-time default) is unlimited — a scanned permutation
705    /// stays resident exactly as before. A finite cap evicts least-recently-used
706    /// tiles, so a full scan's tile residency is bounded.
707    ///
708    /// Only safe on an index whose tiles can be **re-faulted** (a remote/ranged
709    /// index with a loader). A locally-built index has no loader — an evicted
710    /// tile could never be reloaded — so the caller must not cap those; the
711    /// bounded-export path caps only the ranged default-graph index and leaves
712    /// resident named-graph indexes unlimited (released per slot instead).
713    pub fn set_cache_cap(&self, cap: u64) {
714        self.cache.set_cap(cap);
715    }
716
717    /// The current tile-cache byte cap (`u64::MAX` = unlimited).
718    pub fn cache_cap(&self) -> u64 {
719        self.cache.cap()
720    }
721
722    /// Does this index fault its tiles through a loader (a remote/ranged open)?
723    /// Only such an index is safe to give a finite tile-cache cap: a locally
724    /// decoded index has no loader, so an evicted tile could never be reloaded.
725    pub fn is_lazy(&self) -> bool {
726        self.loader.is_some()
727    }
728
729    /// A snapshot of the tile cache's observability counters.
730    pub fn cache_stats(&self) -> crate::chunk_cache::CacheStats {
731        self.cache.stats()
732    }
733
734    /// A **remote** index: only the tile directories (leading-id ranges per
735    /// permutation, ascending) are known; tile payloads fault in through
736    /// `loader` on first scan. Check [`load_incomplete`](Self::load_incomplete)
737    /// after evaluating — a failed fetch must become an error, never a
738    /// silently smaller result.
739    #[allow(clippy::type_complexity)]
740    pub fn from_remote_directories(
741        directories: [Vec<(u32, u32, Option<(u32, u32, u32, u32)>)>; NUM_PERMS],
742        perms: PermSet,
743        loader: TileLoader,
744    ) -> Self {
745        let sections = directories.map(|dir| {
746            dir.into_iter()
747                .map(|(min_a, max_a, syn)| Tile::remote(min_a, max_a, syn))
748                .collect()
749        });
750        GraphIndex {
751            sections,
752            perms,
753            loader: Some(loader),
754            bulk: None,
755            load_failed: std::sync::atomic::AtomicBool::new(false),
756            read_concurrency: 1,
757            cache: ChunkCache::unlimited_arc(),
758        }
759    }
760
761    /// Attach a batched tile fetcher (see [`TileBulkLoader`]): multi-tile
762    /// scans prefetch their span through it instead of faulting tile by tile.
763    pub fn with_bulk_loader(mut self, bulk: TileBulkLoader) -> Self {
764        self.bulk = Some(bulk);
765        self
766    }
767
768    /// Record each remote tile's encoded (on-disk) byte length, per section —
769    /// known to the ranged opener from the tile directory. Powers the join
770    /// planner's fatness gates; never triggers a fetch.
771    pub(crate) fn set_tile_lens(&mut self, lens: [Vec<u32>; NUM_PERMS]) {
772        for (section, ls) in self.sections.iter_mut().zip(lens) {
773            for (tile, l) in section.iter_mut().zip(ls) {
774                tile.len = l;
775            }
776        }
777    }
778
779    /// Record the reader's concurrent-range fan-out (see
780    /// [`RangeReader::concurrency`](crate::reader::RangeReader::concurrency)) —
781    /// the join planner widens its remote probe budget when round trips
782    /// overlap instead of serializing.
783    pub(crate) fn set_read_concurrency(&mut self, c: usize) {
784        self.read_concurrency = c.max(1);
785    }
786
787    /// The reader's concurrent-range fan-out (1 = strictly sequential).
788    pub(crate) fn read_concurrency(&self) -> usize {
789        self.read_concurrency
790    }
791
792    /// Did any tile fetch fail since this index was opened — or since the last
793    /// [`reset_load_failure`](Self::reset_load_failure)?
794    pub fn load_incomplete(&self) -> bool {
795        self.load_failed.load(std::sync::atomic::Ordering::Relaxed)
796    }
797
798    /// Forget recorded fetch failures — the start-of-evaluation reset for a
799    /// RESIDENT session, making the incompleteness verdict per-query instead of
800    /// per-open (one transient network blip used to fail every later query on
801    /// the session). Safe because failed tiles are never cached: the next scan
802    /// simply retries them.
803    pub fn reset_load_failure(&self) {
804        self.load_failed
805            .store(false, std::sync::atomic::Ordering::Relaxed);
806    }
807
808    /// True for a remote/lazy index (tiles fault in over a `RangeReader`). The
809    /// join planner uses this to pick read-aware strategies: here a per-row index
810    /// probe is a network round-trip, not a memory lookup, so a left-deep scan +
811    /// hash join often beats probing a moderately-sized prefix.
812    pub fn is_remote(&self) -> bool {
813        self.loader.is_some()
814    }
815
816    /// Batch-fault the unloaded tiles in `[start, end)` of `section` through
817    /// the bulk loader, if one is attached and at least two tiles are missing
818    /// (a single missing tile costs the same either way). A failed batch is
819    /// not an error here: the tiles stay unloaded and the per-tile loader
820    /// retries each one (recording failures) when the scan reaches it.
821    fn prefetch_span(&self, section: usize, start: usize, end: usize) {
822        let missing: Vec<usize> = (start..end)
823            .filter(|&ti| !self.tile_resident(section, ti))
824            .collect();
825        self.bulk_fault(section, &missing);
826    }
827
828    /// Is tile `(section, tile)`'s body currently resident in the cache?
829    fn tile_resident(&self, section: usize, tile: usize) -> bool {
830        self.cache.contains((section as u8, tile as u32))
831    }
832
833    /// Bulk-fault a set of (possibly scattered, ascending) missing tile indices in
834    /// `section` through the bulk loader in one coalesced read, storing each image.
835    /// No-op without a bulk loader or for fewer than two tiles (a single fault
836    /// costs the same via the per-tile loader). A failed batch leaves the tiles
837    /// unloaded for the per-tile loader to retry. Shared by the consecutive-span
838    /// scan prefetch and the scattered batch-probe prefetch.
839    fn bulk_fault(&self, section: usize, tiles: &[usize]) {
840        if tiles.len() < 2 {
841            return;
842        }
843        let Some(bulk) = &self.bulk else { return };
844        if let Some(images) = bulk(section, tiles) {
845            if images.len() == tiles.len() {
846                for (&ti, img) in tiles.iter().zip(images) {
847                    self.cache
848                        .insert((section as u8, ti as u32), Arc::from(img));
849                }
850            }
851        }
852    }
853
854    /// Batch-fault the tiles a set of upcoming **probe** patterns will route to,
855    /// before they are probed one at a time. Each probe (a bound leading id)
856    /// routes to a single tile; gathered across the batch those tiles are
857    /// scattered, so faulting them together turns N sequential remote round trips
858    /// into a few coalesced parallel reads — the read-amplification win for a
859    /// label-heavy join over a lazy reader. Honors the synopsis prune (never
860    /// fetches a tile a bound secondary proves can't match). No-op for a local
861    /// index or when fewer than two tiles per section need faulting.
862    pub(crate) fn prefetch_probe_tiles(&self, patterns: &[Pattern]) {
863        if self.bulk.is_none() {
864            return;
865        }
866        let mut want: [std::collections::BTreeSet<usize>; NUM_PERMS] = Default::default();
867        for &pat in patterns {
868            let perm = Self::best_permutation(pat);
869            let [pa, pb, pc] = perm.order_pattern(pat);
870            let si = perm.section_index();
871            let (start, end) = self.tile_span(si, pa);
872            for ti in start..end {
873                if self.sections[si][ti].syn_admits(pb, pc) && !self.tile_resident(si, ti) {
874                    want[si].insert(ti);
875                }
876            }
877        }
878        for (si, set) in want.iter().enumerate() {
879            let tiles: Vec<usize> = set.iter().copied().collect();
880            self.bulk_fault(si, &tiles);
881        }
882    }
883
884    /// The tile's cache entry, faulting its block image in through the loader if
885    /// remote. Returns an owned [`CacheEntry`] handle (an `Arc`) the caller
886    /// holds for the duration of one parse/scan — valid even if a later eviction
887    /// drops the map slot (the handout invariant).
888    ///
889    /// A FAILED fetch records the failure and caches nothing (returns `None`),
890    /// so a later evaluation retries the tile — a transient network error must
891    /// not permanently poison a long-lived (resident) session. A local tile
892    /// with no body returns an empty entry (constructed empty on purpose).
893    fn tile_entry(&self, section: usize, tile: usize) -> Option<Arc<CacheEntry>> {
894        let key = (section as u8, tile as u32);
895        if let Some(e) = self.cache.get(key) {
896            return Some(e);
897        }
898        match &self.loader {
899            Some(load) => match load(section, tile) {
900                Some(bytes) => Some(self.cache.insert(key, Arc::from(bytes))),
901                None => {
902                    self.load_failed
903                        .store(true, std::sync::atomic::Ordering::Relaxed);
904                    None
905                }
906            },
907            None => Some(self.cache.insert(key, Arc::from(Vec::new()))),
908        }
909    }
910
911    /// The tile's decompressed block image as a shared handle, for the file
912    /// writer (local indexes, all tiles resident). Empty on a miss.
913    pub(crate) fn tile_body(&self, section: usize, tile: usize) -> Arc<[u8]> {
914        self.tile_entry(section, tile)
915            .map(|e| e.body_arc())
916            .unwrap_or_else(|| Arc::from(Vec::new()))
917    }
918
919    /// Total triple count (sum of the SPO tiles' zone counts). For a remote
920    /// index this faults in the SPO tiles — prefer the header's quad count.
921    pub fn triple_count(&self) -> u32 {
922        self.prefetch_span(0, 0, self.sections[0].len());
923        (0..self.sections[0].len())
924            .filter_map(|ti| {
925                let entry = self.tile_entry(0, ti)?;
926                TripleBlock::parse(entry.body())
927                    .ok()
928                    .map(|b| b.zone().count)
929            })
930            .sum()
931    }
932
933    /// The tiles of each permutation section (in `ALL_PERMS` order), for the
934    /// file writer.
935    pub fn tile_sections(&self) -> [&[Tile]; NUM_PERMS] {
936        [
937            &self.sections[0],
938            &self.sections[1],
939            &self.sections[2],
940            &self.sections[3],
941            &self.sections[4],
942            &self.sections[5],
943        ]
944    }
945
946    /// The permutation selected for a pattern: the one with the longest bound
947    /// prefix. Ties keep the canonical SPO order (then POS, then OSP), which
948    /// makes provenance stable for unbound or equally selective shapes and routes
949    /// a fully unbound pattern to the SPO block rather than fetching all three.
950    ///
951    /// Six-permutation shorthand for [`best_permutation_in`]. Its answer is
952    /// always inside [`PermSet::CORE`] (see `perm_routing_never_leaves_core`),
953    /// so a file with any legal [`PermSet`] routes identically — but call the
954    /// `_in` form wherever the file's set is at hand, so that stays true by
955    /// construction rather than by argument.
956    ///
957    /// [`best_permutation_in`]: GraphIndex::best_permutation_in
958    pub fn best_permutation(pattern: Pattern) -> IndexPermutation {
959        Self::best_permutation_in(PermSet::ALL, pattern)
960    }
961
962    /// [`best_permutation`](GraphIndex::best_permutation) restricted to the
963    /// permutations a file carries.
964    pub fn best_permutation_in(perms: PermSet, pattern: Pattern) -> IndexPermutation {
965        let mut best = IndexPermutation::Spo;
966        let mut best_score = best.leading_bound(pattern);
967        for perm in perms.iter() {
968            let score = perm.leading_bound(pattern);
969            if score > best_score {
970                best = perm;
971                best_score = score;
972            }
973        }
974        best
975    }
976
977    /// Choose a permutation that **routes** on `pattern`'s bound prefix *and*
978    /// streams sorted on the canonical column `sort_col` (0=subject, 1=predicate,
979    /// 2=object) — i.e. `sort_col` is the leading *free* component after the bound
980    /// prefix. Returns `None` if `sort_col` is itself bound (then every row shares
981    /// that value — any permutation is "sorted" on it) or no permutation qualifies.
982    /// This is the precondition a merge join needs: both inputs sorted on the join
983    /// key. Among qualifiers, the longest bound prefix (best routing) wins.
984    ///
985    /// Six-permutation shorthand for [`permutation_sorted_on_in`].
986    ///
987    /// [`permutation_sorted_on_in`]: GraphIndex::permutation_sorted_on_in
988    pub fn permutation_sorted_on(pattern: Pattern, sort_col: usize) -> Option<IndexPermutation> {
989        Self::permutation_sorted_on_in(PermSet::ALL, pattern, sort_col)
990    }
991
992    /// [`permutation_sorted_on`](GraphIndex::permutation_sorted_on) restricted
993    /// to the permutations a file carries — **and it will not trade routing for
994    /// sort order**.
995    ///
996    /// A qualifier must match the bound prefix that
997    /// [`best_permutation_in`](GraphIndex::best_permutation_in) achieves.
998    /// With all six that costs nothing: every ordering of `(s, p, o)` exists, so
999    /// "the bound components first, then `sort_col`" is always available at the
1000    /// maximal prefix. With [`PermSet::CORE`] it is what stops a merge join from
1001    /// buying its co-sorted stream with a **whole-section scan** — e.g.
1002    /// `?s :p ?o` sorted on `?s` has no `P*S` order left, and the only remaining
1003    /// sorted-on-subject stream is an unrouted SPO scan of the entire graph. The
1004    /// merge seed then declines and the hash/probe path answers, which is
1005    /// strictly the cheaper plan.
1006    pub fn permutation_sorted_on_in(
1007        perms: PermSet,
1008        pattern: Pattern,
1009        sort_col: usize,
1010    ) -> Option<IndexPermutation> {
1011        let bound = [
1012            pattern.0.is_some(),
1013            pattern.1.is_some(),
1014            pattern.2.is_some(),
1015        ];
1016        if bound[sort_col] {
1017            return None;
1018        }
1019        let routed = Self::best_permutation_in(perms, pattern).leading_bound(pattern);
1020        let mut best: Option<(IndexPermutation, usize)> = None;
1021        for perm in perms.iter() {
1022            let roles = perm.roles();
1023            let lead = perm.leading_bound(pattern);
1024            // The component at slot `lead` is the first free one; it must be the
1025            // sort column (so the stream is sorted on it after the bound prefix).
1026            if lead < 3 && roles[lead] == sort_col && best.map(|(_, s)| lead > s).unwrap_or(true) {
1027                best = Some((perm, lead));
1028            }
1029        }
1030        best.filter(|&(_, lead)| lead >= routed).map(|(p, _)| p)
1031    }
1032
1033    /// Match one already-decoded serialized permutation block (a single v0.1
1034    /// section or one v0.2 tile). This is the core primitive for range-routed
1035    /// readers: the caller fetches only the selected payload, then this scans
1036    /// it as if it came from a full [`GraphIndex`].
1037    pub fn match_serialized_block(
1038        bytes: &[u8],
1039        permutation: IndexPermutation,
1040        pattern: Pattern,
1041    ) -> Vec<Triple> {
1042        let [pa, pb, pc] = permutation.order_pattern(pattern);
1043        let mut out: Vec<Triple> = TripleBlock::parse(bytes)
1044            .ok()
1045            .filter(|b| b.zone().may_contain(pa, pb, pc))
1046            .map(|b| b.scan(pa, pb, pc))
1047            .into_iter()
1048            .flatten()
1049            .map(move |abc| permutation.back(abc))
1050            .collect();
1051        out.sort_unstable();
1052        out
1053    }
1054
1055    /// All triples matching `pattern`, returned in canonical `(s, p, o)` order.
1056    ///
1057    /// Thin eager wrapper over [`scan_iter`](Self::scan_iter): collect the lazy
1058    /// stream and restore the canonical sort (the stream is sorted in the chosen
1059    /// permutation's order, which differs from canonical once `perm.back`
1060    /// permutes the free components).
1061    pub fn match_pattern(&self, pattern: Pattern) -> Vec<Triple> {
1062        let mut out: Vec<Triple> = self.scan_iter(pattern).collect();
1063        out.sort_unstable();
1064        out
1065    }
1066
1067    /// Lazily stream the triples matching `pattern` in canonical `(s, p, o)`
1068    /// order *within the chosen permutation* — the streaming entry point for
1069    /// callers that can stop early (ASK, `LIMIT`, BGP probes) or that don't need
1070    /// the canonical re-sort. Decodes only the matching groups (see
1071    /// [`TripleBlock::scan`]); a malformed/absent block yields nothing rather
1072    /// than panicking. The permutation is chosen for the longest bound prefix.
1073    pub fn scan_iter(&self, pattern: Pattern) -> impl Iterator<Item = Triple> + '_ {
1074        self.scan_iter_with(pattern, Self::best_permutation(pattern))
1075    }
1076
1077    /// What [`scan_iter`](Self::scan_iter) *would* fetch for `pattern`, computed
1078    /// from the tile directories alone — **no tile is fetched, decoded, or even
1079    /// looked at**. This is the cost-preview half of the routed scan: it walks
1080    /// the same three decisions the scan makes (permutation, routed tile span,
1081    /// synopsis prune) and reports what survives, so a caller can say "this dump
1082    /// will pull 16 MB of index, not 376 MB" before starting it.
1083    ///
1084    /// It is a preview, not a promise: `tile_bytes` is an **upper bound** on the
1085    /// index bytes the scan reads (an admitted tile can still be rejected by its
1086    /// in-tile zone map once in hand, and a re-scan of an already-faulted tile
1087    /// reads nothing), and it says nothing about the dictionary, which is where
1088    /// a term-resolving dump's real cost usually lives.
1089    pub fn scan_plan(&self, pattern: Pattern) -> ScanPlan {
1090        let perm = Self::best_permutation_in(self.perms, pattern);
1091        let [_, pb, pc] = perm.order_pattern(pattern);
1092        let si = perm.section_index();
1093        let tiles = &self.sections[si];
1094        let (start, end) = self.tile_span(si, perm.order_pattern(pattern)[0]);
1095        let mut admitted = 0usize;
1096        let mut tile_bytes = 0u64;
1097        for t in &tiles[start..end] {
1098            if t.syn_admits(pb, pc) {
1099                admitted += 1;
1100                tile_bytes += t.encoded_len();
1101            }
1102        }
1103        ScanPlan {
1104            permutation: perm,
1105            tiles_total: tiles.len(),
1106            tiles_routed: end - start,
1107            tiles_admitted: admitted,
1108            tile_bytes,
1109            section_bytes: tiles.iter().map(|t| t.encoded_len()).sum(),
1110        }
1111    }
1112
1113    /// One bounded, **resumable** slice of `pattern`'s matches: the pull form of
1114    /// [`scan_iter`](Self::scan_iter), for a caller that cannot hold a borrow
1115    /// across calls.
1116    ///
1117    /// Returns `(triples, next_cursor, done)`; start at `cursor = 0` and feed the
1118    /// returned cursor back until `done`. The whole resume state is one opaque
1119    /// `u64` — `(tile index, next a-group id)` — so no iterator, no thread and no
1120    /// self-referential struct has to survive between calls. That is the shape a
1121    /// foreign-function boundary needs: a wasm module cannot suspend a Rust
1122    /// iterator mid-scan and hand control back to its host, but it can be handed
1123    /// a `u64` and told to carry on.
1124    ///
1125    /// `max_rows` is a **floor**, not a hard cut: a batch always ends on an
1126    /// a-group boundary of the chosen permutation, so no group is ever split
1127    /// across two calls and the overshoot is one group's fanout. Nothing is ever
1128    /// rescanned ([`TripleBlock::scan_resume`] jumps straight to the group
1129    /// through the tile's directory), so a full drain is O(n) overall — unlike
1130    /// `(offset, limit)`, which is O(n²/limit).
1131    ///
1132    /// Every call either yields at least one row or reports `done`, so a caller
1133    /// never has to guard against an empty non-final batch.
1134    ///
1135    /// **Order.** Rows arrive in the chosen permutation's order and each batch is
1136    /// re-sorted canonically, exactly as [`match_pattern`](Self::match_pattern)
1137    /// does for a whole result. For a fully unbound pattern (routed to SPO) that
1138    /// makes the concatenation of every batch identical to `match_pattern`'s
1139    /// output; for a bound pattern the *set* is identical but the canonical
1140    /// re-sort is per batch rather than global.
1141    ///
1142    /// **Memory** is O(`max_rows` + one a-group + the tiles faulted so far) —
1143    /// never O(matches). That is the whole point: an unbounded `(?s ?p ?o)` over
1144    /// a 26-million-quad graph used to materialize every match before the caller
1145    /// saw one row.
1146    pub fn scan_batch(
1147        &self,
1148        pattern: Pattern,
1149        cursor: u64,
1150        max_rows: usize,
1151    ) -> (Vec<Triple>, u64, bool) {
1152        let max_rows = max_rows.max(1);
1153        let perm = Self::best_permutation(pattern);
1154        let [pa, pb, pc] = perm.order_pattern(pattern);
1155        let si = perm.section_index();
1156        // Same routing as `scan_iter_with`: the tile directory alone decides the
1157        // span, so this is free on a lazy/remote open.
1158        let (start, end) = self.tile_span(si, pa);
1159        let mut ti = ((cursor >> 32) as usize).max(start);
1160        let mut from_a = cursor as u32;
1161        let mut out: Vec<Triple> = Vec::new();
1162        // Same geometric prefetch ramp as the streaming scan: a batch that stops
1163        // after one tile never fetches the ones behind it.
1164        let mut window = PREFETCH_WINDOW_START;
1165        while ti < end {
1166            // Synopsis pre-fault prune — no tile is fetched to reject it.
1167            if !self.sections[si][ti].syn_admits(pb, pc) {
1168                ti += 1;
1169                from_a = 0;
1170                continue;
1171            }
1172            if !self.tile_resident(si, ti) {
1173                self.prefetch_span(si, ti, (ti + window).min(end));
1174                window = window.saturating_mul(2).min(PREFETCH_WINDOW_MAX);
1175            }
1176            let tile = &self.sections[si][ti];
1177            let entry = self.tile_entry(si, ti);
1178            let mut resume_at: Option<u32> = None;
1179            if let Some(block) = entry
1180                .as_deref()
1181                .and_then(|e| TripleBlock::parse(e.body()).ok())
1182                .filter(|b| b.zone().may_contain(pa, pb, pc))
1183            {
1184                let rows = match pa {
1185                    // A bound leading id is one a-group; it cannot be split
1186                    // inside a tile, only across tiles (a mega-group split), so
1187                    // there is nothing finer to resume at than the tile.
1188                    Some(a) => {
1189                        let dir = tile.dir.get_or_init(|| block.group_directory());
1190                        block.scan_from(dir, a, pb, pc)
1191                    }
1192                    // Fresh tile: no directory needed, so a straight drain never
1193                    // pays for one.
1194                    None if from_a == 0 => block.scan(None, pb, pc),
1195                    None => {
1196                        let dir = tile.dir.get_or_init(|| block.group_directory());
1197                        block.scan_resume(dir, from_a, pb, pc)
1198                    }
1199                };
1200                // Only a leading-unbound scan can be resumed mid-tile: with `pa`
1201                // bound there is one a-group and `scan_from` always restarts it
1202                // from the top, so cutting inside it would duplicate rows.
1203                let splittable = pa.is_none();
1204                let mut last_a: Option<u32> = None;
1205                for abc in rows {
1206                    // Full, and this row opens a new a-group ⇒ cut here and
1207                    // resume at this group. The row is dropped, not lost: the
1208                    // next call re-enters at exactly this id.
1209                    if splittable && out.len() >= max_rows && last_a != Some(abc.0) {
1210                        resume_at = Some(abc.0);
1211                        break;
1212                    }
1213                    last_a = Some(abc.0);
1214                    out.push(perm.back(abc));
1215                }
1216            }
1217            if let Some(a) = resume_at {
1218                out.sort_unstable();
1219                return (out, ((ti as u64) << 32) | a as u64, false);
1220            }
1221            ti += 1;
1222            from_a = 0;
1223            if out.len() >= max_rows {
1224                break;
1225            }
1226        }
1227        let done = ti >= end;
1228        out.sort_unstable();
1229        (out, (ti as u64) << 32, done)
1230    }
1231
1232    /// Stream `pattern`'s matches **sorted on the canonical column `sort_col`**
1233    /// (0=subject, 1=predicate, 2=object): the stream's `sort_col` values are
1234    /// ascending. Chooses a permutation that routes on the bound prefix *and*
1235    /// orders by `sort_col` ([`permutation_sorted_on`](Self::permutation_sorted_on));
1236    /// `None` when none qualifies (e.g. `sort_col` is itself bound). The
1237    /// precondition for feeding a [merge join](crate::bgp): both inputs sorted on
1238    /// the shared join key.
1239    pub(crate) fn scan_iter_sorted_on(
1240        &self,
1241        pattern: Pattern,
1242        sort_col: usize,
1243    ) -> Option<impl Iterator<Item = Triple> + '_> {
1244        Some(self.scan_iter_with(
1245            pattern,
1246            Self::permutation_sorted_on_in(self.perms, pattern, sort_col)?,
1247        ))
1248    }
1249
1250    /// Stream `pattern`'s matches using a **specific** permutation `perm`; the
1251    /// stream is sorted in `perm`'s `(a, b, c)` order. Shared core of `scan_iter`
1252    /// and `scan_iter_sorted_on`.
1253    fn scan_iter_with(
1254        &self,
1255        pattern: Pattern,
1256        perm: IndexPermutation,
1257    ) -> impl Iterator<Item = Triple> + '_ {
1258        // Route: a bound leading id binary-searches the tile directory to exactly
1259        // one tile (groups are never split across tiles); an unbound one chains
1260        // every tile's cursor. Within a tile, a bound leading scan jumps to its
1261        // a-group via the tile's lazily-built group directory — built on first
1262        // use, costing one walk of that (budget-sized) tile.
1263        let [pa, pb, pc] = perm.order_pattern(pattern);
1264        let si = perm.section_index();
1265        let (start, end) = self.tile_span(si, pa);
1266        // Coalesce tile faults, but ramp the prefetch window geometrically as
1267        // the scan advances rather than fetching the whole span up front: a
1268        // small `LIMIT` (which stops pulling early) then never faults tiles
1269        // past the rows it needs, while a full scan still batches into a
1270        // handful of coalesced reads (4, 8, 16, … tiles). A bound leading scan
1271        // spans a single tile, so the prefetch no-ops and it faults just that
1272        // one tile, unchanged.
1273        let window = std::cell::Cell::new(PREFETCH_WINDOW_START);
1274        (start..end)
1275            // Synopsis pre-fault prune: drop a routed tile the directory proves
1276            // can't match a bound secondary component, **without fetching it**
1277            // (the remote win — a negative/sparse lookup costs zero tile reads).
1278            // A bound leading id routes to a single tile, so this is where it
1279            // bites; a fully-unbound leading scan leaves `pb`/`pc` unbound, so
1280            // `syn_admits` keeps every tile, unchanged.
1281            .filter(move |&ti| self.sections[si][ti].syn_admits(pb, pc))
1282            .flat_map(move |ti| {
1283                // Fault in (if remote), parse (untrusted bytes ⇒ `None` on
1284                // malformed), and zone-prune per tile, then stream the
1285                // matching groups.
1286                if !self.tile_resident(si, ti) {
1287                    let w = window.get();
1288                    self.prefetch_span(si, ti, (ti + w).min(end));
1289                    window.set(w.saturating_mul(2).min(PREFETCH_WINDOW_MAX));
1290                }
1291                let tile = &self.sections[si][ti];
1292                // Decode the tile once and materialize its matching rows into an
1293                // owned `Vec`, so the borrowed block/cursor never outlives the
1294                // tile's cache handle (an `Arc` dropped at the end of this
1295                // closure). The buffer is bounded to one tile's rows; the scan
1296                // stays lazy *across* tiles, so a `LIMIT` still stops early.
1297                let rows: Vec<Triple> = match self.tile_entry(si, ti) {
1298                    Some(entry) => TripleBlock::parse(entry.body())
1299                        .ok()
1300                        .filter(|b| b.zone().may_contain(pa, pb, pc))
1301                        .map(|b| match pa {
1302                            Some(a) => {
1303                                let dir = tile.dir.get_or_init(|| b.group_directory());
1304                                b.scan_from(dir, a, pb, pc).collect()
1305                            }
1306                            None => b.scan(pa, pb, pc).collect(),
1307                        })
1308                        .unwrap_or_default(),
1309                    None => Vec::new(),
1310                };
1311                rows.into_iter()
1312            })
1313            .map(move |abc| perm.back(abc))
1314    }
1315
1316    /// The tile index span a scan must visit: every tile when the leading
1317    /// component is unbound, else the run of tiles whose leading-id ranges
1318    /// cover it. Tile ranges are ascending; consecutive tiles may SHARE a
1319    /// leading id when a mega-group was split across tiles (see
1320    /// [`build_tiles`]), so the span is a range scan, not a single hit —
1321    /// files without splits still yield a span of at most one tile.
1322    pub(crate) fn tile_span(&self, section: usize, pa: Option<u32>) -> (usize, usize) {
1323        let tiles = &self.sections[section];
1324        match pa {
1325            None => (0, tiles.len()),
1326            Some(a) => {
1327                let i = tiles.partition_point(|t| t.max_a < a);
1328                let mut j = i;
1329                while j < tiles.len() && tiles[j].min_a <= a {
1330                    j += 1;
1331                }
1332                (i, j)
1333            }
1334        }
1335    }
1336}
1337
1338#[cfg(test)]
1339mod tests {
1340    use super::*;
1341
1342    fn graph() -> (GraphIndex, Vec<Triple>) {
1343        let data = vec![
1344            (1, 10, 100),
1345            (1, 10, 101),
1346            (1, 11, 100),
1347            (2, 10, 100),
1348            (2, 12, 200),
1349            (3, 11, 300),
1350        ];
1351        let mut b = GraphIndexBuilder::new();
1352        for &t in &data {
1353            b.push(t);
1354        }
1355        (b.build(), data)
1356    }
1357
1358    /// Brute-force reference for a pattern.
1359    fn reference(data: &[Triple], (s, p, o): Pattern) -> Vec<Triple> {
1360        let mut v: Vec<Triple> = data
1361            .iter()
1362            .copied()
1363            .filter(|&(a, b, c)| {
1364                s.is_none_or(|x| x == a) && p.is_none_or(|x| x == b) && o.is_none_or(|x| x == c)
1365            })
1366            .collect();
1367        v.sort_unstable();
1368        v
1369    }
1370
1371    #[test]
1372    fn every_pattern_shape_matches_reference() {
1373        let (idx, data) = graph();
1374        let vals = |opts: &[u32]| {
1375            let mut v: Vec<Option<u32>> = opts.iter().map(|&x| Some(x)).collect();
1376            v.push(None);
1377            v
1378        };
1379        // Exercise all 8 bound/unbound shapes across representative values.
1380        for s in vals(&[1, 2, 9]) {
1381            for p in vals(&[10, 11, 99]) {
1382                for o in vals(&[100, 300, 999]) {
1383                    let pat = (s, p, o);
1384                    assert_eq!(
1385                        idx.match_pattern(pat),
1386                        reference(&data, pat),
1387                        "pattern {pat:?}"
1388                    );
1389                    // The lazy stream must match the eager result once sorted —
1390                    // same triples, just without the up-front canonical re-sort.
1391                    let mut streamed: Vec<Triple> = idx.scan_iter(pat).collect();
1392                    streamed.sort_unstable();
1393                    assert_eq!(streamed, reference(&data, pat), "scan_iter {pat:?}");
1394                }
1395            }
1396        }
1397    }
1398
1399    #[test]
1400    fn unbound_returns_everything_sorted() {
1401        let (idx, data) = graph();
1402        let mut sorted = data.clone();
1403        sorted.sort_unstable();
1404        assert_eq!(idx.match_pattern((None, None, None)), sorted);
1405    }
1406
1407    /// A tiny tile budget must split sections into many tiles — including
1408    /// MID-GROUP cuts (budget 1/16 makes every a-group oversized) — and every
1409    /// pattern shape must still match the brute-force reference: bound
1410    /// leading ids route to the run of covering tiles, unbound scans chain
1411    /// all tiles.
1412    #[test]
1413    fn multi_tile_sections_match_reference_every_shape() {
1414        // Enough distinct leading ids to split under a tiny budget.
1415        let mut data: Vec<Triple> = Vec::new();
1416        for s in 1..=40u32 {
1417            for p in [10u32, 11] {
1418                for o in [100u32, 100 + s] {
1419                    data.push((s, p, o));
1420                }
1421            }
1422        }
1423        for budget in [1usize, 16, 64, 1 << 20] {
1424            let mut b = GraphIndexBuilder::new().with_tile_budget(budget);
1425            for &t in &data {
1426                b.push(t);
1427            }
1428            let idx = b.build();
1429            let spo_tiles = idx.tile_sections()[0].len();
1430            if budget <= 16 {
1431                assert!(spo_tiles > 1, "budget {budget} should force tiling");
1432            }
1433            // Tile ranges must be ascending; a split mega-group may leave
1434            // consecutive tiles SHARING a leading id (never overlapping past it).
1435            for w in idx.tile_sections()[0].windows(2) {
1436                assert!(w[0].leading_range().1 <= w[1].leading_range().0);
1437            }
1438            assert_eq!(idx.triple_count() as usize, data.len(), "budget {budget}");
1439
1440            let vals = |opts: &[u32]| {
1441                let mut v: Vec<Option<u32>> = opts.iter().map(|&x| Some(x)).collect();
1442                v.push(None);
1443                v
1444            };
1445            for _round in 0..2 {
1446                for s in vals(&[1, 20, 40, 99]) {
1447                    for p in vals(&[10, 11, 99]) {
1448                        for o in vals(&[100, 120, 999]) {
1449                            let pat = (s, p, o);
1450                            assert_eq!(
1451                                idx.match_pattern(pat),
1452                                reference(&data, pat),
1453                                "budget {budget} pattern {pat:?}"
1454                            );
1455                        }
1456                    }
1457                }
1458            }
1459        }
1460    }
1461
1462    /// The Crossref shape: ONE mega-predicate carrying most of the graph. At
1463    /// the real 64 KiB budget its P-leading a-group must be split across
1464    /// multiple tiles (bounding builder memory), and a bound leading id must
1465    /// route to the whole run of covering tiles with complete results.
1466    #[test]
1467    fn mega_group_splits_across_tiles_and_lookups_stay_complete() {
1468        let hot_p = 7u32;
1469        let mut data: Vec<Triple> = Vec::new();
1470        for i in 0..40_000u32 {
1471            data.push((1_000 + i % 200, hot_p, 50_000 + i));
1472        }
1473        data.push((1, 1, 1));
1474        data.push((2, 2, 2));
1475        let mut b = GraphIndexBuilder::new(); // default INDEX_TILE_BUDGET
1476        for &t in &data {
1477            b.push(t);
1478        }
1479        let idx = b.build();
1480
1481        // PSO (a = predicate): the hot_p group must span several tiles.
1482        let pso = ALL_PERMS
1483            .iter()
1484            .position(|p| matches!(p, IndexPermutation::Pso))
1485            .unwrap();
1486        let covering = idx.tile_sections()[pso]
1487            .iter()
1488            .filter(|t| {
1489                let (lo, hi) = t.leading_range();
1490                lo <= hot_p && hot_p <= hi
1491            })
1492            .count();
1493        assert!(
1494            covering > 1,
1495            "expected the hot predicate split across tiles, got {covering}"
1496        );
1497
1498        // Bound-p lookup must still return every triple of the mega-group —
1499        // probing bound objects in the FIRST, MIDDLE and LAST slices of the
1500        // split run (a mid-run object once returned 0 on the first split file).
1501        for pat in [
1502            (None, Some(hot_p), None),
1503            (Some(1_050), Some(hot_p), None),
1504            (None, Some(hot_p), Some(50_123)), // first slice
1505            (None, Some(hot_p), Some(70_000)), // middle of the run
1506            (None, Some(hot_p), Some(89_999)), // last slice
1507            (None, Some(hot_p), Some(49_000)), // below the range → empty
1508            (None, Some(hot_p), Some(95_000)), // above the range → empty
1509        ] {
1510            assert_eq!(idx.match_pattern(pat), reference(&data, pat), "{pat:?}");
1511        }
1512    }
1513
1514    /// Scans must yield identical results across the directory lifecycle: the
1515    /// first leading-bound scan walks linearly, the second builds the
1516    /// directory, and later ones jump through it — including absent ids
1517    /// (between, below, and above the stored groups).
1518    #[test]
1519    fn directory_backed_scans_match_reference_every_shape() {
1520        let (idx, data) = graph();
1521        let vals = |opts: &[u32]| {
1522            let mut v: Vec<Option<u32>> = opts.iter().map(|&x| Some(x)).collect();
1523            v.push(None);
1524            v
1525        };
1526        for round in 0..3 {
1527            for s in vals(&[1, 2, 3, 0, 9]) {
1528                for p in vals(&[10, 11, 12, 99]) {
1529                    for o in vals(&[100, 200, 300, 999]) {
1530                        let pat = (s, p, o);
1531                        let mut got: Vec<Triple> = idx.scan_iter(pat).collect();
1532                        got.sort_unstable();
1533                        assert_eq!(got, reference(&data, pat), "round {round} scan {pat:?}");
1534                    }
1535                }
1536            }
1537        }
1538    }
1539
1540    /// A tile synopsis must let a routed scan **skip the fetch** of a tile its
1541    /// secondary-column range rules out — and never skip one that could match.
1542    #[test]
1543    fn synopsis_prunes_the_routed_tile_before_fetch() {
1544        use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
1545        use std::sync::Arc;
1546
1547        // One real SPO tile: subject a=5, predicates b∈{10,11}, object c=100.
1548        let block = {
1549            let mut b = TripleBlockBuilder::new();
1550            b.push((5, 10, 100));
1551            b.push((5, 11, 100));
1552            b.build()
1553        };
1554        let fetches = Arc::new(AtomicUsize::new(0));
1555        let (blk, fc) = (block.clone(), fetches.clone());
1556        // The loader returns the (already-decompressed) tile image and counts calls.
1557        let loader: TileLoader = Box::new(move |_si, _ti| {
1558            fc.fetch_add(1, SeqCst);
1559            Some(blk.clone())
1560        });
1561        // Remote SPO directory: one tile, leading a∈[5,5], synopsis b∈[10,11], c∈[100,100].
1562        let dirs = [
1563            vec![(5u32, 5u32, Some((10u32, 11u32, 100u32, 100u32)))],
1564            Vec::new(),
1565            Vec::new(),
1566            Vec::new(),
1567            Vec::new(),
1568            Vec::new(),
1569        ];
1570        let idx = GraphIndex::from_remote_directories(dirs, PermSet::ALL, loader);
1571
1572        // Predicate 99 is outside the tile's b-range → prune, zero fetches, empty.
1573        assert!(idx.match_pattern((Some(5), Some(99), None)).is_empty());
1574        assert_eq!(fetches.load(SeqCst), 0, "synopsis must skip the fetch");
1575
1576        // Object 999 outside the tile's c-range → prune, still zero fetches.
1577        assert!(idx.match_pattern((Some(5), None, Some(999))).is_empty());
1578        assert_eq!(
1579            fetches.load(SeqCst),
1580            0,
1581            "secondary-c prune also skips the fetch"
1582        );
1583
1584        // Predicate 10 is in range → the tile is fetched and the match returned.
1585        assert_eq!(
1586            idx.match_pattern((Some(5), Some(10), None)),
1587            vec![(5, 10, 100)]
1588        );
1589        assert_eq!(
1590            fetches.load(SeqCst),
1591            1,
1592            "an admissible secondary still fetches"
1593        );
1594    }
1595
1596    /// Without a synopsis (`None`), nothing is pruned early — the tile is always
1597    /// fetched and the in-tile zone map does the (correct) pruning, as before.
1598    #[test]
1599    fn absent_synopsis_never_prunes() {
1600        use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
1601        use std::sync::Arc;
1602        let block = {
1603            let mut b = TripleBlockBuilder::new();
1604            b.push((5, 10, 100));
1605            b.build()
1606        };
1607        let fetches = Arc::new(AtomicUsize::new(0));
1608        let (blk, fc) = (block.clone(), fetches.clone());
1609        let loader: TileLoader = Box::new(move |_si, _ti| {
1610            fc.fetch_add(1, SeqCst);
1611            Some(blk.clone())
1612        });
1613        let dirs = [
1614            vec![(5u32, 5u32, None)],
1615            Vec::new(),
1616            Vec::new(),
1617            Vec::new(),
1618            Vec::new(),
1619            Vec::new(),
1620        ];
1621        let idx = GraphIndex::from_remote_directories(dirs, PermSet::ALL, loader);
1622        // Predicate 99 absent, but with no synopsis the tile is fetched (then the
1623        // zone map yields no match) — correctness preserved, just no fetch saved.
1624        assert!(idx.match_pattern((Some(5), Some(99), None)).is_empty());
1625        assert_eq!(fetches.load(SeqCst), 1, "no synopsis ⇒ no early prune");
1626    }
1627
1628    /// The eight triple-pattern shapes, as `(s, p, o)` boundness.
1629    fn all_patterns() -> Vec<Pattern> {
1630        (0..8u8)
1631            .map(|mask| {
1632                (
1633                    (mask & 1 != 0).then_some(1u32),
1634                    (mask & 2 != 0).then_some(2u32),
1635                    (mask & 4 != 0).then_some(3u32),
1636                )
1637            })
1638            .collect()
1639    }
1640
1641    /// The claim the whole option rests on: **{SPO, POS, OSP} ties the longest
1642    /// bound prefix on all eight pattern shapes**, so the other three never win
1643    /// routing — only sort order. Enumerated and printed rather than asserted in
1644    /// prose, because if it were ever false a lean file would answer a pattern
1645    /// from a worse index without anything failing.
1646    #[test]
1647    fn perm_routing_never_leaves_core() {
1648        for pat in all_patterns() {
1649            let best_all = ALL_PERMS
1650                .iter()
1651                .map(|p| p.leading_bound(pat))
1652                .max()
1653                .unwrap();
1654            let best_core = PermSet::CORE
1655                .iter()
1656                .map(|p| p.leading_bound(pat))
1657                .max()
1658                .unwrap();
1659            let scores: Vec<String> = ALL_PERMS
1660                .iter()
1661                .map(|p| format!("{}={}", p.name(), p.leading_bound(pat)))
1662                .collect();
1663            eprintln!(
1664                "s={} p={} o={} | {} | best6={best_all} best3={best_core} chosen={}",
1665                pat.0.is_some() as u8,
1666                pat.1.is_some() as u8,
1667                pat.2.is_some() as u8,
1668                scores.join(" "),
1669                GraphIndex::best_permutation(pat).name(),
1670            );
1671            assert_eq!(
1672                best_core, best_all,
1673                "the three routing permutations must tie the best of six on {pat:?}"
1674            );
1675            let chosen = GraphIndex::best_permutation(pat);
1676            assert!(
1677                PermSet::CORE.contains(chosen),
1678                "best_permutation chose {} for {pat:?}, outside the routing three",
1679                chosen.name()
1680            );
1681            assert_eq!(
1682                chosen,
1683                GraphIndex::best_permutation_in(PermSet::CORE, pat),
1684                "routing must be identical with three permutations and with six"
1685            );
1686        }
1687    }
1688
1689    /// Restricting the set may never make a merge join pick a *worse-routed*
1690    /// stream than the plain lookup would use: with six the co-sorted
1691    /// permutation always ties the best prefix, and with three the guard
1692    /// declines rather than buying sort order with a whole-section scan.
1693    #[test]
1694    fn sorted_permutation_never_sacrifices_routing() {
1695        let mut declined = Vec::new();
1696        for pat in all_patterns() {
1697            for col in 0..3usize {
1698                let routed_all = GraphIndex::best_permutation(pat).leading_bound(pat);
1699                if let Some(p) = GraphIndex::permutation_sorted_on(pat, col) {
1700                    assert_eq!(
1701                        p.leading_bound(pat),
1702                        routed_all,
1703                        "six permutations must always co-sort at the best prefix"
1704                    );
1705                    assert_eq!(p.roles()[p.leading_bound(pat)], col);
1706                }
1707                match GraphIndex::permutation_sorted_on_in(PermSet::CORE, pat, col) {
1708                    Some(p) => {
1709                        assert!(PermSet::CORE.contains(p));
1710                        assert_eq!(
1711                            p.leading_bound(pat),
1712                            GraphIndex::best_permutation_in(PermSet::CORE, pat).leading_bound(pat)
1713                        );
1714                    }
1715                    None => {
1716                        if GraphIndex::permutation_sorted_on(pat, col).is_some() {
1717                            declined.push((pat, col));
1718                        }
1719                    }
1720                }
1721            }
1722        }
1723        // Exactly the three shapes SOP / PSO / OPS exist for: one bound
1724        // component, sorted on a column no remaining order can lead with after
1725        // it — subject-bound sorted on object, predicate-bound sorted on
1726        // subject, object-bound sorted on predicate.
1727        for (pat, col) in &declined {
1728            let bound: Vec<usize> = (0..3)
1729                .filter(|&i| [pat.0, pat.1, pat.2][i].is_some())
1730                .collect();
1731            assert_eq!(
1732                bound.len(),
1733                1,
1734                "only single-bound shapes lose a merge order"
1735            );
1736            assert_ne!(bound[0], *col);
1737        }
1738        eprintln!("three-permutation merge-join declines: {declined:?}");
1739        assert_eq!(declined.len(), 3, "exactly three (bound, sort) shapes lost");
1740    }
1741
1742    #[test]
1743    fn perm_set_masks_and_positions() {
1744        assert_eq!(PermSet::ALL.len(), 6);
1745        assert_eq!(PermSet::CORE.len(), 3);
1746        assert_eq!(PermSet::CORE.bits(), 0b0000_0111);
1747        assert_eq!(PermSet::CORE.names(), vec!["SPO", "POS", "OSP"]);
1748        assert_eq!(
1749            PermSet::ALL.names(),
1750            vec!["SPO", "POS", "OSP", "SOP", "PSO", "OPS"]
1751        );
1752        // Container position is rank among the STORED permutations, which is
1753        // why the routing three had to lead ALL_PERMS.
1754        for (i, p) in PermSet::CORE.iter().enumerate() {
1755            assert_eq!(PermSet::CORE.position(p), Some(i));
1756            assert_eq!(p.section_index(), i);
1757        }
1758        assert_eq!(PermSet::CORE.position(IndexPermutation::Sop), None);
1759        assert_eq!(PermSet::ALL.position(IndexPermutation::Ops), Some(5));
1760        // Only supersets of the routing three are legal.
1761        assert!(PermSet::from_bits(0b0000_0111).is_ok());
1762        assert!(PermSet::from_bits(0b0011_1111).is_ok());
1763        assert!(PermSet::from_bits(0b0000_1111).is_ok());
1764        assert!(PermSet::from_bits(0b0000_0011).is_err());
1765        assert!(PermSet::from_bits(0b0011_1000).is_err());
1766        assert!(PermSet::from_bits(0b1000_0111).is_err());
1767    }
1768
1769    /// A lean index answers every pattern shape with exactly the rows a full
1770    /// one does — the property that makes the option safe at all.
1771    #[test]
1772    fn three_and_six_agree_on_every_pattern() {
1773        let triples: Vec<Triple> = (0..200u32)
1774            .map(|i| (i % 17 + 1, i % 5 + 1, i % 23 + 1))
1775            .collect();
1776        let six = GraphIndexBuilder::from_triples(triples.clone())
1777            .with_tile_budget(128)
1778            .build();
1779        let three = GraphIndexBuilder::from_triples(triples)
1780            .with_tile_budget(128)
1781            .with_perms(PermSet::CORE)
1782            .build();
1783        assert_eq!(six.perms(), PermSet::ALL);
1784        assert_eq!(three.perms(), PermSet::CORE);
1785        for perm in [
1786            IndexPermutation::Sop,
1787            IndexPermutation::Pso,
1788            IndexPermutation::Ops,
1789        ] {
1790            assert!(three.tile_sections()[perm.section_index()].is_empty());
1791        }
1792        for a in [None, Some(3u32)] {
1793            for b in [None, Some(2u32)] {
1794                for c in [None, Some(9u32)] {
1795                    let mut x = six.match_pattern((a, b, c));
1796                    let mut y = three.match_pattern((a, b, c));
1797                    x.sort_unstable();
1798                    y.sort_unstable();
1799                    assert_eq!(x, y, "pattern {:?} disagreed", (a, b, c));
1800                }
1801            }
1802        }
1803    }
1804}