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