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