use std::sync::{Arc, OnceLock};
use crate::chunk_cache::{CacheEntry, ChunkCache};
use crate::triples::{GroupDirectory, Triple, TripleBlock, TripleBlockBuilder};
pub type Pattern = (Option<u32>, Option<u32>, Option<u32>);
pub const INDEX_TILE_BUDGET: usize = 64 * 1024;
const PREFETCH_WINDOW_START: usize = 4;
const PREFETCH_WINDOW_MAX: usize = 512;
pub type TileLoader = Box<dyn Fn(usize, usize) -> Option<Vec<u8>> + Send + Sync>;
pub type TileBulkLoader = Box<dyn Fn(usize, &[usize]) -> Option<Vec<Vec<u8>>> + Send + Sync>;
pub struct Tile {
min_a: u32,
max_a: u32,
pub(crate) syn: Option<(u32, u32, u32, u32)>,
len: u32,
staged: Option<Vec<u8>>,
dir: OnceLock<GroupDirectory>,
}
impl Tile {
fn local(min_a: u32, max_a: u32, bytes: Vec<u8>) -> Self {
let len = bytes.len().min(u32::MAX as usize) as u32;
Tile {
min_a,
max_a,
syn: None,
len,
staged: Some(bytes),
dir: OnceLock::new(),
}
}
fn remote(min_a: u32, max_a: u32, syn: Option<(u32, u32, u32, u32)>) -> Self {
Tile {
min_a,
max_a,
syn,
len: 0,
staged: None,
dir: OnceLock::new(),
}
}
pub(crate) fn encoded_len(&self) -> u64 {
if self.len > 0 {
self.len as u64
} else {
self.staged.as_ref().map_or(0, |d| d.len() as u64)
}
}
pub fn leading_range(&self) -> (u32, u32) {
(self.min_a, self.max_a)
}
pub(crate) fn syn_admits(&self, pb: Option<u32>, pc: Option<u32>) -> bool {
match self.syn {
None => true,
Some((min_b, max_b, min_c, max_c)) => {
let ok = |v: Option<u32>, lo: u32, hi: u32| v.is_none_or(|x| lo <= x && x <= hi);
ok(pb, min_b, max_b) && ok(pc, min_c, max_c)
}
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ScanPlan {
pub permutation: IndexPermutation,
pub tiles_total: usize,
pub tiles_routed: usize,
pub tiles_admitted: usize,
pub tile_bytes: u64,
pub section_bytes: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum IndexPermutation {
Spo,
Sop,
Pso,
Pos,
Osp,
Ops,
}
pub(crate) const NUM_PERMS: usize = 6;
pub(crate) const ALL_PERMS: [IndexPermutation; NUM_PERMS] = [
IndexPermutation::Spo,
IndexPermutation::Pos,
IndexPermutation::Osp,
IndexPermutation::Sop,
IndexPermutation::Pso,
IndexPermutation::Ops,
];
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct PermSet(u8);
impl Default for PermSet {
fn default() -> Self {
PermSet::ALL
}
}
impl PermSet {
pub const ALL: PermSet = PermSet(0b0011_1111);
pub const CORE: PermSet = PermSet(0b0000_0111);
pub fn from_bits(bits: u8) -> Result<PermSet, &'static str> {
if bits & !PermSet::ALL.0 != 0 {
return Err("permutation mask addresses more than six permutations");
}
if bits & PermSet::CORE.0 != PermSet::CORE.0 {
return Err("permutation mask must contain SPO, POS and OSP");
}
Ok(PermSet(bits))
}
pub fn bits(self) -> u8 {
self.0
}
pub fn contains(self, perm: IndexPermutation) -> bool {
self.0 & (1 << perm.section_index()) != 0
}
pub fn len(self) -> usize {
self.0.count_ones() as usize
}
pub fn is_empty(self) -> bool {
false
}
pub fn iter(self) -> impl Iterator<Item = IndexPermutation> {
ALL_PERMS.into_iter().filter(move |p| self.contains(*p))
}
pub fn position(self, perm: IndexPermutation) -> Option<usize> {
self.contains(perm)
.then(|| (self.0 & ((1u8 << perm.section_index()) - 1)).count_ones() as usize)
}
pub fn names(self) -> Vec<&'static str> {
self.iter().map(|p| p.name()).collect()
}
pub fn has_merge_orders(self) -> bool {
self == PermSet::ALL
}
}
impl IndexPermutation {
pub fn name(self) -> &'static str {
match self {
IndexPermutation::Spo => "SPO",
IndexPermutation::Sop => "SOP",
IndexPermutation::Pso => "PSO",
IndexPermutation::Pos => "POS",
IndexPermutation::Osp => "OSP",
IndexPermutation::Ops => "OPS",
}
}
pub fn section_index(self) -> usize {
match self {
IndexPermutation::Spo => 0,
IndexPermutation::Pos => 1,
IndexPermutation::Osp => 2,
IndexPermutation::Sop => 3,
IndexPermutation::Pso => 4,
IndexPermutation::Ops => 5,
}
}
pub(crate) const fn roles(self) -> [usize; 3] {
match self {
IndexPermutation::Spo => [0, 1, 2],
IndexPermutation::Sop => [0, 2, 1],
IndexPermutation::Pso => [1, 0, 2],
IndexPermutation::Pos => [1, 2, 0],
IndexPermutation::Osp => [2, 0, 1],
IndexPermutation::Ops => [2, 1, 0],
}
}
pub(crate) fn forward(self, t: Triple) -> Triple {
let c = [t.0, t.1, t.2];
let r = self.roles();
(c[r[0]], c[r[1]], c[r[2]])
}
fn back(self, abc: Triple) -> Triple {
let a = [abc.0, abc.1, abc.2];
let r = self.roles();
let mut out = [0u32; 3];
out[r[0]] = a[0];
out[r[1]] = a[1];
out[r[2]] = a[2];
(out[0], out[1], out[2])
}
pub(crate) fn order_pattern(self, p: Pattern) -> [Option<u32>; 3] {
let c = [p.0, p.1, p.2];
let r = self.roles();
[c[r[0]], c[r[1]], c[r[2]]]
}
fn leading_bound(self, pat: Pattern) -> usize {
self.order_pattern(pat)
.iter()
.take_while(|c| c.is_some())
.count()
}
}
pub struct GraphIndexBuilder {
triples: Vec<Triple>,
tile_budget: usize,
perms: PermSet,
}
impl Default for GraphIndexBuilder {
fn default() -> Self {
Self {
triples: Vec::new(),
tile_budget: INDEX_TILE_BUDGET,
perms: PermSet::ALL,
}
}
}
impl GraphIndexBuilder {
pub fn new() -> Self {
Self::default()
}
pub fn from_triples(triples: Vec<Triple>) -> Self {
Self {
triples,
tile_budget: INDEX_TILE_BUDGET,
perms: PermSet::ALL,
}
}
pub fn with_tile_budget(mut self, bytes: usize) -> Self {
self.tile_budget = bytes.max(1);
self
}
pub fn with_perms(mut self, perms: PermSet) -> Self {
self.perms = perms;
self
}
pub fn push(&mut self, t: Triple) {
self.triples.push(t);
}
pub fn build_seq(self) -> GraphIndex {
let triples = &self.triples;
let budget = self.tile_budget;
let wanted: Vec<IndexPermutation> = self.perms.iter().collect();
let build_one = |perm: IndexPermutation| -> Vec<Tile> {
#[cfg(feature = "parallel")]
let permuted: Vec<Triple> = {
use rayon::prelude::*;
triples.par_iter().map(|&t| perm.forward(t)).collect()
};
#[cfg(not(feature = "parallel"))]
let permuted: Vec<Triple> = triples.iter().map(|&t| perm.forward(t)).collect();
build_tiles(permuted, budget)
};
#[cfg(feature = "parallel")]
let built: Vec<Vec<Tile>> = {
use rayon::prelude::*;
let mut built: Vec<Vec<Tile>> = Vec::with_capacity(wanted.len());
for chunk in wanted.chunks(2) {
built.extend(
chunk
.to_vec()
.into_par_iter()
.map(build_one)
.collect::<Vec<_>>(),
);
}
built
};
#[cfg(not(feature = "parallel"))]
let built: Vec<Vec<Tile>> = wanted.iter().copied().map(build_one).collect();
GraphIndex::from_sections(scatter(&wanted, built), self.perms)
}
pub fn build(self) -> GraphIndex {
let wanted: Vec<IndexPermutation> = self.perms.iter().collect();
let triples = &self.triples;
let budget = self.tile_budget;
let build_one = move |perm: IndexPermutation| -> Vec<Tile> {
let permuted: Vec<Triple> = triples.iter().map(|&t| perm.forward(t)).collect();
build_tiles(permuted, budget)
};
#[cfg(feature = "parallel")]
let built: Vec<Vec<Tile>> = {
use rayon::iter::{IntoParallelIterator, ParallelIterator};
wanted.clone().into_par_iter().map(build_one).collect()
};
#[cfg(not(feature = "parallel"))]
let built: Vec<Vec<Tile>> = wanted.iter().copied().map(build_one).collect();
GraphIndex::from_sections(scatter(&wanted, built), self.perms)
}
}
fn scatter(wanted: &[IndexPermutation], built: Vec<Vec<Tile>>) -> [Vec<Tile>; NUM_PERMS] {
let mut sections: [Vec<Tile>; NUM_PERMS] = Default::default();
for (perm, tiles) in wanted.iter().zip(built) {
sections[perm.section_index()] = tiles;
}
sections
}
fn varint_len(mut v: u64) -> usize {
let mut n = 1;
while v >= 0x80 {
v >>= 7;
n += 1;
}
n
}
pub(crate) struct GroupSizer {
size: usize,
num_b: u64,
cur_b: u32,
num_c: u64,
prev_c: u32,
empty: bool,
}
impl GroupSizer {
pub(crate) fn start(a: u32, prev_a: u32) -> Self {
GroupSizer {
size: varint_len((a - prev_a) as u64),
num_b: 0,
cur_b: 0,
num_c: 0,
prev_c: 0,
empty: true,
}
}
pub(crate) fn push(&mut self, b: u32, c: u32) -> usize {
if self.empty || b != self.cur_b {
if self.empty {
self.size += varint_len(b as u64); } else {
self.size += varint_len(self.num_c); self.size += varint_len((b - self.cur_b) as u64);
}
self.cur_b = b;
self.num_c = 0;
self.prev_c = 0;
self.num_b += 1;
self.empty = false;
}
self.size += varint_len((c - self.prev_c) as u64);
self.prev_c = c;
self.num_c += 1;
self.total()
}
pub(crate) fn total(&self) -> usize {
self.size
+ if self.empty {
0
} else {
varint_len(self.num_c)
}
+ varint_len(self.num_b)
}
}
fn build_tiles(mut triples: Vec<Triple>, budget: usize) -> Vec<Tile> {
#[cfg(feature = "parallel")]
{
use rayon::slice::ParallelSliceMut;
triples.par_sort_unstable();
}
#[cfg(not(feature = "parallel"))]
triples.sort_unstable();
triples.dedup();
if triples.is_empty() {
return Vec::new();
}
let make_tile = |run: &[Triple]| -> Tile {
let mut b = TripleBlockBuilder::new();
for &t in run {
b.push(t);
}
Tile::local(run[0].0, run[run.len() - 1].0, b.build())
};
let mut tiles = Vec::new();
let mut tile_start = 0usize;
let mut tile_size = 0usize; let mut prev_a = 0u32;
let mut i = 0usize;
while i < triples.len() {
let a = triples[i].0;
let mut slice_start = i; let mut sizer = GroupSizer::start(a, prev_a);
let mut gtotal = 0usize;
while i < triples.len() && triples[i].0 == a {
gtotal = sizer.push(triples[i].1, triples[i].2);
i += 1;
if gtotal > budget {
tiles.push(make_tile(&triples[tile_start..i]));
tile_start = i;
tile_size = 0;
prev_a = a;
slice_start = i;
sizer = GroupSizer::start(a, a);
gtotal = 0;
}
}
if slice_start == i {
continue; }
if slice_start > tile_start && tile_size + gtotal > budget {
tiles.push(make_tile(&triples[tile_start..slice_start]));
tile_start = slice_start;
tile_size = 0;
}
tile_size += gtotal;
prev_a = a;
}
if tile_start < triples.len() {
tiles.push(make_tile(&triples[tile_start..]));
}
tiles
}
pub struct GraphIndex {
pub(crate) sections: [Vec<Tile>; NUM_PERMS],
perms: PermSet,
loader: Option<TileLoader>,
bulk: Option<TileBulkLoader>,
load_failed: std::sync::atomic::AtomicBool,
read_concurrency: usize,
cache: Arc<ChunkCache>,
}
impl GraphIndex {
fn from_sections(mut sections: [Vec<Tile>; NUM_PERMS], perms: PermSet) -> Self {
let cache = ChunkCache::unlimited_arc();
for (si, tiles) in sections.iter_mut().enumerate() {
for (ti, tile) in tiles.iter_mut().enumerate() {
if let Some(body) = tile.staged.take() {
cache.insert((si as u8, ti as u32), Arc::from(body));
}
}
}
GraphIndex {
sections,
perms,
loader: None,
bulk: None,
load_failed: std::sync::atomic::AtomicBool::new(false),
read_concurrency: 1,
cache,
}
}
pub fn from_tiles(sections: [Vec<(u32, u32, Vec<u8>)>; NUM_PERMS], perms: PermSet) -> Self {
let sections = sections.map(|tiles| {
tiles
.into_iter()
.map(|(min_a, max_a, bytes)| Tile::local(min_a, max_a, bytes))
.collect()
});
Self::from_sections(sections, perms)
}
pub fn perms(&self) -> PermSet {
self.perms
}
pub fn set_cache_cap(&self, cap: u64) {
self.cache.set_cap(cap);
}
pub fn cache_cap(&self) -> u64 {
self.cache.cap()
}
pub fn is_lazy(&self) -> bool {
self.loader.is_some()
}
pub fn cache_stats(&self) -> crate::chunk_cache::CacheStats {
self.cache.stats()
}
#[allow(clippy::type_complexity)]
pub fn from_remote_directories(
directories: [Vec<(u32, u32, Option<(u32, u32, u32, u32)>)>; NUM_PERMS],
perms: PermSet,
loader: TileLoader,
) -> Self {
let sections = directories.map(|dir| {
dir.into_iter()
.map(|(min_a, max_a, syn)| Tile::remote(min_a, max_a, syn))
.collect()
});
GraphIndex {
sections,
perms,
loader: Some(loader),
bulk: None,
load_failed: std::sync::atomic::AtomicBool::new(false),
read_concurrency: 1,
cache: ChunkCache::unlimited_arc(),
}
}
pub fn with_bulk_loader(mut self, bulk: TileBulkLoader) -> Self {
self.bulk = Some(bulk);
self
}
pub(crate) fn set_tile_lens(&mut self, lens: [Vec<u32>; NUM_PERMS]) {
for (section, ls) in self.sections.iter_mut().zip(lens) {
for (tile, l) in section.iter_mut().zip(ls) {
tile.len = l;
}
}
}
pub(crate) fn set_read_concurrency(&mut self, c: usize) {
self.read_concurrency = c.max(1);
}
pub(crate) fn read_concurrency(&self) -> usize {
self.read_concurrency
}
pub fn load_incomplete(&self) -> bool {
self.load_failed.load(std::sync::atomic::Ordering::Relaxed)
}
pub fn reset_load_failure(&self) {
self.load_failed
.store(false, std::sync::atomic::Ordering::Relaxed);
}
pub fn is_remote(&self) -> bool {
self.loader.is_some()
}
fn prefetch_span(&self, section: usize, start: usize, end: usize) {
let missing: Vec<usize> = (start..end)
.filter(|&ti| !self.tile_resident(section, ti))
.collect();
self.bulk_fault(section, &missing);
}
fn tile_resident(&self, section: usize, tile: usize) -> bool {
self.cache.contains((section as u8, tile as u32))
}
fn bulk_fault(&self, section: usize, tiles: &[usize]) {
if tiles.len() < 2 {
return;
}
let Some(bulk) = &self.bulk else { return };
if let Some(images) = bulk(section, tiles) {
if images.len() == tiles.len() {
for (&ti, img) in tiles.iter().zip(images) {
self.cache
.insert((section as u8, ti as u32), Arc::from(img));
}
}
}
}
pub(crate) fn prefetch_probe_tiles(&self, patterns: &[Pattern]) {
if self.bulk.is_none() {
return;
}
let mut want: [std::collections::BTreeSet<usize>; NUM_PERMS] = Default::default();
for &pat in patterns {
let perm = Self::best_permutation(pat);
let [pa, pb, pc] = perm.order_pattern(pat);
let si = perm.section_index();
let (start, end) = self.tile_span(si, pa);
for ti in start..end {
if self.sections[si][ti].syn_admits(pb, pc) && !self.tile_resident(si, ti) {
want[si].insert(ti);
}
}
}
for (si, set) in want.iter().enumerate() {
let tiles: Vec<usize> = set.iter().copied().collect();
self.bulk_fault(si, &tiles);
}
}
fn tile_entry(&self, section: usize, tile: usize) -> Option<Arc<CacheEntry>> {
let key = (section as u8, tile as u32);
if let Some(e) = self.cache.get(key) {
return Some(e);
}
match &self.loader {
Some(load) => match load(section, tile) {
Some(bytes) => Some(self.cache.insert(key, Arc::from(bytes))),
None => {
self.load_failed
.store(true, std::sync::atomic::Ordering::Relaxed);
None
}
},
None => Some(self.cache.insert(key, Arc::from(Vec::new()))),
}
}
pub(crate) fn tile_body(&self, section: usize, tile: usize) -> Arc<[u8]> {
self.tile_entry(section, tile)
.map(|e| e.body_arc())
.unwrap_or_else(|| Arc::from(Vec::new()))
}
pub fn triple_count(&self) -> u32 {
self.prefetch_span(0, 0, self.sections[0].len());
(0..self.sections[0].len())
.filter_map(|ti| {
let entry = self.tile_entry(0, ti)?;
TripleBlock::parse(entry.body())
.ok()
.map(|b| b.zone().count)
})
.sum()
}
pub fn tile_sections(&self) -> [&[Tile]; NUM_PERMS] {
[
&self.sections[0],
&self.sections[1],
&self.sections[2],
&self.sections[3],
&self.sections[4],
&self.sections[5],
]
}
pub fn best_permutation(pattern: Pattern) -> IndexPermutation {
Self::best_permutation_in(PermSet::ALL, pattern)
}
pub fn best_permutation_in(perms: PermSet, pattern: Pattern) -> IndexPermutation {
let mut best = IndexPermutation::Spo;
let mut best_score = best.leading_bound(pattern);
for perm in perms.iter() {
let score = perm.leading_bound(pattern);
if score > best_score {
best = perm;
best_score = score;
}
}
best
}
pub fn permutation_sorted_on(pattern: Pattern, sort_col: usize) -> Option<IndexPermutation> {
Self::permutation_sorted_on_in(PermSet::ALL, pattern, sort_col)
}
pub fn permutation_sorted_on_in(
perms: PermSet,
pattern: Pattern,
sort_col: usize,
) -> Option<IndexPermutation> {
let bound = [
pattern.0.is_some(),
pattern.1.is_some(),
pattern.2.is_some(),
];
if bound[sort_col] {
return None;
}
let routed = Self::best_permutation_in(perms, pattern).leading_bound(pattern);
let mut best: Option<(IndexPermutation, usize)> = None;
for perm in perms.iter() {
let roles = perm.roles();
let lead = perm.leading_bound(pattern);
if lead < 3 && roles[lead] == sort_col && best.map(|(_, s)| lead > s).unwrap_or(true) {
best = Some((perm, lead));
}
}
best.filter(|&(_, lead)| lead >= routed).map(|(p, _)| p)
}
pub fn match_serialized_block(
bytes: &[u8],
permutation: IndexPermutation,
pattern: Pattern,
) -> Vec<Triple> {
let [pa, pb, pc] = permutation.order_pattern(pattern);
let mut out: Vec<Triple> = TripleBlock::parse(bytes)
.ok()
.filter(|b| b.zone().may_contain(pa, pb, pc))
.map(|b| b.scan(pa, pb, pc))
.into_iter()
.flatten()
.map(move |abc| permutation.back(abc))
.collect();
out.sort_unstable();
out
}
pub fn match_pattern(&self, pattern: Pattern) -> Vec<Triple> {
let mut out: Vec<Triple> = self.scan_iter(pattern).collect();
out.sort_unstable();
out
}
pub fn scan_iter(&self, pattern: Pattern) -> impl Iterator<Item = Triple> + '_ {
self.scan_iter_with(pattern, Self::best_permutation(pattern))
}
pub fn scan_plan(&self, pattern: Pattern) -> ScanPlan {
let perm = Self::best_permutation_in(self.perms, pattern);
let [_, pb, pc] = perm.order_pattern(pattern);
let si = perm.section_index();
let tiles = &self.sections[si];
let (start, end) = self.tile_span(si, perm.order_pattern(pattern)[0]);
let mut admitted = 0usize;
let mut tile_bytes = 0u64;
for t in &tiles[start..end] {
if t.syn_admits(pb, pc) {
admitted += 1;
tile_bytes += t.encoded_len();
}
}
ScanPlan {
permutation: perm,
tiles_total: tiles.len(),
tiles_routed: end - start,
tiles_admitted: admitted,
tile_bytes,
section_bytes: tiles.iter().map(|t| t.encoded_len()).sum(),
}
}
pub fn scan_batch(
&self,
pattern: Pattern,
cursor: u64,
max_rows: usize,
) -> (Vec<Triple>, u64, bool) {
let max_rows = max_rows.max(1);
let perm = Self::best_permutation(pattern);
let [pa, pb, pc] = perm.order_pattern(pattern);
let si = perm.section_index();
let (start, end) = self.tile_span(si, pa);
let mut ti = ((cursor >> 32) as usize).max(start);
let mut from_a = cursor as u32;
let mut out: Vec<Triple> = Vec::new();
let mut window = PREFETCH_WINDOW_START;
while ti < end {
if !self.sections[si][ti].syn_admits(pb, pc) {
ti += 1;
from_a = 0;
continue;
}
if !self.tile_resident(si, ti) {
self.prefetch_span(si, ti, (ti + window).min(end));
window = window.saturating_mul(2).min(PREFETCH_WINDOW_MAX);
}
let tile = &self.sections[si][ti];
let entry = self.tile_entry(si, ti);
let mut resume_at: Option<u32> = None;
if let Some(block) = entry
.as_deref()
.and_then(|e| TripleBlock::parse(e.body()).ok())
.filter(|b| b.zone().may_contain(pa, pb, pc))
{
let rows = match pa {
Some(a) => {
let dir = tile.dir.get_or_init(|| block.group_directory());
block.scan_from(dir, a, pb, pc)
}
None if from_a == 0 => block.scan(None, pb, pc),
None => {
let dir = tile.dir.get_or_init(|| block.group_directory());
block.scan_resume(dir, from_a, pb, pc)
}
};
let splittable = pa.is_none();
let mut last_a: Option<u32> = None;
for abc in rows {
if splittable && out.len() >= max_rows && last_a != Some(abc.0) {
resume_at = Some(abc.0);
break;
}
last_a = Some(abc.0);
out.push(perm.back(abc));
}
}
if let Some(a) = resume_at {
out.sort_unstable();
return (out, ((ti as u64) << 32) | a as u64, false);
}
ti += 1;
from_a = 0;
if out.len() >= max_rows {
break;
}
}
let done = ti >= end;
out.sort_unstable();
(out, (ti as u64) << 32, done)
}
pub(crate) fn scan_iter_sorted_on(
&self,
pattern: Pattern,
sort_col: usize,
) -> Option<impl Iterator<Item = Triple> + '_> {
Some(self.scan_iter_with(
pattern,
Self::permutation_sorted_on_in(self.perms, pattern, sort_col)?,
))
}
fn scan_iter_with(
&self,
pattern: Pattern,
perm: IndexPermutation,
) -> impl Iterator<Item = Triple> + '_ {
let [pa, pb, pc] = perm.order_pattern(pattern);
let si = perm.section_index();
let (start, end) = self.tile_span(si, pa);
let window = std::cell::Cell::new(PREFETCH_WINDOW_START);
(start..end)
.filter(move |&ti| self.sections[si][ti].syn_admits(pb, pc))
.flat_map(move |ti| {
if !self.tile_resident(si, ti) {
let w = window.get();
self.prefetch_span(si, ti, (ti + w).min(end));
window.set(w.saturating_mul(2).min(PREFETCH_WINDOW_MAX));
}
let tile = &self.sections[si][ti];
let rows: Vec<Triple> = match self.tile_entry(si, ti) {
Some(entry) => TripleBlock::parse(entry.body())
.ok()
.filter(|b| b.zone().may_contain(pa, pb, pc))
.map(|b| match pa {
Some(a) => {
let dir = tile.dir.get_or_init(|| b.group_directory());
b.scan_from(dir, a, pb, pc).collect()
}
None => b.scan(pa, pb, pc).collect(),
})
.unwrap_or_default(),
None => Vec::new(),
};
rows.into_iter()
})
.map(move |abc| perm.back(abc))
}
pub(crate) fn tile_span(&self, section: usize, pa: Option<u32>) -> (usize, usize) {
let tiles = &self.sections[section];
match pa {
None => (0, tiles.len()),
Some(a) => {
let i = tiles.partition_point(|t| t.max_a < a);
let mut j = i;
while j < tiles.len() && tiles[j].min_a <= a {
j += 1;
}
(i, j)
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn graph() -> (GraphIndex, Vec<Triple>) {
let data = vec![
(1, 10, 100),
(1, 10, 101),
(1, 11, 100),
(2, 10, 100),
(2, 12, 200),
(3, 11, 300),
];
let mut b = GraphIndexBuilder::new();
for &t in &data {
b.push(t);
}
(b.build(), data)
}
fn reference(data: &[Triple], (s, p, o): Pattern) -> Vec<Triple> {
let mut v: Vec<Triple> = data
.iter()
.copied()
.filter(|&(a, b, c)| {
s.is_none_or(|x| x == a) && p.is_none_or(|x| x == b) && o.is_none_or(|x| x == c)
})
.collect();
v.sort_unstable();
v
}
#[test]
fn every_pattern_shape_matches_reference() {
let (idx, data) = graph();
let vals = |opts: &[u32]| {
let mut v: Vec<Option<u32>> = opts.iter().map(|&x| Some(x)).collect();
v.push(None);
v
};
for s in vals(&[1, 2, 9]) {
for p in vals(&[10, 11, 99]) {
for o in vals(&[100, 300, 999]) {
let pat = (s, p, o);
assert_eq!(
idx.match_pattern(pat),
reference(&data, pat),
"pattern {pat:?}"
);
let mut streamed: Vec<Triple> = idx.scan_iter(pat).collect();
streamed.sort_unstable();
assert_eq!(streamed, reference(&data, pat), "scan_iter {pat:?}");
}
}
}
}
#[test]
fn unbound_returns_everything_sorted() {
let (idx, data) = graph();
let mut sorted = data.clone();
sorted.sort_unstable();
assert_eq!(idx.match_pattern((None, None, None)), sorted);
}
#[test]
fn multi_tile_sections_match_reference_every_shape() {
let mut data: Vec<Triple> = Vec::new();
for s in 1..=40u32 {
for p in [10u32, 11] {
for o in [100u32, 100 + s] {
data.push((s, p, o));
}
}
}
for budget in [1usize, 16, 64, 1 << 20] {
let mut b = GraphIndexBuilder::new().with_tile_budget(budget);
for &t in &data {
b.push(t);
}
let idx = b.build();
let spo_tiles = idx.tile_sections()[0].len();
if budget <= 16 {
assert!(spo_tiles > 1, "budget {budget} should force tiling");
}
for w in idx.tile_sections()[0].windows(2) {
assert!(w[0].leading_range().1 <= w[1].leading_range().0);
}
assert_eq!(idx.triple_count() as usize, data.len(), "budget {budget}");
let vals = |opts: &[u32]| {
let mut v: Vec<Option<u32>> = opts.iter().map(|&x| Some(x)).collect();
v.push(None);
v
};
for _round in 0..2 {
for s in vals(&[1, 20, 40, 99]) {
for p in vals(&[10, 11, 99]) {
for o in vals(&[100, 120, 999]) {
let pat = (s, p, o);
assert_eq!(
idx.match_pattern(pat),
reference(&data, pat),
"budget {budget} pattern {pat:?}"
);
}
}
}
}
}
}
#[test]
fn mega_group_splits_across_tiles_and_lookups_stay_complete() {
let hot_p = 7u32;
let mut data: Vec<Triple> = Vec::new();
for i in 0..40_000u32 {
data.push((1_000 + i % 200, hot_p, 50_000 + i));
}
data.push((1, 1, 1));
data.push((2, 2, 2));
let mut b = GraphIndexBuilder::new(); for &t in &data {
b.push(t);
}
let idx = b.build();
let pso = ALL_PERMS
.iter()
.position(|p| matches!(p, IndexPermutation::Pso))
.unwrap();
let covering = idx.tile_sections()[pso]
.iter()
.filter(|t| {
let (lo, hi) = t.leading_range();
lo <= hot_p && hot_p <= hi
})
.count();
assert!(
covering > 1,
"expected the hot predicate split across tiles, got {covering}"
);
for pat in [
(None, Some(hot_p), None),
(Some(1_050), Some(hot_p), None),
(None, Some(hot_p), Some(50_123)), (None, Some(hot_p), Some(70_000)), (None, Some(hot_p), Some(89_999)), (None, Some(hot_p), Some(49_000)), (None, Some(hot_p), Some(95_000)), ] {
assert_eq!(idx.match_pattern(pat), reference(&data, pat), "{pat:?}");
}
}
#[test]
fn directory_backed_scans_match_reference_every_shape() {
let (idx, data) = graph();
let vals = |opts: &[u32]| {
let mut v: Vec<Option<u32>> = opts.iter().map(|&x| Some(x)).collect();
v.push(None);
v
};
for round in 0..3 {
for s in vals(&[1, 2, 3, 0, 9]) {
for p in vals(&[10, 11, 12, 99]) {
for o in vals(&[100, 200, 300, 999]) {
let pat = (s, p, o);
let mut got: Vec<Triple> = idx.scan_iter(pat).collect();
got.sort_unstable();
assert_eq!(got, reference(&data, pat), "round {round} scan {pat:?}");
}
}
}
}
}
#[test]
fn synopsis_prunes_the_routed_tile_before_fetch() {
use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
use std::sync::Arc;
let block = {
let mut b = TripleBlockBuilder::new();
b.push((5, 10, 100));
b.push((5, 11, 100));
b.build()
};
let fetches = Arc::new(AtomicUsize::new(0));
let (blk, fc) = (block.clone(), fetches.clone());
let loader: TileLoader = Box::new(move |_si, _ti| {
fc.fetch_add(1, SeqCst);
Some(blk.clone())
});
let dirs = [
vec![(5u32, 5u32, Some((10u32, 11u32, 100u32, 100u32)))],
Vec::new(),
Vec::new(),
Vec::new(),
Vec::new(),
Vec::new(),
];
let idx = GraphIndex::from_remote_directories(dirs, PermSet::ALL, loader);
assert!(idx.match_pattern((Some(5), Some(99), None)).is_empty());
assert_eq!(fetches.load(SeqCst), 0, "synopsis must skip the fetch");
assert!(idx.match_pattern((Some(5), None, Some(999))).is_empty());
assert_eq!(
fetches.load(SeqCst),
0,
"secondary-c prune also skips the fetch"
);
assert_eq!(
idx.match_pattern((Some(5), Some(10), None)),
vec![(5, 10, 100)]
);
assert_eq!(
fetches.load(SeqCst),
1,
"an admissible secondary still fetches"
);
}
#[test]
fn absent_synopsis_never_prunes() {
use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
use std::sync::Arc;
let block = {
let mut b = TripleBlockBuilder::new();
b.push((5, 10, 100));
b.build()
};
let fetches = Arc::new(AtomicUsize::new(0));
let (blk, fc) = (block.clone(), fetches.clone());
let loader: TileLoader = Box::new(move |_si, _ti| {
fc.fetch_add(1, SeqCst);
Some(blk.clone())
});
let dirs = [
vec![(5u32, 5u32, None)],
Vec::new(),
Vec::new(),
Vec::new(),
Vec::new(),
Vec::new(),
];
let idx = GraphIndex::from_remote_directories(dirs, PermSet::ALL, loader);
assert!(idx.match_pattern((Some(5), Some(99), None)).is_empty());
assert_eq!(fetches.load(SeqCst), 1, "no synopsis ⇒ no early prune");
}
fn all_patterns() -> Vec<Pattern> {
(0..8u8)
.map(|mask| {
(
(mask & 1 != 0).then_some(1u32),
(mask & 2 != 0).then_some(2u32),
(mask & 4 != 0).then_some(3u32),
)
})
.collect()
}
#[test]
fn perm_routing_never_leaves_core() {
for pat in all_patterns() {
let best_all = ALL_PERMS
.iter()
.map(|p| p.leading_bound(pat))
.max()
.unwrap();
let best_core = PermSet::CORE
.iter()
.map(|p| p.leading_bound(pat))
.max()
.unwrap();
let scores: Vec<String> = ALL_PERMS
.iter()
.map(|p| format!("{}={}", p.name(), p.leading_bound(pat)))
.collect();
eprintln!(
"s={} p={} o={} | {} | best6={best_all} best3={best_core} chosen={}",
pat.0.is_some() as u8,
pat.1.is_some() as u8,
pat.2.is_some() as u8,
scores.join(" "),
GraphIndex::best_permutation(pat).name(),
);
assert_eq!(
best_core, best_all,
"the three routing permutations must tie the best of six on {pat:?}"
);
let chosen = GraphIndex::best_permutation(pat);
assert!(
PermSet::CORE.contains(chosen),
"best_permutation chose {} for {pat:?}, outside the routing three",
chosen.name()
);
assert_eq!(
chosen,
GraphIndex::best_permutation_in(PermSet::CORE, pat),
"routing must be identical with three permutations and with six"
);
}
}
#[test]
fn sorted_permutation_never_sacrifices_routing() {
let mut declined = Vec::new();
for pat in all_patterns() {
for col in 0..3usize {
let routed_all = GraphIndex::best_permutation(pat).leading_bound(pat);
if let Some(p) = GraphIndex::permutation_sorted_on(pat, col) {
assert_eq!(
p.leading_bound(pat),
routed_all,
"six permutations must always co-sort at the best prefix"
);
assert_eq!(p.roles()[p.leading_bound(pat)], col);
}
match GraphIndex::permutation_sorted_on_in(PermSet::CORE, pat, col) {
Some(p) => {
assert!(PermSet::CORE.contains(p));
assert_eq!(
p.leading_bound(pat),
GraphIndex::best_permutation_in(PermSet::CORE, pat).leading_bound(pat)
);
}
None => {
if GraphIndex::permutation_sorted_on(pat, col).is_some() {
declined.push((pat, col));
}
}
}
}
}
for (pat, col) in &declined {
let bound: Vec<usize> = (0..3)
.filter(|&i| [pat.0, pat.1, pat.2][i].is_some())
.collect();
assert_eq!(
bound.len(),
1,
"only single-bound shapes lose a merge order"
);
assert_ne!(bound[0], *col);
}
eprintln!("three-permutation merge-join declines: {declined:?}");
assert_eq!(declined.len(), 3, "exactly three (bound, sort) shapes lost");
}
#[test]
fn perm_set_masks_and_positions() {
assert_eq!(PermSet::ALL.len(), 6);
assert_eq!(PermSet::CORE.len(), 3);
assert_eq!(PermSet::CORE.bits(), 0b0000_0111);
assert_eq!(PermSet::CORE.names(), vec!["SPO", "POS", "OSP"]);
assert_eq!(
PermSet::ALL.names(),
vec!["SPO", "POS", "OSP", "SOP", "PSO", "OPS"]
);
for (i, p) in PermSet::CORE.iter().enumerate() {
assert_eq!(PermSet::CORE.position(p), Some(i));
assert_eq!(p.section_index(), i);
}
assert_eq!(PermSet::CORE.position(IndexPermutation::Sop), None);
assert_eq!(PermSet::ALL.position(IndexPermutation::Ops), Some(5));
assert!(PermSet::from_bits(0b0000_0111).is_ok());
assert!(PermSet::from_bits(0b0011_1111).is_ok());
assert!(PermSet::from_bits(0b0000_1111).is_ok());
assert!(PermSet::from_bits(0b0000_0011).is_err());
assert!(PermSet::from_bits(0b0011_1000).is_err());
assert!(PermSet::from_bits(0b1000_0111).is_err());
}
#[test]
fn three_and_six_agree_on_every_pattern() {
let triples: Vec<Triple> = (0..200u32)
.map(|i| (i % 17 + 1, i % 5 + 1, i % 23 + 1))
.collect();
let six = GraphIndexBuilder::from_triples(triples.clone())
.with_tile_budget(128)
.build();
let three = GraphIndexBuilder::from_triples(triples)
.with_tile_budget(128)
.with_perms(PermSet::CORE)
.build();
assert_eq!(six.perms(), PermSet::ALL);
assert_eq!(three.perms(), PermSet::CORE);
for perm in [
IndexPermutation::Sop,
IndexPermutation::Pso,
IndexPermutation::Ops,
] {
assert!(three.tile_sections()[perm.section_index()].is_empty());
}
for a in [None, Some(3u32)] {
for b in [None, Some(2u32)] {
for c in [None, Some(9u32)] {
let mut x = six.match_pattern((a, b, c));
let mut y = three.match_pattern((a, b, c));
x.sort_unstable();
y.sort_unstable();
assert_eq!(x, y, "pattern {:?} disagreed", (a, b, c));
}
}
}
}
}