use crate::dictionary::Dictionary;
use crate::header::{
Header, FLAG_HAS_QUADS, FLAG_HAS_QUOTED_TRIPLES, FLAG_TILE_SYNOPSIS, HEADER_LEN, MAGIC,
};
use crate::index::{GraphIndex, IndexPermutation, Pattern, PermSet, NUM_PERMS};
use crate::meta::{ClassNode, CommunityDescriptor, LevelLinks, LevelRollup, PyramidMeta};
use crate::pyramid::{build_dendrogram, project_graph, PyramidAlgo};
use crate::reader::RangeReader;
use crate::tiling::{choose_round_for_budget, summarize, SuperEdge};
use crate::triples::Triple;
use crate::varint::{read_uvarint, write_uvarint};
pub const DEFAULT_TILE_BUDGET: usize = 64 * 1024;
pub fn build_pyramid_meta(
dict: &Dictionary,
triples: &[(u32, u32, u32)],
budget: usize,
) -> (Vec<u8>, u16) {
build_pyramid_meta_with(dict, triples, budget, None)
}
pub fn build_pyramid_meta_with(
dict: &Dictionary,
triples: &[(u32, u32, u32)],
budget: usize,
type_override: Option<&str>,
) -> (Vec<u8>, u16) {
build_pyramid_meta_algo(dict, triples, budget, type_override, PyramidAlgo::Louvain)
}
pub fn build_pyramid_meta_algo(
dict: &Dictionary,
triples: &[(u32, u32, u32)],
budget: usize,
type_override: Option<&str>,
algo: PyramidAlgo,
) -> (Vec<u8>, u16) {
let timing = std::env::var_os("RETE_BUILD_TIMING").is_some();
let mut t = timing.then(std::time::Instant::now);
let mut lap = |label: &str| {
if let Some(t0) = &mut t {
eprintln!(
" [pyramid] {label}: {:.0} ms",
t0.elapsed().as_secs_f64() * 1000.0
);
*t0 = std::time::Instant::now();
}
};
let louvain = |lap: &mut dyn FnMut(&str)| {
let g = project_graph(dict, triples);
lap("project_graph");
let d = build_dendrogram(&g);
lap("build_dendrogram (Louvain)");
d
};
let dend = match algo {
PyramidAlgo::Louvain => louvain(&mut lap),
PyramidAlgo::Types => {
match crate::schema_pyramid::build_type_dendrogram(dict, triples, type_override) {
Some(d) => {
lap("build_type_dendrogram");
d
}
None => {
eprintln!(
" [pyramid] --pyramid-algo types: no usable rdf:type \
predicate — falling back to louvain"
);
louvain(&mut lap)
}
}
}
};
let round = choose_round_for_budget(dict, triples, &dend, budget);
lap("choose_round_for_budget");
let summary = summarize(dict, triples, &dend, round);
lap("summarize");
let sp = crate::schema_pyramid::build_schema_pyramid_with(
dict,
triples,
&dend,
round,
type_override,
);
lap("build_schema_pyramid");
let predicate_stats = compute_predicate_stats(triples);
lap("compute_predicate_stats");
let char_sets = compute_char_sets(triples);
lap("compute_char_sets");
let label_index = compute_label_index(dict, triples);
lap("compute_label_index");
let meta = PyramidMeta::new(round as u32, summary, &[])
.with_schema(
sp.class_hierarchy,
sp.level_rollups,
sp.level_links,
sp.descriptors,
sp.subclass_cycles,
sp.disjoint_pairs,
sp.equivalent_pairs,
)
.with_predicate_stats(predicate_stats)
.with_char_sets(char_sets)
.with_label_index(label_index);
let out = (meta.encode(), dend.rounds() as u16);
lap("encode");
out
}
const LABEL_PREDICATES: &[&str] = &[
"<http://www.w3.org/2000/01/rdf-schema#label>",
"<http://www.w3.org/2004/02/skos/core#prefLabel>",
"<http://www.w3.org/2004/02/skos/core#altLabel>",
"<http://xmlns.com/foaf/0.1/name>",
"<http://purl.org/dc/terms/title>",
"<http://purl.org/dc/elements/1.1/title>",
"<http://schema.org/name>",
];
fn compute_label_index(
dict: &Dictionary,
triples: &[(u32, u32, u32)],
) -> Vec<crate::meta::LabelEntry> {
use crate::terms::{is_literal, literal_lexical};
use std::collections::{HashMap, HashSet};
const MAX_LABELS: usize = 8192;
let label_pids: HashSet<u32> = LABEL_PREDICATES
.iter()
.filter_map(|p| dict.predicate_id(p))
.collect();
if label_pids.is_empty() {
return Vec::new();
}
let mut degree: HashMap<u32, u32> = HashMap::new();
for &(s, _p, _o) in triples {
*degree.entry(s).or_insert(0) += 1;
}
let mut seen: HashSet<(u32, String)> = HashSet::new();
let mut candidates: Vec<(u32, String, u32)> = Vec::new(); for &(s, p, o) in triples {
if !label_pids.contains(&p) {
continue;
}
let Some(term) = dict.object_term(o) else {
continue;
};
if !is_literal(&term) {
continue;
}
let Some(label) = literal_lexical(&term) else {
continue;
};
if label.is_empty() {
continue;
}
if seen.insert((s, label.to_lowercase())) {
candidates.push((*degree.get(&s).unwrap_or(&0), label, s));
}
}
if candidates.len() > MAX_LABELS {
candidates.sort_by(|a, b| {
b.0.cmp(&a.0)
.then_with(|| a.2.cmp(&b.2))
.then_with(|| a.1.cmp(&b.1))
});
candidates.truncate(MAX_LABELS);
}
candidates.sort_by(|a, b| {
a.1.to_lowercase()
.cmp(&b.1.to_lowercase())
.then_with(|| a.1.cmp(&b.1))
.then_with(|| a.2.cmp(&b.2))
});
candidates
.into_iter()
.map(|(_deg, label, subject)| crate::meta::LabelEntry { label, subject })
.collect()
}
pub(crate) fn compute_text_index(dict: &Dictionary, triples: &[(u32, u32, u32)]) -> Vec<u8> {
use crate::terms::{is_literal, literal_lexical};
let mut b = crate::text_index::TextIndexBuilder::new();
for &(s, _p, o) in triples {
let Some(term) = dict.object_term(o) else {
continue;
};
if !is_literal(&term) {
continue;
}
if let Some(lit) = literal_lexical(&term) {
b.add_text(&lit, s);
}
}
if b.is_empty() {
Vec::new()
} else {
b.build(writer_codec())
}
}
fn compute_char_sets(triples: &[(u32, u32, u32)]) -> Vec<crate::meta::CharSet> {
use std::collections::{BTreeSet, HashMap};
const MAX_CHAR_SETS: usize = 128;
let mut by_subject: HashMap<u32, BTreeSet<u32>> = HashMap::new();
for &(s, p, _o) in triples {
by_subject.entry(s).or_default().insert(p);
}
let mut shapes: HashMap<Vec<u32>, u64> = HashMap::new();
for set in by_subject.into_values() {
*shapes.entry(set.into_iter().collect()).or_insert(0) += 1;
}
let mut v: Vec<crate::meta::CharSet> = shapes
.into_iter()
.map(|(predicates, subjects)| crate::meta::CharSet {
predicates,
subjects,
})
.collect();
v.sort_by(|a, b| {
b.subjects
.cmp(&a.subjects)
.then_with(|| a.predicates.cmp(&b.predicates))
});
v.truncate(MAX_CHAR_SETS);
v
}
fn compute_predicate_stats(triples: &[(u32, u32, u32)]) -> Vec<crate::meta::PredStat> {
use std::collections::HashMap;
#[allow(clippy::type_complexity)]
let mut acc: HashMap<u32, (HashMap<u32, u32>, HashMap<u32, u32>, u64)> = HashMap::new();
for &(s, p, o) in triples {
let e = acc.entry(p).or_default();
*e.0.entry(s).or_insert(0) += 1;
*e.1.entry(o).or_insert(0) += 1;
e.2 += 1;
}
let mut stats: Vec<crate::meta::PredStat> = acc
.into_iter()
.map(|(predicate, (subj, obj, count))| crate::meta::PredStat {
predicate,
count,
distinct_subjects: subj.len() as u64,
distinct_objects: obj.len() as u64,
max_objects_per_subject: subj.values().copied().max().unwrap_or(0),
max_subjects_per_object: obj.values().copied().max().unwrap_or(0),
})
.collect();
stats.sort_by_key(|p| p.predicate);
stats
}
pub const CODEC_NONE: u8 = 0;
pub const CODEC_ZSTD: u8 = 1;
#[cfg(feature = "compression")]
const ZSTD_LEVEL: i32 = 9;
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum FileError {
#[error("header: {0}")]
Header(#[from] crate::header::HeaderError),
#[error("malformed container: {0}")]
Container(&'static str),
#[error("unknown codec: {0}")]
UnknownCodec(u8),
#[error("decompression failed: {0}")]
Decompress(std::io::Error),
#[error("io: {0}")]
Io(#[from] std::io::Error),
}
pub(crate) fn writer_codec() -> u8 {
if cfg!(feature = "compression") {
CODEC_ZSTD
} else {
CODEC_NONE
}
}
fn intersect_sorted(a: &[u32], b: &[u32]) -> Vec<u32> {
let mut out = Vec::with_capacity(a.len().min(b.len()));
let (mut i, mut j) = (0, 0);
while i < a.len() && j < b.len() {
match a[i].cmp(&b[j]) {
std::cmp::Ordering::Less => i += 1,
std::cmp::Ordering::Greater => j += 1,
std::cmp::Ordering::Equal => {
out.push(a[i]);
i += 1;
j += 1;
}
}
}
out
}
pub(crate) fn compress(codec: u8, bytes: &[u8]) -> Vec<u8> {
match codec {
#[cfg(feature = "compression")]
CODEC_ZSTD => {
zstd::encode_all(bytes, ZSTD_LEVEL).expect("zstd encode is infallible in-memory")
}
_ => bytes.to_vec(),
}
}
pub(crate) fn decompress(codec: u8, bytes: &[u8]) -> Result<Vec<u8>, FileError> {
match codec {
CODEC_NONE => Ok(bytes.to_vec()),
#[cfg(feature = "compression")]
CODEC_ZSTD => zstd::decode_all(bytes).map_err(FileError::Decompress),
#[cfg(not(feature = "compression"))]
CODEC_ZSTD => {
use std::io::Read;
let mut dec = ruzstd::StreamingDecoder::new(bytes)
.map_err(|e| FileError::Decompress(std::io::Error::other(e.to_string())))?;
let mut out = Vec::new();
dec.read_to_end(&mut out).map_err(FileError::Decompress)?;
Ok(out)
}
other => Err(FileError::UnknownCodec(other)),
}
}
const TILE_COALESCE_GAP: u64 = 4096;
const DICT_COALESCE_GAP: u64 = 64 * 1024;
pub const DEFAULT_EXPORT_BUDGET_MB: u64 = 4096;
const MIN_DICT_CACHE_CAP: u64 = 64 << 20;
#[derive(Debug, Clone, Copy)]
pub struct MemoryBudget {
pub block_cache: u64,
pub dict_cache: u64,
pub tile_cache: u64,
}
pub fn split_memory_budget(total: u64) -> MemoryBudget {
if total == u64::MAX {
return MemoryBudget {
block_cache: u64::MAX,
dict_cache: u64::MAX,
tile_cache: u64::MAX,
};
}
let block_cache = (total / 8).min(crate::block_cache::DEFAULT_CACHE_CAP);
let tile_cache = (total / 16).min(256 << 20);
let dict_cache = total
.saturating_sub(block_cache)
.saturating_sub(tile_cache)
.max(MIN_DICT_CACHE_CAP);
MemoryBudget {
block_cache,
dict_cache,
tile_cache,
}
}
fn fmt_cap(cap: u64) -> String {
if cap == u64::MAX {
"unlimited".to_string()
} else {
format!("{}MiB", cap >> 20)
}
}
fn read_coalesced<R: RangeReader + ?Sized>(
reader: &R,
ranges: &[ByteRange],
gap: u64,
) -> Option<Vec<Vec<u8>>> {
let mut spans: Vec<(u64, u64)> = Vec::new();
let mut span_of: Vec<usize> = Vec::with_capacity(ranges.len());
let mut i = 0;
while i < ranges.len() {
let start = ranges[i].offset;
let mut end = ranges[i].offset.checked_add(ranges[i].len)?;
let mut j = i + 1;
while j < ranges.len() {
let r = &ranges[j];
if r.offset < end || r.offset - end > gap {
break;
}
end = r.offset.checked_add(r.len)?;
j += 1;
}
let si = spans.len();
spans.push((start, end - start));
for _ in i..j {
span_of.push(si);
}
i = j;
}
let blobs = reader.read_many(&spans).ok()?;
if blobs.len() != spans.len() {
return None;
}
let mut out = Vec::with_capacity(ranges.len());
for (k, r) in ranges.iter().enumerate() {
let (span_start, _) = spans[span_of[k]];
let blob = &blobs[span_of[k]];
let lo = (r.offset - span_start) as usize;
let hi = lo.checked_add(r.len as usize)?;
out.push(blob.get(lo..hi)?.to_vec());
}
Some(out)
}
fn content_hash(parts: &[&[u8]]) -> [u8; 16] {
let mut h = blake3::Hasher::new();
for p in parts {
h.update(p);
}
let mut out = [0u8; 16];
out.copy_from_slice(&h.finalize().as_bytes()[..16]);
out
}
pub type TermTriple = (String, String, String);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DumpPlan {
pub scan: Option<crate::index::ScanPlan>,
pub dictionary_bytes: u64,
}
#[derive(Debug, Clone)]
pub struct LayoutSegment {
pub kind: &'static str,
pub label: String,
pub offset: u64,
pub len: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ByteRange {
pub offset: u64,
pub len: u64,
}
impl ByteRange {
pub fn end(self) -> u64 {
self.offset + self.len
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TripleProvenance {
pub terms: TermTriple,
pub ids: Triple,
pub graph: Option<String>,
pub matched_pattern: Pattern,
pub index_permutation: IndexPermutation,
pub dictionary_range: ByteRange,
pub index_range: ByteRange,
pub index_section_range: ByteRange,
pub pyramid_range: Option<ByteRange>,
pub tile: Option<String>,
pub tile_range: Option<ByteRange>,
}
fn encode_container(sections: &[&[u8]], codec: u8) -> Vec<u8> {
let mut out = Vec::new();
write_uvarint(&mut out, sections.len() as u64);
for s in sections {
let payload = compress(codec, s);
write_uvarint(&mut out, payload.len() as u64);
out.extend_from_slice(&payload);
}
out
}
fn decode_container(bytes: &[u8], codec: u8) -> Result<Vec<Vec<u8>>, FileError> {
let (n, mut pos) = read_uvarint(bytes).ok_or(FileError::Container("truncated count"))?;
let mut out = Vec::with_capacity((n as usize).min(bytes.len()));
for _ in 0..n {
let (len, used) =
read_uvarint(&bytes[pos..]).ok_or(FileError::Container("truncated length"))?;
pos += used;
let end = pos + len as usize;
if end > bytes.len() {
return Err(FileError::Container("section overruns buffer"));
}
out.push(decompress(codec, &bytes[pos..end])?);
pos = end;
}
Ok(out)
}
fn checked_end(off: u64, len: u64) -> Result<u64, FileError> {
off.checked_add(len)
.ok_or(FileError::Container("section range overflows"))
}
const DICT_CHUNK_BUDGET: usize = 64 * 1024;
fn encode_chunked_dict_section(raw: &[u8], codec: u8) -> Vec<u8> {
let meta = crate::dict::parse_meta(raw).unwrap_or(crate::dict::SectionMeta {
term_count: 0,
restart_interval: 1,
restart_offsets: Vec::new(),
});
let body_start = meta
.restart_offsets
.first()
.copied()
.unwrap_or(raw.len() as u64);
let header = &raw[..(body_start.min(raw.len() as u64)) as usize];
let n_runs = meta.restart_offsets.len();
let mut bounds: Vec<(usize, u64, u64)> = Vec::new(); let mut r = 0;
while r < n_runs {
let start = meta.restart_offsets[r];
let mut r2 = r + 1;
while r2 < n_runs && meta.restart_offsets[r2] - start < DICT_CHUNK_BUDGET as u64 {
r2 += 1;
}
let end = if r2 < n_runs {
meta.restart_offsets[r2]
} else {
raw.len() as u64
};
bounds.push((r, start, end));
r = r2;
}
let compressed: Vec<Vec<u8>> = bounds
.iter()
.map(|&(_, s, e)| compress(codec, &raw[s as usize..e as usize]))
.collect();
let mut out = Vec::new();
write_uvarint(&mut out, header.len() as u64);
out.extend_from_slice(header);
write_uvarint(&mut out, bounds.len() as u64);
let mut prev_run = 0usize;
let mut prev_last: Option<Vec<u8>> = None;
for (i, (&(first_run, start, end), comp)) in bounds.iter().zip(&compressed).enumerate() {
let key = if i == 0 {
Vec::new()
} else {
let first = crate::dict::run_first_term(raw, start as usize).unwrap_or_default();
match &prev_last {
None => first,
Some(pl) => crate::dict::shortest_separator(pl, &first),
}
};
if i + 1 < bounds.len() {
let last_run_off = meta.restart_offsets[bounds[i + 1].0 - 1] as usize;
prev_last = crate::dict::run_last_term(raw, last_run_off, end as usize);
}
write_uvarint(&mut out, (first_run - prev_run) as u64);
write_uvarint(&mut out, key.len() as u64);
out.extend_from_slice(&key);
write_uvarint(&mut out, comp.len() as u64);
prev_run = first_run;
}
for comp in &compressed {
out.extend_from_slice(comp);
}
out
}
struct DictChunkEntry {
first_run: usize,
key: Vec<u8>,
body_start: u64,
start: u64,
end: u64,
}
fn parse_chunked_dict_dir(
bytes: &[u8],
total_len: u64,
) -> Result<(crate::dict::SectionMeta, Vec<DictChunkEntry>), FileError> {
let mut pos = 0usize;
let take = |pos: &mut usize| -> Result<u64, FileError> {
let (v, n) = read_uvarint(bytes.get(*pos..).unwrap_or(&[]))
.ok_or(FileError::Container("truncated dict chunk directory"))?;
*pos += n;
Ok(v)
};
let header_len = take(&mut pos)? as usize;
let header = bytes
.get(pos..pos.saturating_add(header_len))
.ok_or(FileError::Container("truncated dict header"))?;
let meta = crate::dict::parse_meta(header)
.map_err(|_| FileError::Container("malformed dict header"))?;
pos += header_len;
let num_chunks = take(&mut pos)? as usize;
let mut entries = Vec::with_capacity(num_chunks.min(bytes.len()));
let mut lens = Vec::with_capacity(num_chunks.min(bytes.len()));
let mut prev_run = 0usize;
for _ in 0..num_chunks {
let drun = take(&mut pos)? as usize;
let klen = take(&mut pos)? as usize;
let key = bytes
.get(pos..pos.saturating_add(klen))
.ok_or(FileError::Container("truncated dict chunk key"))?
.to_vec();
pos += klen;
let clen = take(&mut pos)?;
let first_run = prev_run + drun;
let body_start = meta
.restart_offsets
.get(first_run)
.copied()
.ok_or(FileError::Container("dict chunk run out of range"))?;
entries.push(DictChunkEntry {
first_run,
key,
body_start,
start: 0,
end: 0,
});
lens.push(clen);
prev_run = first_run;
}
let mut start = pos as u64;
for (e, len) in entries.iter_mut().zip(lens) {
let end = start
.checked_add(len)
.filter(|&e| e <= total_len)
.ok_or(FileError::Container("dict chunk overruns section"))?;
e.start = start;
e.end = end;
start = end;
}
Ok((meta, entries))
}
fn read_dict_dir_ranged<R: RangeReader>(
reader: &R,
section: ByteRange,
) -> Result<(crate::dict::SectionMeta, Vec<DictChunkEntry>), FileError> {
let total = section.len;
let init = 8192.min(total); let head = reader.read_at(section.offset, init)?;
let (header_len, n0) =
read_uvarint(&head).ok_or(FileError::Container("truncated dict header len"))?;
let hbase = n0; let (term_count, n1) = read_uvarint(head.get(hbase..).unwrap_or(&[]))
.ok_or(FileError::Container("truncated dict term_count"))?;
let (restart_interval, _n2) = read_uvarint(head.get(hbase + n1..).unwrap_or(&[]))
.ok_or(FileError::Container("truncated dict interval"))?;
if restart_interval == 0 {
return Err(FileError::Container("zero restart interval"));
}
let dir_start = (hbase as u64)
.checked_add(header_len)
.filter(|&d| d <= total)
.ok_or(FileError::Container("dict header overruns section"))?;
let dir_total = total - dir_start;
let meta = crate::dict::SectionMeta {
term_count: term_count as u32,
restart_interval: restart_interval as u32,
restart_offsets: Vec::new(),
};
let finish = |mut entries: Vec<DictChunkEntry>| {
for e in &mut entries {
e.start += dir_start; e.end += dir_start;
}
(meta.clone(), entries)
};
let mut have: Vec<u8> = Vec::new();
if dir_start < head.len() as u64 {
match parse_chunk_dir_only(&head[dir_start as usize..], dir_total)? {
ChunkDirParse::Done(entries) => return Ok(finish(entries)),
ChunkDirParse::Truncated { .. } => have = head[dir_start as usize..].to_vec(),
}
}
let mut want = 4096u64.min(dir_total).max(1);
loop {
let held = have.len() as u64;
if want > held {
let extra = reader.read_at(section.offset + dir_start + held, want - held)?;
if extra.is_empty() {
return Err(FileError::Container("truncated dict chunk directory"));
}
have.extend_from_slice(&extra);
}
let held = have.len() as u64;
match parse_chunk_dir_only(&have, dir_total)? {
ChunkDirParse::Done(entries) => return Ok(finish(entries)),
ChunkDirParse::Truncated { .. } if held >= dir_total => {
return Err(FileError::Container("truncated dict chunk directory"));
}
ChunkDirParse::Truncated {
parsed,
used,
total,
} => {
let est = if parsed >= 16 && used > 0 {
let whole = ((used as u64) / (parsed as u64)).saturating_mul(total as u64);
whole.saturating_add(whole / 8).saturating_add(64)
} else {
held.saturating_mul(2)
};
want = est
.max(held.saturating_add(1))
.min(held.saturating_mul(4))
.min(dir_total);
}
}
}
}
enum ChunkDirParse {
Done(Vec<DictChunkEntry>),
Truncated {
parsed: usize,
used: usize,
total: usize,
},
}
fn parse_chunk_dir_only(dir: &[u8], dir_total: u64) -> Result<ChunkDirParse, FileError> {
let mut pos = 0usize;
let take = |pos: &mut usize| -> Option<u64> {
let (v, n) = read_uvarint(dir.get(*pos..).unwrap_or(&[]))?;
*pos += n;
Some(v)
};
let Some(num_chunks) = take(&mut pos) else {
return Ok(ChunkDirParse::Truncated {
parsed: 0,
used: 0,
total: 0,
});
};
let num_chunks = num_chunks as usize;
let mut entries = Vec::with_capacity(num_chunks.min(dir.len()));
let mut lens: Vec<u64> = Vec::with_capacity(num_chunks.min(dir.len()));
let mut prev_run = 0usize;
for _ in 0..num_chunks {
let entry_start = pos;
let short = ChunkDirParse::Truncated {
parsed: entries.len(),
used: entry_start,
total: num_chunks,
};
let (Some(drun), Some(klen)) = (take(&mut pos), take(&mut pos)) else {
return Ok(short);
};
let (drun, klen) = (drun as usize, klen as usize);
let Some(key) = dir.get(pos..pos.saturating_add(klen)) else {
return Ok(short);
};
let key = key.to_vec();
pos += klen;
let Some(clen) = take(&mut pos) else {
return Ok(short);
};
let first_run = prev_run + drun;
entries.push(DictChunkEntry {
first_run,
key,
body_start: 0,
start: 0,
end: 0,
});
lens.push(clen);
prev_run = first_run;
}
let mut start = pos as u64;
for (e, len) in entries.iter_mut().zip(lens) {
let end = start
.checked_add(len)
.filter(|&e| e <= dir_total)
.ok_or(FileError::Container("dict chunk overruns section"))?;
e.start = start;
e.end = end;
start = end;
}
Ok(ChunkDirParse::Done(entries))
}
fn decode_chunked_dict_section(
payload: &[u8],
codec: u8,
cache: std::sync::Arc<crate::chunk_cache::ChunkCache>,
section_index: u8,
) -> Result<crate::dict::ChunkedSection, FileError> {
let (meta, entries) = parse_chunked_dict_dir(payload, payload.len() as u64)?;
let chunks = entries
.into_iter()
.map(|e| {
let body = decompress(codec, &payload[e.start as usize..e.end as usize])?;
Ok((
crate::dict::SectionChunk::new(e.first_run, e.key, e.body_start),
body,
))
})
.collect::<Result<Vec<_>, FileError>>()?;
Ok(crate::dict::ChunkedSection::resident(
meta,
chunks,
cache,
section_index,
))
}
fn decode_dictionary_container(bytes: &[u8], codec: u8) -> Result<Dictionary, FileError> {
let dsecs = decode_container(bytes, CODEC_NONE)?;
if dsecs.len() != 4 {
return Err(FileError::Container("expected 4 dictionary sections"));
}
let cache = crate::chunk_cache::ChunkCache::unlimited_arc();
let mut sections = Vec::with_capacity(4);
for (si, sec) in dsecs.iter().enumerate() {
sections.push(decode_chunked_dict_section(
sec,
codec,
cache.clone(),
si as u8,
)?);
}
let arr: [crate::dict::ChunkedSection; 4] = sections
.try_into()
.map_err(|_| FileError::Container("expected 4 dictionary sections"))?;
Ok(Dictionary::from_chunked_sections(arr))
}
fn encode_tiled_section(index: &GraphIndex, si: usize, codec: u8) -> Vec<u8> {
let tiles = index.tile_sections()[si];
let n = tiles.len();
#[cfg(feature = "parallel")]
let compressed: Vec<Vec<u8>> = {
use rayon::prelude::*;
(0..n)
.into_par_iter()
.map(|ti| compress(codec, &index.tile_body(si, ti)))
.collect()
};
#[cfg(not(feature = "parallel"))]
let compressed: Vec<Vec<u8>> = (0..n)
.map(|ti| compress(codec, &index.tile_body(si, ti)))
.collect();
let mut out = Vec::new();
write_uvarint(&mut out, tiles.len() as u64);
let mut prev_min = 0u32;
for (tile, comp) in tiles.iter().zip(&compressed) {
let (min_a, max_a) = tile.leading_range();
write_uvarint(&mut out, (min_a - prev_min) as u64);
write_uvarint(&mut out, (max_a - min_a) as u64);
write_uvarint(&mut out, comp.len() as u64);
prev_min = min_a;
}
for comp in &compressed {
out.extend_from_slice(comp);
}
for ti in 0..n {
let body = index.tile_body(si, ti);
let (min_b, max_b, min_c, max_c) = match crate::triples::TripleBlock::parse(&body) {
Ok(b) => {
let z = b.zone();
(z.min_b, z.max_b, z.min_c, z.max_c)
}
Err(_) => (0, u32::MAX, 0, u32::MAX),
};
write_uvarint(&mut out, min_b as u64);
write_uvarint(&mut out, (max_b - min_b) as u64);
write_uvarint(&mut out, min_c as u64);
write_uvarint(&mut out, (max_c - min_c) as u64);
}
out
}
struct TileDirEntry {
min_a: u32,
max_a: u32,
start: u64,
end: u64,
}
type TileSynopsis = (u32, u32, u32, u32);
fn parse_tile_synopsis(
payload: &[u8],
trailer_start: usize,
num_tiles: usize,
) -> Option<Vec<TileSynopsis>> {
let mut pos = trailer_start;
let take = |pos: &mut usize| -> Option<u32> {
let (v, n) = read_uvarint(payload.get(*pos..)?)?;
*pos += n;
u32::try_from(v).ok()
};
let mut out = Vec::with_capacity(num_tiles.min(payload.len()));
for _ in 0..num_tiles {
let min_b = take(&mut pos)?;
let max_b = min_b.checked_add(take(&mut pos)?)?;
let min_c = take(&mut pos)?;
let max_c = min_c.checked_add(take(&mut pos)?)?;
out.push((min_b, max_b, min_c, max_c));
}
Some(out)
}
fn parse_tile_directory(bytes: &[u8], total_len: u64) -> Result<Vec<TileDirEntry>, FileError> {
let mut pos = 0usize;
let take = |pos: &mut usize| -> Result<u64, FileError> {
let (v, n) = read_uvarint(bytes.get(*pos..).unwrap_or(&[]))
.ok_or(FileError::Container("truncated tile directory"))?;
*pos += n;
Ok(v)
};
let num_tiles = take(&mut pos)? as usize;
let mut entries = Vec::with_capacity(num_tiles.min(bytes.len()));
let mut prev_min = 0u32;
let mut lens = Vec::with_capacity(num_tiles.min(bytes.len()));
for _ in 0..num_tiles {
let dmin = take(&mut pos)? as u32;
let span = take(&mut pos)? as u32;
let len = take(&mut pos)?;
let min_a = prev_min.wrapping_add(dmin);
entries.push(TileDirEntry {
min_a,
max_a: min_a.wrapping_add(span),
start: 0,
end: 0,
});
lens.push(len);
prev_min = min_a;
}
let mut start = pos as u64;
for (e, len) in entries.iter_mut().zip(lens) {
let end = start
.checked_add(len)
.filter(|&e| e <= total_len)
.ok_or(FileError::Container("tile overruns section"))?;
e.start = start;
e.end = end;
start = end;
}
Ok(entries)
}
fn read_tile_directory_ranged<R: RangeReader>(
reader: &R,
section: ByteRange,
) -> Result<Vec<TileDirEntry>, FileError> {
let total = section.len;
let mut prefetch = 4096u64.min(total);
loop {
let prefix = reader.read_at(section.offset, prefetch)?;
match parse_tile_directory(&prefix, total) {
Ok(dir) => return Ok(dir),
Err(_) if prefetch < total => prefetch = prefetch.saturating_mul(2).min(total),
Err(e) => return Err(e),
}
}
}
fn read_tile_synopsis_ranged<R: RangeReader>(
reader: &R,
section: ByteRange,
dir: &[TileDirEntry],
) -> Vec<Option<TileSynopsis>> {
let n = dir.len();
let none = vec![None; n];
let trailer_start = dir.iter().map(|e| e.end).max().unwrap_or(0);
let total = section.len;
if n == 0 || trailer_start >= total {
return none; }
let trailer_len = total - trailer_start;
let Ok(bytes) = reader.read_at(section.offset + trailer_start, trailer_len) else {
return none;
};
match parse_tile_synopsis(&bytes, 0, n) {
Some(v) => v.into_iter().map(Some).collect(),
None => none,
}
}
fn tile_file_ranges(
index_bytes: &[u8],
container_offset: u64,
section_ranges: &[ByteRange; NUM_PERMS],
) -> [Vec<(u32, u32, ByteRange)>; NUM_PERMS] {
let mut out: [Vec<(u32, u32, ByteRange)>; NUM_PERMS] = Default::default();
for (section, range) in out.iter_mut().zip(section_ranges) {
if range.len == 0 || range.offset < container_offset {
continue;
}
let start = (range.offset - container_offset) as usize;
let Some(payload) = index_bytes.get(start..start + range.len as usize) else {
continue;
};
if let Ok(dir) = parse_tile_directory(payload, payload.len() as u64) {
*section = dir
.into_iter()
.map(|e| {
(
e.min_a,
e.max_a,
ByteRange {
offset: range.offset + e.start,
len: (e.end - e.start),
},
)
})
.collect();
}
}
out
}
fn decode_tiled_section(payload: &[u8], codec: u8) -> Result<Vec<(u32, u32, Vec<u8>)>, FileError> {
parse_tile_directory(payload, payload.len() as u64)?
.into_iter()
.map(|e| {
Ok((
e.min_a,
e.max_a,
decompress(codec, &payload[e.start as usize..e.end as usize])?,
))
})
.collect()
}
fn decode_index_container(
bytes: &[u8],
codec: u8,
perms: PermSet,
) -> Result<GraphIndex, FileError> {
let mut isecs = decode_container(bytes, CODEC_NONE)?;
if isecs.len() != perms.len() {
return Err(FileError::Container(
"index container section count does not match the header permutation mask",
));
}
let mut sections: [Vec<(u32, u32, Vec<u8>)>; NUM_PERMS] = Default::default();
for (perm, sec) in perms.iter().zip(isecs.iter_mut()) {
sections[perm.section_index()] = decode_tiled_section(sec, codec)?;
}
Ok(GraphIndex::from_tiles(sections, perms))
}
fn container_section_payload_ranges(
bytes: &[u8],
container_offset: u64,
expected_sections: usize,
) -> Result<Vec<ByteRange>, FileError> {
let (section_count, mut pos) =
read_uvarint(bytes).ok_or(FileError::Container("truncated count"))?;
let section_count = usize::try_from(section_count)
.map_err(|_| FileError::Container("section count too large"))?;
if section_count != expected_sections {
return Err(FileError::Container("unexpected section count"));
}
let mut ranges = Vec::with_capacity(section_count);
for _ in 0..section_count {
let remaining = bytes
.get(pos..)
.ok_or(FileError::Container("truncated length"))?;
let (payload_len, used) =
read_uvarint(remaining).ok_or(FileError::Container("truncated length"))?;
pos = pos
.checked_add(used)
.ok_or(FileError::Container("section range overflows"))?;
let payload_len_usize = usize::try_from(payload_len)
.map_err(|_| FileError::Container("section length too large"))?;
let payload_end = pos
.checked_add(payload_len_usize)
.ok_or(FileError::Container("section range overflows"))?;
if payload_end > bytes.len() {
return Err(FileError::Container("section overruns buffer"));
}
ranges.push(ByteRange {
offset: checked_end(container_offset, pos as u64)?,
len: payload_len,
});
pos = payload_end;
}
Ok(ranges)
}
fn decode_index_section_ranges(
bytes: &[u8],
container_offset: u64,
perms: PermSet,
) -> Result<[ByteRange; NUM_PERMS], FileError> {
let ranges = container_section_payload_ranges(bytes, container_offset, perms.len())?;
let mut out = [ByteRange { offset: 0, len: 0 }; NUM_PERMS];
for (perm, range) in perms.iter().zip(ranges) {
out[perm.section_index()] = range;
}
Ok(out)
}
fn read_uvarint_at<R: RangeReader>(
reader: &R,
absolute_offset: u64,
container_end: u64,
) -> Result<(u64, u64), FileError> {
if absolute_offset >= container_end {
return Err(FileError::Container("truncated container varint"));
}
let remaining = container_end - absolute_offset;
let probe_len = remaining.min(10);
let bytes = reader.read_at(absolute_offset, probe_len)?;
read_uvarint(&bytes)
.map(|(value, used)| (value, used as u64))
.ok_or(FileError::Container("truncated container varint"))
}
fn locate_container_section_ranged<R: RangeReader>(
reader: &R,
container_offset: u64,
container_len: u64,
section_index: usize,
expected_sections: u64,
) -> Result<ByteRange, FileError> {
let container_end = checked_end(container_offset, container_len)?;
let (section_count, used) = read_uvarint_at(reader, container_offset, container_end)?;
if section_count != expected_sections {
return Err(FileError::Container("unexpected container section count"));
}
if section_index >= section_count as usize {
return Err(FileError::Container(
"container section index out of bounds",
));
}
let mut pos = checked_end(container_offset, used)?;
for i in 0..section_count as usize {
let (payload_len, len_used) = read_uvarint_at(reader, pos, container_end)?;
pos = checked_end(pos, len_used)?;
let payload_end = checked_end(pos, payload_len)?;
if payload_end > container_end {
return Err(FileError::Container("section overruns buffer"));
}
if i == section_index {
return Ok(ByteRange {
offset: pos,
len: payload_len,
});
}
pos = payload_end;
}
Err(FileError::Container("container section not found"))
}
#[allow(clippy::type_complexity)]
fn open_index_container_lazy(
reader: &std::sync::Arc<dyn RangeReader + Send + Sync>,
container: ByteRange,
block_codec: u8,
has_synopsis: bool,
read_concurrency: usize,
perms: PermSet,
) -> Result<
(
GraphIndex,
[ByteRange; NUM_PERMS],
[Vec<(u32, u32, ByteRange)>; NUM_PERMS],
),
FileError,
> {
let mut index_section_ranges = [ByteRange { offset: 0, len: 0 }; NUM_PERMS];
let mut tile_ranges: [Vec<(u32, u32, ByteRange)>; NUM_PERMS] = Default::default();
#[allow(clippy::type_complexity)]
let mut directories: [Vec<(u32, u32, Option<TileSynopsis>)>; NUM_PERMS] = Default::default();
for (pos, perm) in perms.iter().enumerate() {
let si = perm.section_index();
let section = locate_container_section_ranged(
reader,
container.offset,
container.len,
pos,
perms.len() as u64,
)?;
index_section_ranges[si] = section;
let dir = read_tile_directory_ranged(reader, section)?;
let syn = if has_synopsis {
read_tile_synopsis_ranged(reader, section, &dir)
} else {
vec![None; dir.len()]
};
directories[si] = dir
.iter()
.zip(syn)
.map(|(e, s)| (e.min_a, e.max_a, s))
.collect();
tile_ranges[si] = dir
.into_iter()
.map(|e| {
(
e.min_a,
e.max_a,
ByteRange {
offset: section.offset + e.start,
len: (e.end - e.start),
},
)
})
.collect();
}
let codec = block_codec;
let loader_ranges = tile_ranges.clone();
let loader_reader = reader.clone();
let loader: crate::index::TileLoader = Box::new(move |si, ti| {
let (_, _, range) = loader_ranges.get(si)?.get(ti)?;
let bytes = loader_reader.read_at(range.offset, range.len).ok()?;
decompress(codec, &bytes).ok()
});
let bulk_ranges = tile_ranges.clone();
let bulk_reader = reader.clone();
let bulk: crate::index::TileBulkLoader = Box::new(move |si, tis| {
let section = bulk_ranges.get(si)?;
let want: Option<Vec<ByteRange>> = tis
.iter()
.map(|&ti| section.get(ti).map(|&(_, _, r)| r))
.collect();
let blobs = read_coalesced(bulk_reader.as_ref(), &want?, TILE_COALESCE_GAP)?;
blobs.iter().map(|b| decompress(codec, b).ok()).collect()
});
let mut index =
GraphIndex::from_remote_directories(directories, perms, loader).with_bulk_loader(bulk);
index.set_tile_lens(std::array::from_fn(|si| {
tile_ranges[si]
.iter()
.map(|&(_, _, r)| r.len.min(u32::MAX as u64) as u32)
.collect()
}));
index.set_read_concurrency(read_concurrency);
Ok((index, index_section_ranges, tile_ranges))
}
pub fn write_file(
dict: &Dictionary,
index: &GraphIndex,
has_quads: bool,
pyramid_meta: &[u8],
pyramid_levels: u16,
) -> Vec<u8> {
write_dataset(dict, index, &[], has_quads, pyramid_meta, pyramid_levels)
}
pub(crate) fn encode_index_container(index: &GraphIndex, codec: u8) -> Vec<u8> {
let payloads: Vec<Vec<u8>> = index
.perms()
.iter()
.map(|perm| encode_tiled_section(index, perm.section_index(), codec))
.collect();
let refs: Vec<&[u8]> = payloads.iter().map(|p| p.as_slice()).collect();
encode_container(&refs, CODEC_NONE)
}
fn encode_named_graphs(named: &[(String, GraphIndex)], codec: u8) -> Vec<u8> {
let mut out = Vec::new();
write_uvarint(&mut out, named.len() as u64);
for (iri, index) in named {
write_uvarint(&mut out, iri.len() as u64);
out.extend_from_slice(iri.as_bytes());
let container = encode_index_container(index, codec);
write_uvarint(&mut out, container.len() as u64);
out.extend_from_slice(&container);
}
out
}
fn decode_named_graphs(
bytes: &[u8],
codec: u8,
perms: PermSet,
) -> Result<Vec<(String, GraphIndex)>, FileError> {
let (n, mut pos) = read_uvarint(bytes).ok_or(FileError::Container("truncated graph count"))?;
let bound = |start: usize, len: u64| -> Result<usize, FileError> {
start
.checked_add(len as usize)
.filter(|&e| e <= bytes.len())
.ok_or(FileError::Container("named-graph field overruns buffer"))
};
let mut out = Vec::with_capacity((n as usize).min(bytes.len()));
for _ in 0..n {
let (ilen, u1) = read_uvarint(bytes.get(pos..).unwrap_or(&[]))
.ok_or(FileError::Container("truncated iri len"))?;
pos += u1;
let iend = bound(pos, ilen)?;
let iri = String::from_utf8_lossy(&bytes[pos..iend]).into_owned();
pos = iend;
let (clen, u2) = read_uvarint(bytes.get(pos..).unwrap_or(&[]))
.ok_or(FileError::Container("truncated container len"))?;
pos += u2;
let cend = bound(pos, clen)?;
let index = decode_index_container(&bytes[pos..cend], codec, perms)?;
out.push((iri, index));
pos = cend;
}
Ok(out)
}
pub fn write_dataset(
dict: &Dictionary,
default_index: &GraphIndex,
named: &[(String, GraphIndex)],
has_quads: bool,
pyramid_meta: &[u8],
pyramid_levels: u16,
) -> Vec<u8> {
write_dataset_with_metadata(
dict,
default_index,
named,
has_quads,
pyramid_meta,
pyramid_levels,
&[],
&[],
)
}
pub(crate) fn encode_dict_container(dict: &Dictionary, codec: u8) -> Vec<u8> {
let raw_sections = dict.sections();
let dict_payloads: Vec<Vec<u8>> = raw_sections
.iter()
.map(|raw| encode_chunked_dict_section(raw, codec))
.collect();
encode_container(
&[
dict_payloads[0].as_slice(),
dict_payloads[1].as_slice(),
dict_payloads[2].as_slice(),
dict_payloads[3].as_slice(),
],
CODEC_NONE,
)
}
#[allow(clippy::too_many_arguments)]
pub fn write_dataset_with_metadata(
dict: &Dictionary,
default_index: &GraphIndex,
named: &[(String, GraphIndex)],
has_quads: bool,
pyramid_meta: &[u8],
pyramid_levels: u16,
metadata: &[u8],
text_index: &[u8],
) -> Vec<u8> {
let codec = writer_codec();
let dict_container = encode_dict_container(dict, codec);
write_dataset_from_parts(
&dict_container,
dict.term_count() as u64,
default_index,
named,
has_quads,
dict.has_quoted_triples(),
pyramid_meta,
pyramid_levels,
metadata,
text_index,
codec,
)
}
#[allow(clippy::too_many_arguments)]
pub(crate) fn write_dataset_from_parts(
dict_container: &[u8],
term_count: u64,
default_index: &GraphIndex,
named: &[(String, GraphIndex)],
has_quads: bool,
has_quoted_triples: bool,
pyramid_meta: &[u8],
pyramid_levels: u16,
metadata: &[u8],
text_index: &[u8],
codec: u8,
) -> Vec<u8> {
let index_container = encode_index_container(default_index, codec);
let named_section = encode_named_graphs(named, codec);
let meta_section_len = metadata.len() as u64;
let dict_offset = HEADER_LEN as u64 + meta_section_len;
let dict_len = dict_container.len() as u64;
let index_offset = dict_offset + dict_len;
let index_len = index_container.len() as u64;
let pyr_offset = index_offset + index_len;
let pyr_len = pyramid_meta.len() as u64;
let text_offset = pyr_offset + pyr_len;
let text_len = text_index.len() as u64;
let named_offset = text_offset + text_len;
let named_len = if named.is_empty() {
0
} else {
named_section.len() as u64
};
let mut parts: Vec<&[u8]> = Vec::with_capacity(5);
if meta_section_len > 0 {
parts.push(metadata);
}
parts.push(dict_container);
parts.push(&index_container);
parts.push(pyramid_meta);
if text_len > 0 {
parts.push(text_index);
}
if named_len > 0 {
parts.push(&named_section);
}
let schema_meta_len = crate::meta::schema_block_len(pyramid_meta);
let header = Header {
version: crate::header::CURRENT_FORMAT_VERSION,
flags: FLAG_TILE_SYNOPSIS
| if has_quads { FLAG_HAS_QUADS } else { 0 }
| if has_quoted_triples {
FLAG_HAS_QUOTED_TRIPLES
} else {
0
},
metadata_offset: HEADER_LEN as u64,
metadata_len: meta_section_len,
dictionary_offset: dict_offset,
dictionary_len: dict_len,
root_dir_offset: index_offset,
root_dir_len: index_len,
pyramid_meta_offset: if pyr_len > 0 { pyr_offset } else { 0 },
pyramid_meta_len: pyr_len,
dict_codec: codec,
block_codec: codec,
pyramid_levels,
perms: default_index.perms(),
quad_count: default_index.triple_count() as u64
+ named
.iter()
.map(|(_, idx)| idx.triple_count() as u64)
.sum::<u64>(),
term_count,
content_hash: content_hash(&parts),
named_graphs_offset: if named_len > 0 { named_offset } else { 0 },
named_graphs_len: named_len,
schema_meta_len,
text_index_offset: if text_len > 0 { text_offset } else { 0 },
text_index_len: text_len,
build_info_offset: 0,
build_info_len: 0,
extra_sections: Vec::new(),
};
let mut out = Vec::with_capacity(
HEADER_LEN
+ metadata.len()
+ dict_container.len()
+ index_container.len()
+ pyramid_meta.len()
+ text_index.len()
+ named_section.len()
+ MAGIC.len(),
);
out.extend_from_slice(&header.to_bytes());
if meta_section_len > 0 {
out.extend_from_slice(metadata);
}
out.extend_from_slice(dict_container);
out.extend_from_slice(&index_container);
out.extend_from_slice(pyramid_meta);
if text_len > 0 {
out.extend_from_slice(text_index);
}
if named_len > 0 {
out.extend_from_slice(&named_section);
}
out.extend_from_slice(&MAGIC); out
}
pub const RDF_TYPE: &str = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
pub fn schema_summary(rete: &Rete) -> Vec<(String, String, String, u32)> {
use std::collections::{BTreeMap, HashMap};
let triples = rete.dump(None);
let mut class_of: HashMap<&str, &str> = HashMap::new();
for (s, p, o) in &triples {
if p == RDF_TYPE {
class_of.insert(s.as_str(), o.as_str());
}
}
let classify = |t: &str| -> String {
if let Some(c) = class_of.get(t) {
(*c).to_string()
} else if t.starts_with('"') {
"(literal)".to_string()
} else {
"(untyped)".to_string()
}
};
let mut counts: BTreeMap<(String, String, String), u32> = BTreeMap::new();
for (s, p, o) in &triples {
if p == RDF_TYPE {
continue; }
*counts
.entry((classify(s), p.clone(), classify(o)))
.or_default() += 1;
}
counts
.into_iter()
.map(|((a, p, b), c)| (a, p, b, c))
.collect()
}
pub fn schema_classes(rete: &Rete) -> Vec<(String, u32)> {
use std::collections::BTreeMap;
let mut counts: BTreeMap<String, u32> = BTreeMap::new();
for (_s, p, o) in rete.dump(None) {
if p == RDF_TYPE {
*counts.entry(o).or_default() += 1;
}
}
let mut out: Vec<(String, u32)> = counts.into_iter().collect();
out.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
out
}
pub fn read_metadata_ranged<R: RangeReader>(reader: &R) -> Result<Option<Vec<u8>>, FileError> {
let head = reader.read_at(0, HEADER_LEN as u64)?;
let header = Header::from_bytes(&head)?;
if header.metadata_len == 0 {
return Ok(None);
}
let bytes = reader.read_at(header.metadata_offset, header.metadata_len)?;
Ok(Some(bytes))
}
#[allow(clippy::type_complexity)]
pub fn read_card_and_build_info_ranged<R: RangeReader>(
reader: &R,
) -> Result<(Option<Vec<u8>>, Option<Vec<u8>>), FileError> {
read_card_and_build_info_with_header(reader).map(|(_, m, b)| (m, b))
}
#[allow(clippy::type_complexity)]
pub fn read_card_and_build_info_with_header<R: RangeReader>(
reader: &R,
) -> Result<(Header, Option<Vec<u8>>, Option<Vec<u8>>), FileError> {
let head = reader.read_at(0, HEADER_LEN as u64)?;
let header = Header::from_bytes(&head)?;
let meta = (header.metadata_offset, header.metadata_len);
let build = (header.build_info_offset, header.build_info_len);
if meta.1 > 0 && build.1 > 0 && build.0 == meta.0 + meta.1 {
let both = reader.read_at(meta.0, meta.1 + build.1)?;
let (m, b) = both.split_at(meta.1 as usize);
return Ok((header, Some(m.to_vec()), Some(b.to_vec())));
}
let fetch = |off: u64, len: u64| -> Result<Option<Vec<u8>>, FileError> {
if len == 0 {
return Ok(None);
}
Ok(Some(reader.read_at(off, len)?))
};
Ok((header, fetch(meta.0, meta.1)?, fetch(build.0, build.1)?))
}
pub fn read_build_info(bytes: &[u8]) -> Result<Option<Vec<u8>>, FileError> {
let header = Header::from_bytes(bytes)?;
if header.build_info_len == 0 {
return Ok(None);
}
let start = header.build_info_offset as usize;
let end = start
.checked_add(header.build_info_len as usize)
.filter(|&e| e <= bytes.len())
.ok_or(FileError::Container("build-info section overruns buffer"))?;
Ok(Some(bytes[start..end].to_vec()))
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BuildInfoPlan {
pub header: [u8; HEADER_LEN],
pub insert: u64,
pub tail_start: u64,
pub new_len: u64,
}
pub fn plan_build_info(
head: &[u8],
file_len: u64,
info_len: u64,
) -> Result<BuildInfoPlan, FileError> {
let mut header = Header::from_bytes(head)?;
let insert = HEADER_LEN as u64 + header.metadata_len;
let old_len = header.build_info_len;
if old_len > 0 && header.build_info_offset != insert {
return Err(FileError::Container(
"existing build-info section is not adjacent to the metadata",
));
}
let tail_start = insert.saturating_add(old_len);
if tail_start > file_len || insert > file_len {
return Err(FileError::Container("build-info splice out of bounds"));
}
let shift = |off: &mut u64, len: u64| {
if len > 0 && *off >= tail_start {
*off = *off - old_len + info_len;
}
};
shift(&mut header.dictionary_offset, header.dictionary_len);
shift(&mut header.root_dir_offset, header.root_dir_len);
shift(&mut header.pyramid_meta_offset, header.pyramid_meta_len);
shift(&mut header.named_graphs_offset, header.named_graphs_len);
shift(&mut header.text_index_offset, header.text_index_len);
for s in &mut header.extra_sections {
if s.length > 0 && s.offset >= tail_start {
s.offset = s.offset - old_len + info_len;
}
}
if info_len == 0 {
header.build_info_offset = 0;
header.build_info_len = 0;
} else {
header.build_info_offset = insert;
header.build_info_len = info_len;
}
Ok(BuildInfoPlan {
header: header.to_bytes(),
insert,
tail_start,
new_len: file_len - old_len + info_len,
})
}
pub fn attach_build_info(image: &[u8], info: &[u8]) -> Result<Vec<u8>, FileError> {
let plan = plan_build_info(image, image.len() as u64, info.len() as u64)?;
let mut out = Vec::with_capacity(plan.new_len as usize);
out.extend_from_slice(&plan.header);
out.extend_from_slice(&image[HEADER_LEN..plan.insert as usize]);
out.extend_from_slice(info);
out.extend_from_slice(&image[plan.tail_start as usize..]);
debug_assert_eq!(out.len() as u64, plan.new_len);
Ok(out)
}
pub fn replace_metadata(image: &[u8], metadata: &[u8]) -> Result<Vec<u8>, FileError> {
let mut header = Header::from_bytes(image)?;
let insert = HEADER_LEN as u64;
if header.metadata_len > 0 && header.metadata_offset != insert {
return Err(FileError::Container(
"metadata section does not sit immediately after the header",
));
}
let old_len = header.metadata_len;
let tail_start = (insert + old_len) as usize;
if tail_start > image.len() {
return Err(FileError::Container("metadata splice out of bounds"));
}
let new_len = metadata.len() as u64;
let shift = |off: &mut u64, len: u64| {
if len > 0 && *off >= insert + old_len {
*off = *off - old_len + new_len;
}
};
shift(&mut header.dictionary_offset, header.dictionary_len);
shift(&mut header.root_dir_offset, header.root_dir_len);
shift(&mut header.pyramid_meta_offset, header.pyramid_meta_len);
shift(&mut header.named_graphs_offset, header.named_graphs_len);
shift(&mut header.text_index_offset, header.text_index_len);
shift(&mut header.build_info_offset, header.build_info_len);
for s in &mut header.extra_sections {
if s.length > 0 && s.offset >= insert + old_len {
s.offset = s.offset - old_len + new_len;
}
}
header.metadata_offset = insert;
header.metadata_len = new_len;
let mut out = Vec::with_capacity(image.len() - old_len as usize + metadata.len());
out.extend_from_slice(&header.to_bytes());
out.extend_from_slice(metadata);
out.extend_from_slice(&image[tail_start..]);
header.content_hash = hash_of_sections(&header, &out)?;
out[..HEADER_LEN].copy_from_slice(&header.to_bytes());
Ok(out)
}
fn hash_of_sections(header: &Header, bytes: &[u8]) -> Result<[u8; 16], FileError> {
let slice = |off: u64, len: u64| -> Result<&[u8], FileError> {
let end = off.saturating_add(len);
bytes
.get(off as usize..end as usize)
.ok_or(FileError::Container("section overruns buffer"))
};
let d = slice(header.dictionary_offset, header.dictionary_len)?;
let i = slice(header.root_dir_offset, header.root_dir_len)?;
let m = if header.pyramid_meta_len > 0 {
slice(header.pyramid_meta_offset, header.pyramid_meta_len)?
} else {
&[]
};
let mut parts: Vec<&[u8]> = Vec::with_capacity(6);
if header.metadata_len > 0 {
parts.push(slice(header.metadata_offset, header.metadata_len)?);
}
parts.push(d);
parts.push(i);
parts.push(m);
if header.text_index_len > 0 {
parts.push(slice(header.text_index_offset, header.text_index_len)?);
}
if header.named_graphs_len > 0 {
parts.push(slice(header.named_graphs_offset, header.named_graphs_len)?);
}
Ok(content_hash(&parts))
}
pub fn verify(bytes: &[u8]) -> Result<bool, FileError> {
let header = Header::from_bytes(bytes)?;
Ok(hash_of_sections(&header, bytes)? == header.content_hash)
}
type PyramidLoader = Box<dyn Fn() -> Option<PyramidMeta> + Send + Sync>;
enum PyramidSlot {
Resident(Option<PyramidMeta>),
Lazy {
loader: PyramidLoader,
cell: std::sync::OnceLock<Option<PyramidMeta>>,
},
}
type TextIndexLoader = Box<dyn Fn() -> Option<crate::text_index::TextIndex> + Send + Sync>;
type TokenTableProbe = Box<dyn Fn() -> Option<u64> + Send + Sync>;
enum TextIndexSlot {
Resident {
index: Option<crate::text_index::TextIndex>,
token_table_len: Option<u64>,
},
Lazy {
loader: TextIndexLoader,
cell: std::sync::OnceLock<Option<crate::text_index::TextIndex>>,
token_table: TokenTableProbe,
token_table_cell: std::sync::OnceLock<Option<u64>>,
},
}
impl TextIndexSlot {
fn index(&self) -> Option<&crate::text_index::TextIndex> {
match self {
TextIndexSlot::Resident { index, .. } => index.as_ref(),
TextIndexSlot::Lazy { loader, cell, .. } => cell.get_or_init(loader).as_ref(),
}
}
fn token_table_len(&self) -> Option<u64> {
match self {
TextIndexSlot::Resident {
token_table_len, ..
} => *token_table_len,
TextIndexSlot::Lazy {
token_table,
token_table_cell,
..
} => *token_table_cell.get_or_init(token_table),
}
}
}
fn resident_text_index_slot(section: Option<&[u8]>, codec: u8) -> Result<TextIndexSlot, FileError> {
let Some(section) = section else {
return Ok(TextIndexSlot::Resident {
index: None,
token_table_len: None,
});
};
Ok(TextIndexSlot::Resident {
index: Some(
crate::text_index::TextIndex::from_section(section, codec)
.map_err(|_| FileError::Container("malformed text index"))?,
),
token_table_len: crate::text_index::TextIndex::postings_base(section).map(|b| b as u64),
})
}
pub fn read_text_index_token_table_len_ranged<R: RangeReader + ?Sized>(
reader: &R,
header: &Header,
) -> Option<u64> {
if header.text_index_len == 0 {
return None;
}
read_token_table_len(reader, header.text_index_offset, header.text_index_len)
}
fn read_token_table_len<R: RangeReader + ?Sized>(reader: &R, off: u64, len: u64) -> Option<u64> {
let head = reader.read_at(off, 10u64.min(len)).ok()?;
let (ttlen, n) = crate::varint::read_uvarint(&head)?;
Some((n as u64 + ttlen).min(len))
}
const NAMED_GRAPH_RESIDENT_MAX: u64 = 1 << 20;
const NAMED_SLAB: usize = 1024;
const NAMED_WALK_CHUNK: u64 = 64 * 1024;
const NAMED_WALK_CHUNK_MAX: u64 = 8 * 1024 * 1024;
#[derive(Default)]
struct NamedEntry {
meta: std::sync::OnceLock<(String, ByteRange)>,
index: std::sync::OnceLock<Box<GraphIndex>>,
}
#[derive(Default)]
struct NamedWalk {
next: usize,
pos: u64,
buf: Vec<u8>,
buf_off: u64,
chunk: u64,
big_seen: bool,
}
struct LazyNamedGraphs {
reader: std::sync::Arc<dyn RangeReader + Send + Sync>,
section: ByteRange,
codec: u8,
has_synopsis: bool,
read_concurrency: usize,
perms: PermSet,
#[allow(clippy::type_complexity)]
dir: std::sync::OnceLock<(usize, Box<[std::sync::OnceLock<Box<[NamedEntry]>>]>)>,
walk: std::sync::Mutex<NamedWalk>,
failed: std::sync::atomic::AtomicBool,
tile_cap: std::sync::atomic::AtomicU64,
}
impl LazyNamedGraphs {
fn new(
reader: std::sync::Arc<dyn RangeReader + Send + Sync>,
section: ByteRange,
codec: u8,
has_synopsis: bool,
read_concurrency: usize,
perms: PermSet,
) -> Self {
LazyNamedGraphs {
reader,
section,
codec,
has_synopsis,
read_concurrency,
perms,
dir: std::sync::OnceLock::new(),
walk: std::sync::Mutex::new(NamedWalk::default()),
failed: std::sync::atomic::AtomicBool::new(false),
tile_cap: std::sync::atomic::AtomicU64::new(u64::MAX),
}
}
fn set_tile_cap(&self, cap: u64) {
self.tile_cap
.store(cap, std::sync::atomic::Ordering::Relaxed);
self.for_each_opened(|g| {
if g.is_lazy() {
g.set_cache_cap(cap);
}
});
}
fn fail(&self) {
self.failed
.store(true, std::sync::atomic::Ordering::Relaxed);
}
#[allow(clippy::type_complexity)]
fn directory(&self) -> Option<&(usize, Box<[std::sync::OnceLock<Box<[NamedEntry]>>]>)> {
if let Some(d) = self.dir.get() {
return Some(d);
}
let end = self.section.offset.checked_add(self.section.len)?;
let (n, used) = match read_uvarint_at(&self.reader, self.section.offset, end) {
Ok(v) => v,
Err(_) => {
self.fail();
return None;
}
};
if n > self.section.len / 2 {
self.fail();
return None;
}
let n = n as usize;
let slabs = n.div_ceil(NAMED_SLAB);
let table: Box<[std::sync::OnceLock<Box<[NamedEntry]>>]> =
(0..slabs).map(|_| std::sync::OnceLock::new()).collect();
let _ = self.dir.set((n, table));
{
let mut w = self.walk.lock().unwrap();
if w.pos == 0 {
w.pos = self.section.offset + used;
}
}
self.dir.get()
}
fn count(&self) -> usize {
self.directory().map(|(n, _)| *n).unwrap_or(0)
}
fn entry(&self, i: usize) -> Option<&NamedEntry> {
let (n, table) = self.directory()?;
if i >= *n {
return None;
}
let slab = table[i / NAMED_SLAB].get_or_init(|| {
let len = NAMED_SLAB.min(n - (i / NAMED_SLAB) * NAMED_SLAB);
(0..len).map(|_| NamedEntry::default()).collect()
});
slab.get(i % NAMED_SLAB)
}
fn next_read_len(w: &mut NamedWalk, pos: u64, end: u64, need: u64) -> u64 {
let cur = w.chunk.max(NAMED_WALK_CHUNK);
w.chunk = (cur * 2).min(NAMED_WALK_CHUNK_MAX);
cur.max(need).min(end - pos)
}
fn ensure_meta(&self, upto: usize, exhaustive: bool) -> Option<()> {
let target = self.entry(upto)?;
if target.meta.get().is_some() {
return Some(());
}
let end = self.section.offset.checked_add(self.section.len)?;
let mut w = self.walk.lock().unwrap();
if exhaustive && !w.big_seen {
w.chunk = NAMED_WALK_CHUNK_MAX;
}
while w.next <= upto {
let pos = w.pos;
if pos >= end {
self.fail();
return None;
}
let have =
|b: &[u8], off: u64, need: u64| pos >= off && pos + need <= off + b.len() as u64;
if !have(&w.buf, w.buf_off, 20.min(end - pos)) {
let len = Self::next_read_len(&mut w, pos, end, 20.min(end - pos));
w.buf = match self.reader.read_at(pos, len) {
Ok(b) => b,
Err(_) => {
self.fail();
return None;
}
};
w.buf_off = pos;
}
let rel = (pos - w.buf_off) as usize;
let Some((ilen, u1)) = read_uvarint(&w.buf[rel..]) else {
self.fail();
return None;
};
let header_need = u1 as u64 + ilen + 10; if pos + u1 as u64 + ilen > end {
self.fail(); return None;
}
if !have(&w.buf, w.buf_off, header_need.min(end - pos)) {
let len = Self::next_read_len(&mut w, pos, end, header_need);
w.buf = match self.reader.read_at(pos, len) {
Ok(b) => b,
Err(_) => {
self.fail();
return None;
}
};
w.buf_off = pos;
}
let rel = (pos - w.buf_off) as usize;
let istart = rel + u1;
let iend = istart + ilen as usize;
let iri = String::from_utf8_lossy(&w.buf[istart..iend]).into_owned();
let Some((clen, u2)) = read_uvarint(&w.buf[iend..]) else {
self.fail();
return None;
};
let cstart = pos + u1 as u64 + ilen + u2 as u64;
let cend = match cstart.checked_add(clen) {
Some(e) if e <= end => e,
_ => {
self.fail(); return None;
}
};
if clen > NAMED_GRAPH_RESIDENT_MAX {
w.big_seen = true;
w.chunk = NAMED_WALK_CHUNK;
}
let range = ByteRange {
offset: cstart,
len: clen,
};
if let Some(e) = self.entry(w.next) {
let _ = e.meta.set((iri, range));
}
w.pos = cend;
w.next += 1;
}
Some(())
}
fn name_at(&self, i: usize, exhaustive: bool) -> Option<&str> {
self.ensure_meta(i, exhaustive)?;
self.entry(i)?.meta.get().map(|(iri, _)| iri.as_str())
}
fn graph_at(&self, i: usize, exhaustive: bool) -> Option<(&str, &GraphIndex)> {
self.ensure_meta(i, exhaustive)?;
let e = self.entry(i)?;
let (iri, range) = e.meta.get()?;
if e.index.get().is_none() {
let opened = self.open_graph(*range)?;
let _ = e.index.set(Box::new(opened));
}
Some((iri.as_str(), e.index.get()?.as_ref()))
}
fn container_bytes(&self, range: ByteRange) -> Option<Vec<u8>> {
{
let w = self.walk.lock().unwrap();
let buf_end = w.buf_off + w.buf.len() as u64;
if range.offset >= w.buf_off && range.offset + range.len <= buf_end {
let a = (range.offset - w.buf_off) as usize;
return Some(w.buf[a..a + range.len as usize].to_vec());
}
}
match self.reader.read_at(range.offset, range.len) {
Ok(b) => Some(b),
Err(_) => {
self.fail();
None
}
}
}
fn open_graph(&self, range: ByteRange) -> Option<GraphIndex> {
if range.len <= NAMED_GRAPH_RESIDENT_MAX {
let bytes = self.container_bytes(range)?;
match decode_index_container(&bytes, self.codec, self.perms) {
Ok(g) => Some(g),
Err(_) => {
self.fail();
None
}
}
} else {
match open_index_container_lazy(
&self.reader,
range,
self.codec,
self.has_synopsis,
self.read_concurrency,
self.perms,
) {
Ok((g, _, _)) => {
let cap = self.tile_cap.load(std::sync::atomic::Ordering::Relaxed);
if cap != u64::MAX {
g.set_cache_cap(cap);
}
Some(g)
}
Err(_) => {
self.fail();
None
}
}
}
}
fn release(&mut self, iri: &str) {
let Some((_, table)) = self.dir.get_mut() else {
return;
};
for slab in table.iter_mut() {
let Some(entries) = slab.get_mut() else {
continue;
};
for e in entries.iter_mut() {
let matches = e
.meta
.get()
.map(|(name, _)| name.as_str() == iri)
.unwrap_or(false);
if matches {
e.index.take();
return;
}
}
}
}
fn for_each_opened(&self, mut f: impl FnMut(&GraphIndex)) {
if let Some((_, table)) = self.dir.get() {
for slab in table.iter().filter_map(|s| s.get()) {
for e in slab.iter() {
if let Some(g) = e.index.get() {
f(g);
}
}
}
}
}
}
enum NamedGraphsSlot {
Resident(Vec<(String, GraphIndex)>),
Lazy(LazyNamedGraphs),
}
impl NamedGraphsSlot {
fn count(&self) -> usize {
match self {
NamedGraphsSlot::Resident(v) => v.len(),
NamedGraphsSlot::Lazy(l) => l.count(),
}
}
fn name_at(&self, i: usize, exhaustive: bool) -> Option<&str> {
match self {
NamedGraphsSlot::Resident(v) => v.get(i).map(|(iri, _)| iri.as_str()),
NamedGraphsSlot::Lazy(l) => l.name_at(i, exhaustive),
}
}
fn graph_at(&self, i: usize, exhaustive: bool) -> Option<(&str, &GraphIndex)> {
match self {
NamedGraphsSlot::Resident(v) => v.get(i).map(|(iri, g)| (iri.as_str(), g)),
NamedGraphsSlot::Lazy(l) => l.graph_at(i, exhaustive),
}
}
fn find(&self, iri: &str) -> Option<&GraphIndex> {
match self {
NamedGraphsSlot::Resident(v) => v.iter().find(|(name, _)| name == iri).map(|(_, g)| g),
NamedGraphsSlot::Lazy(l) => {
for i in 0..l.count() {
if l.name_at(i, false)? == iri {
return l.graph_at(i, false).map(|(_, g)| g);
}
}
None
}
}
}
fn load_incomplete(&self) -> bool {
match self {
NamedGraphsSlot::Resident(v) => v.iter().any(|(_, g)| g.load_incomplete()),
NamedGraphsSlot::Lazy(l) => {
if l.failed.load(std::sync::atomic::Ordering::Relaxed) {
return true;
}
let mut bad = false;
l.for_each_opened(|g| bad |= g.load_incomplete());
bad
}
}
}
fn reset_load_failures(&self) {
match self {
NamedGraphsSlot::Resident(v) => {
for (_, g) in v {
g.reset_load_failure();
}
}
NamedGraphsSlot::Lazy(l) => {
l.failed.store(false, std::sync::atomic::Ordering::Relaxed);
l.for_each_opened(|g| g.reset_load_failure());
}
}
}
}
pub struct Rete {
header: Header,
dict: Dictionary,
index: GraphIndex,
index_section_ranges: [ByteRange; NUM_PERMS],
tile_ranges: [Vec<(u32, u32, ByteRange)>; NUM_PERMS],
pyramid: PyramidSlot,
text_index: TextIndexSlot,
named_graphs: NamedGraphsSlot,
metadata: Vec<u8>,
service_client: Option<Box<dyn crate::service::ServiceClient>>,
service_error: std::sync::Mutex<Option<String>>,
}
impl Rete {
pub fn open(bytes: &[u8]) -> Result<Self, FileError> {
let header = Header::from_bytes(bytes)?;
let region = |off: u64, len: u64| -> Result<&[u8], FileError> {
let start = off as usize;
let end = start
.checked_add(len as usize)
.filter(|&e| e <= bytes.len())
.ok_or(FileError::Container("section range out of bounds"))?;
Ok(&bytes[start..end])
};
let dict = decode_dictionary_container(
region(header.dictionary_offset, header.dictionary_len)?,
header.dict_codec,
)?;
let index_bytes = region(header.root_dir_offset, header.root_dir_len)?;
let index = decode_index_container(index_bytes, header.block_codec, header.perms)?;
let index_section_ranges =
decode_index_section_ranges(index_bytes, header.root_dir_offset, header.perms)?;
let pyramid = PyramidSlot::Resident(if header.pyramid_meta_len > 0 {
Some(
PyramidMeta::decode(region(header.pyramid_meta_offset, header.pyramid_meta_len)?)
.map_err(|_| FileError::Container("malformed pyramid meta"))?,
)
} else {
None
});
let text_index = resident_text_index_slot(
if header.text_index_len > 0 {
Some(region(header.text_index_offset, header.text_index_len)?)
} else {
None
},
header.block_codec,
)?;
let named_graphs = NamedGraphsSlot::Resident(if header.named_graphs_len > 0 {
decode_named_graphs(
region(header.named_graphs_offset, header.named_graphs_len)?,
header.block_codec,
header.perms,
)?
} else {
Vec::new()
});
let metadata = if header.metadata_len > 0 {
region(header.metadata_offset, header.metadata_len)?.to_vec()
} else {
Vec::new()
};
let tile_ranges =
tile_file_ranges(index_bytes, header.root_dir_offset, &index_section_ranges);
Ok(Self {
header,
dict,
index,
index_section_ranges,
tile_ranges,
pyramid,
text_index,
named_graphs,
metadata,
service_client: None,
service_error: std::sync::Mutex::new(None),
})
}
pub fn set_service_client(&mut self, client: Box<dyn crate::service::ServiceClient>) {
self.service_client = Some(client);
}
pub(crate) fn service_client(&self) -> Option<&dyn crate::service::ServiceClient> {
self.service_client.as_deref()
}
pub(crate) fn record_service_error(&self, msg: &str) {
let mut e = self.service_error.lock().unwrap();
if e.is_none() {
*e = Some(msg.to_string());
}
}
pub(crate) fn take_service_error(&self) -> Option<String> {
self.service_error.lock().unwrap().take()
}
pub fn header(&self) -> &Header {
&self.header
}
pub fn file_layout(&self) -> Vec<LayoutSegment> {
let h = &self.header;
let seg = |kind: &'static str, label: String, offset: u64, len: u64| LayoutSegment {
kind,
label,
offset,
len,
};
let mut out = vec![seg(
"header",
format!("header (fixed {} bytes)", crate::header::HEADER_LEN),
0,
crate::header::HEADER_LEN as u64,
)];
if h.metadata_len > 0 {
out.push(seg(
"metadata",
"metadata (dataset card)".into(),
h.metadata_offset,
h.metadata_len,
));
}
out.push(seg(
"dictionary",
"dictionary (4 front-coded term sections)".into(),
h.dictionary_offset,
h.dictionary_len,
));
for (si, perm) in crate::index::ALL_PERMS.into_iter().enumerate() {
let sec = self.index_section_ranges[si];
if sec.len == 0 {
continue;
}
let first_tile = self.tile_ranges[si]
.first()
.map(|&(_, _, r)| r.offset)
.unwrap_or(sec.offset + sec.len);
if first_tile > sec.offset {
out.push(seg(
"directory",
format!("{} tile directory", perm.name()),
sec.offset,
first_tile - sec.offset,
));
}
for (ti, &(min_a, max_a, r)) in self.tile_ranges[si].iter().enumerate() {
out.push(seg(
"tile",
format!("{} tile {ti} (leading ids {min_a}..{max_a})", perm.name()),
r.offset,
r.len,
));
}
}
if h.pyramid_meta_len > 0 {
out.push(seg(
"pyramid",
"pyramid summary (communities + superedges)".into(),
h.pyramid_meta_offset,
h.pyramid_meta_len,
));
}
if h.named_graphs_len > 0 {
out.push(seg(
"named-graphs",
format!("named graphs ({})", self.named_graphs.count()),
h.named_graphs_offset,
h.named_graphs_len,
));
}
out.sort_by_key(|s| s.offset);
out
}
pub fn metadata(&self) -> Option<&[u8]> {
if self.metadata.is_empty() {
None
} else {
Some(&self.metadata)
}
}
pub fn dictionary(&self) -> &Dictionary {
&self.dict
}
pub fn pyramid(&self) -> Option<&PyramidMeta> {
match &self.pyramid {
PyramidSlot::Resident(p) => p.as_ref(),
PyramidSlot::Lazy { loader, cell } => cell.get_or_init(loader).as_ref(),
}
}
pub fn pyramid_if_loaded(&self) -> Option<&PyramidMeta> {
match &self.pyramid {
PyramidSlot::Resident(p) => p.as_ref(),
PyramidSlot::Lazy { cell, .. } => cell.get().and_then(|o| o.as_ref()),
}
}
pub fn predicate_stats(&self) -> &[crate::meta::PredStat] {
self.pyramid_if_loaded()
.map(|p| p.predicate_stats.as_slice())
.unwrap_or(&[])
}
pub fn char_sets(&self) -> &[crate::meta::CharSet] {
self.pyramid_if_loaded()
.map(|p| p.char_sets.as_slice())
.unwrap_or(&[])
}
pub fn label_index(&self) -> &[crate::meta::LabelEntry] {
self.pyramid_if_loaded()
.map(|p| p.label_index.as_slice())
.unwrap_or(&[])
}
pub fn prefix_search(&self, prefix: &str, limit: usize) -> Vec<(String, String)> {
let Some(pyr) = self.pyramid() else {
return Vec::new();
};
pyr.prefix_search(prefix, limit)
.into_iter()
.filter_map(|e| {
self.dict
.subject_term(e.subject)
.map(|iri| (e.label.clone(), iri))
})
.collect()
}
pub(crate) fn text_index(&self) -> Option<&crate::text_index::TextIndex> {
self.text_index.index()
}
pub fn has_text_index(&self) -> bool {
self.header.text_index_len > 0
}
pub fn text_index_token_table_len(&self) -> Option<u64> {
self.text_index.token_table_len()
}
pub fn text_search(&self, words: &[&str], prefix: Option<&str>, limit: usize) -> Vec<String> {
text_search_in(self.text_index(), &self.dict, words, prefix, limit)
}
pub fn default_index(&self) -> &GraphIndex {
&self.index
}
pub fn dump(&self, graph: Option<&str>) -> Vec<TermTriple> {
let index = match graph {
None => &self.index,
Some(g) => match self.graph_index(g) {
Some(i) => i,
None => return Vec::new(),
},
};
index
.match_pattern((None, None, None))
.into_iter()
.filter_map(|(s, p, o)| {
Some((
self.dict.subject_term(s)?,
self.dict.predicate_term(p)?,
self.dict.object_term(o)?,
))
})
.collect()
}
pub fn dump_each<F: FnMut(&str, &str, &str)>(&self, graph: Option<&str>, f: F) {
self.dump_filtered_each(graph, None, None, None, f)
}
pub fn dump_filtered_each<F: FnMut(&str, &str, &str)>(
&self,
graph: Option<&str>,
s: Option<&str>,
p: Option<&str>,
o: Option<&str>,
mut f: F,
) {
let index = match graph {
None => Some(&self.index),
Some(g) => self.graph_index(g),
};
let Some(index) = index else { return };
let Some(pattern) = self.resolve_query_pattern(s, p, o) else {
return;
};
let dict = &self.dict;
const WINDOW: usize = 4096;
let cap = dict.cache_cap();
let prefetch_window =
cap == u64::MAX || cap >= (WINDOW as u64) * 2 * (DICT_CHUNK_BUDGET as u64);
let mut ids: Vec<(u32, u32, u32)> = Vec::with_capacity(WINDOW);
let mut nodes: Vec<u32> = Vec::with_capacity(WINDOW * 2);
let mut preds: Vec<u32> = Vec::with_capacity(WINDOW);
let mut resolver = crate::dictionary::WindowResolver::default();
let mut flush = |ids: &mut Vec<(u32, u32, u32)>,
resolver: &mut crate::dictionary::WindowResolver,
f: &mut F| {
if ids.is_empty() {
return;
}
if prefetch_window {
nodes.clear();
preds.clear();
for &(s, p, o) in ids.iter() {
nodes.push(dict.subject_node(s));
nodes.push(dict.object_node(o));
preds.push(p);
}
dict.prefetch_terms(&nodes, &preds);
}
dict.resolve_window(ids, resolver, |s, p, o| f(s, p, o));
ids.clear();
};
for t in index.scan_iter(pattern) {
ids.push(t);
if ids.len() == WINDOW {
flush(&mut ids, &mut resolver, &mut f);
}
}
flush(&mut ids, &mut resolver, &mut f);
}
pub fn dump_plan(
&self,
graph: Option<&str>,
s: Option<&str>,
p: Option<&str>,
o: Option<&str>,
) -> DumpPlan {
let index = match graph {
None => Some(&self.index),
Some(g) => self.graph_index(g),
};
let scan = index
.zip(self.resolve_query_pattern(s, p, o))
.map(|(ix, pattern)| ix.scan_plan(pattern));
DumpPlan {
scan,
dictionary_bytes: self.header.dictionary_len,
}
}
pub fn dump_iter(&self, graph: Option<&str>) -> impl Iterator<Item = TermTriple> + '_ {
self.query_iter(graph, None, None, None)
}
pub fn query_iter(
&self,
graph: Option<&str>,
s: Option<&str>,
p: Option<&str>,
o: Option<&str>,
) -> impl Iterator<Item = TermTriple> + '_ {
let index = match graph {
None => Some(&self.index),
Some(g) => self.graph_index(g),
};
index
.zip(self.resolve_query_pattern(s, p, o))
.into_iter()
.flat_map(|(ix, pattern)| ix.scan_iter(pattern))
.filter_map(move |(s, p, o)| {
Some((
self.dict.subject_term(s)?,
self.dict.predicate_term(p)?,
self.dict.object_term(o)?,
))
})
}
pub fn query_batch(
&self,
graph: Option<&str>,
s: Option<&str>,
p: Option<&str>,
o: Option<&str>,
cursor: u64,
max_quads: usize,
) -> (Vec<TermTriple>, u64, bool) {
let index = match graph {
None => &self.index,
Some(g) => match self.graph_index(g) {
Some(i) => i,
None => return (Vec::new(), cursor, true),
},
};
let Some(pattern) = self.resolve_query_pattern(s, p, o) else {
return (Vec::new(), cursor, true);
};
let (ids, next, done) = index.scan_batch(pattern, cursor, max_quads);
if !ids.is_empty() {
let mut nodes = Vec::with_capacity(ids.len() * 2);
let mut preds = Vec::with_capacity(ids.len());
for &(s, p, o) in &ids {
nodes.push(self.dict.subject_node(s));
nodes.push(self.dict.object_node(o));
preds.push(p);
}
self.dict.prefetch_terms(&nodes, &preds);
}
let triples = ids
.into_iter()
.filter_map(|(s, p, o)| {
Some((
self.dict.subject_term(s)?,
self.dict.predicate_term(p)?,
self.dict.object_term(o)?,
))
})
.collect();
(triples, next, done)
}
pub fn dump_batch(
&self,
graph: Option<&str>,
cursor: u32,
max_quads: usize,
) -> (Vec<TermTriple>, u32, bool) {
fn next_tile_start(tiles: &[crate::index::Tile], sid: u32) -> Option<u32> {
let i = tiles.partition_point(|t| t.leading_range().1 < sid);
tiles.get(i).map(|t| t.leading_range().0.max(sid))
}
let index = match graph {
None => self.default_index(),
Some(g) => match self.graph_index(g) {
Some(i) => i,
None => return (Vec::new(), cursor, true),
},
};
let tiles = index.tile_sections()[IndexPermutation::Spo.section_index()];
let (Some(first), Some(last)) = (
tiles.first().map(|t| t.leading_range().0),
tiles.last().map(|t| t.leading_range().1),
) else {
return (Vec::new(), cursor, true); };
let mut sid = cursor.max(first);
let mut probes = max_quads.saturating_mul(4).max(1 << 16);
let mut ids: Vec<(u32, u32, u32)> = Vec::new();
let mut exhausted = false;
while sid <= last && probes > 0 {
probes -= 1;
let hits = index.match_pattern((Some(sid), None, None));
if hits.is_empty() {
match next_tile_start(tiles, sid) {
None => {
exhausted = true;
break;
}
Some(next) if next > sid => {
sid = next;
continue;
}
Some(_) => {}
}
} else {
ids.extend(hits);
}
if sid == u32::MAX {
exhausted = true;
break;
}
sid += 1;
if ids.len() >= max_quads {
break;
}
}
let done = exhausted || sid > last;
let dict = self.dictionary();
if !ids.is_empty() {
let mut nodes = Vec::with_capacity(ids.len() * 2);
let mut preds = Vec::with_capacity(ids.len());
for &(s, p, o) in &ids {
nodes.push(dict.subject_node(s));
nodes.push(dict.object_node(o));
preds.push(p);
}
dict.prefetch_terms(&nodes, &preds);
}
let triples = ids
.into_iter()
.filter_map(|(s, p, o)| {
Some((
dict.subject_term(s)?,
dict.predicate_term(p)?,
dict.object_term(o)?,
))
})
.collect();
(triples, sid, done)
}
pub fn named_graph_count(&self) -> usize {
self.named_graphs.count()
}
pub fn named_graph_name_at(&self, i: usize) -> Option<&str> {
self.named_graphs.name_at(i, false)
}
pub fn named_graph_at(&self, i: usize) -> Option<(&str, &GraphIndex)> {
self.named_graphs.graph_at(i, false)
}
pub(crate) fn named_graph_name_at_demand(&self, i: usize, exhaustive: bool) -> Option<&str> {
self.named_graphs.name_at(i, exhaustive)
}
pub(crate) fn named_graph_at_demand(
&self,
i: usize,
exhaustive: bool,
) -> Option<(&str, &GraphIndex)> {
self.named_graphs.graph_at(i, exhaustive)
}
pub fn graph_names(&self) -> Vec<&str> {
(0..self.named_graphs.count())
.filter_map(|i| self.named_graphs.name_at(i, false))
.collect()
}
pub fn graph_index(&self, iri: &str) -> Option<&GraphIndex> {
self.named_graphs.find(iri)
}
pub fn match_ids(
&self,
pattern: (Option<u32>, Option<u32>, Option<u32>),
) -> Vec<(u32, u32, u32)> {
self.index.match_pattern(pattern)
}
pub fn predicate_pairs(&self, predicate: &str) -> Vec<(u32, u32)> {
let pid = match self.dict.predicate_id(predicate) {
Some(p) => p,
None => return Vec::new(),
};
self.index
.match_pattern((None, Some(pid), None))
.into_iter()
.map(|(s, _p, o)| (self.dict.subject_node(s), self.dict.object_node(o)))
.collect()
}
pub fn open_ranged<R: RangeReader>(reader: &R) -> Result<Self, FileError> {
let head = reader.read_at(0, HEADER_LEN as u64)?;
let header = Header::from_bytes(&head)?;
let dict_bytes = reader.read_at(header.dictionary_offset, header.dictionary_len)?;
let dict = decode_dictionary_container(&dict_bytes, header.dict_codec)?;
let index_bytes = reader.read_at(header.root_dir_offset, header.root_dir_len)?;
let index = decode_index_container(&index_bytes, header.block_codec, header.perms)?;
let index_section_ranges =
decode_index_section_ranges(&index_bytes, header.root_dir_offset, header.perms)?;
let pyramid = PyramidSlot::Resident(if header.pyramid_meta_len > 0 {
let mb = reader.read_at(header.pyramid_meta_offset, header.pyramid_meta_len)?;
Some(
PyramidMeta::decode(&mb)
.map_err(|_| FileError::Container("malformed pyramid meta"))?,
)
} else {
None
});
let text_index_bytes = if header.text_index_len > 0 {
Some(reader.read_at(header.text_index_offset, header.text_index_len)?)
} else {
None
};
let text_index = resident_text_index_slot(text_index_bytes.as_deref(), header.block_codec)?;
let named_graphs = NamedGraphsSlot::Resident(if header.named_graphs_len > 0 {
let nb = reader.read_at(header.named_graphs_offset, header.named_graphs_len)?;
decode_named_graphs(&nb, header.block_codec, header.perms)?
} else {
Vec::new()
});
let tile_ranges =
tile_file_ranges(&index_bytes, header.root_dir_offset, &index_section_ranges);
Ok(Self {
header,
dict,
index,
index_section_ranges,
tile_ranges,
pyramid,
text_index,
named_graphs,
metadata: Vec::new(),
service_client: None,
service_error: std::sync::Mutex::new(None),
})
}
pub fn open_ranged_lazy<R: RangeReader + Send + Sync + 'static>(
reader: R,
) -> Result<Self, FileError> {
let head = reader.read_at(0, HEADER_LEN as u64)?;
let header = Header::from_bytes(&head)?;
let reader = std::sync::Arc::new(reader);
let read_concurrency = reader.concurrency();
let dict = ranged_chunked_dictionary(&reader, &header, [true; 4])?;
let reader_dyn: std::sync::Arc<dyn RangeReader + Send + Sync> = reader.clone();
let (index, index_section_ranges, tile_ranges) = open_index_container_lazy(
&reader_dyn,
ByteRange {
offset: header.root_dir_offset,
len: header.root_dir_len,
},
header.block_codec,
header.has_tile_synopsis(),
read_concurrency,
header.perms,
)?;
let pyramid = ranged_pyramid_slot(&reader, &header);
let text_index = ranged_text_index_slot(&reader, &header);
let named_graphs = if header.named_graphs_len > 0 {
NamedGraphsSlot::Lazy(LazyNamedGraphs::new(
reader_dyn,
ByteRange {
offset: header.named_graphs_offset,
len: header.named_graphs_len,
},
header.block_codec,
header.has_tile_synopsis(),
read_concurrency,
header.perms,
))
} else {
NamedGraphsSlot::Resident(Vec::new())
};
Ok(Self {
header,
dict,
index,
index_section_ranges,
tile_ranges,
pyramid,
text_index,
named_graphs,
metadata: Vec::new(),
service_client: None,
service_error: std::sync::Mutex::new(None),
})
}
pub fn index_incomplete(&self) -> bool {
self.index.load_incomplete()
|| self.dict.load_incomplete()
|| self.named_graphs.load_incomplete()
}
pub fn reset_load_failures(&self) {
self.index.reset_load_failure();
self.dict.reset_load_failure();
self.named_graphs.reset_load_failures();
}
pub fn set_memory_budget(&self, budget: Option<u64>) -> MemoryBudget {
let total = budget.unwrap_or(u64::MAX);
let split = split_memory_budget(total);
self.dict.set_cache_cap(split.dict_cache);
self.index.set_cache_cap(split.tile_cache);
if let NamedGraphsSlot::Lazy(lazy) = &self.named_graphs {
lazy.set_tile_cap(split.tile_cache);
}
if std::env::var("RETE_OPEN_DEBUG").is_ok() {
eprintln!(
"[memory-budget] total={} dict_cache={} tile_cache={} block_cache={}",
if total == u64::MAX {
"unlimited".to_string()
} else {
format!("{}MiB", total >> 20)
},
fmt_cap(split.dict_cache),
fmt_cap(split.tile_cache),
fmt_cap(split.block_cache),
);
}
split
}
pub fn dict_cache_stats(&self) -> crate::chunk_cache::CacheStats {
self.dict.cache_stats()
}
pub fn index_cache_stats(&self) -> crate::chunk_cache::CacheStats {
self.index.cache_stats()
}
pub fn dict_cache_cap(&self) -> u64 {
self.dict.cache_cap()
}
pub fn release_named_graph(&mut self, iri: &str) {
if let NamedGraphsSlot::Lazy(lazy) = &mut self.named_graphs {
lazy.release(iri);
}
}
fn resolve_query_pattern(
&self,
s: Option<&str>,
p: Option<&str>,
o: Option<&str>,
) -> Option<Pattern> {
let sid = match s {
Some(t) => match self.dict.subject_id(t) {
Some(id) => Some(id),
None => return None,
},
None => None,
};
let pid = match p {
Some(t) => match self.dict.predicate_id(t) {
Some(id) => Some(id),
None => return None,
},
None => None,
};
let oid = match o {
Some(t) => match self.dict.object_id(t) {
Some(id) => Some(id),
None => return None,
},
None => None,
};
Some((sid, pid, oid))
}
pub fn query_with_provenance(
&self,
s: Option<&str>,
p: Option<&str>,
o: Option<&str>,
) -> Vec<TripleProvenance> {
let pattern = match self.resolve_query_pattern(s, p, o) {
Some(pattern) => pattern,
None => return Vec::new(),
};
let index_permutation = GraphIndex::best_permutation_in(self.header.perms, pattern);
let dictionary_range = ByteRange {
offset: self.header.dictionary_offset,
len: self.header.dictionary_len,
};
let index_range = ByteRange {
offset: self.header.root_dir_offset,
len: self.header.root_dir_len,
};
let index_section_range = self.index_section_ranges[index_permutation.section_index()];
let pyramid_range = (self.header.pyramid_meta_len > 0).then_some(ByteRange {
offset: self.header.pyramid_meta_offset,
len: self.header.pyramid_meta_len,
});
let tiles = &self.tile_ranges[index_permutation.section_index()];
self.index
.match_pattern(pattern)
.into_iter()
.filter_map(|(s, p, o)| {
let terms = (
self.dict.subject_term(s)?,
self.dict.predicate_term(p)?,
self.dict.object_term(o)?,
);
let a = index_permutation.forward((s, p, o)).0;
let ti = tiles.partition_point(|&(_, max_a, _)| max_a < a);
let (tile, tile_range) = match tiles.get(ti) {
Some(&(min_a, _, range)) if min_a <= a => (
Some(format!("{}/{ti}", index_permutation.name())),
Some(range),
),
_ => (None, None),
};
Some(TripleProvenance {
terms,
ids: (s, p, o),
graph: None,
matched_pattern: pattern,
index_permutation,
dictionary_range,
index_range,
index_section_range,
pyramid_range,
tile,
tile_range,
})
})
.collect()
}
pub fn query(&self, s: Option<&str>, p: Option<&str>, o: Option<&str>) -> Vec<TermTriple> {
self.query_with_provenance(s, p, o)
.into_iter()
.map(|m| m.terms)
.collect()
}
pub fn query_in_graph(
&self,
graph: Option<&str>,
s: Option<&str>,
p: Option<&str>,
o: Option<&str>,
) -> Vec<TermTriple> {
let pattern = match self.resolve_query_pattern(s, p, o) {
Some(pattern) => pattern,
None => return Vec::new(),
};
let index = match graph {
None => &self.index,
Some(g) => match self.graph_index(g) {
Some(i) => i,
None => return Vec::new(),
},
};
let ids = index.match_pattern(pattern);
if !ids.is_empty() {
let mut nodes = Vec::with_capacity(ids.len() * 2);
let mut preds = Vec::with_capacity(ids.len());
for &(s, p, o) in &ids {
nodes.push(self.dict.subject_node(s));
nodes.push(self.dict.object_node(o));
preds.push(p);
}
self.dict.prefetch_terms(&nodes, &preds);
}
ids.into_iter()
.filter_map(|(s, p, o)| {
Some((
self.dict.subject_term(s)?,
self.dict.predicate_term(p)?,
self.dict.object_term(o)?,
))
})
.collect()
}
pub fn query_quads(
&self,
s: Option<&str>,
p: Option<&str>,
o: Option<&str>,
) -> Vec<(TermTriple, Option<String>)> {
let mut out: Vec<(TermTriple, Option<String>)> = self
.query_in_graph(None, s, p, o)
.into_iter()
.map(|t| (t, None))
.collect();
for i in 0..self.named_graphs.count() {
let Some(iri) = self.named_graphs.name_at(i, true) else {
continue;
};
for triple in self.query_in_graph(Some(iri), s, p, o) {
out.push((triple, Some(iri.to_string())));
}
}
out
}
pub fn query_ranged<R: RangeReader>(
reader: &R,
s: Option<&str>,
p: Option<&str>,
o: Option<&str>,
) -> Result<Vec<TermTriple>, FileError> {
let routed = match route_pattern(reader, s, p, o)? {
Some(routed) => routed,
None => return Ok(Vec::new()),
};
let matches = fetch_routed_matches(reader, &routed)?;
Ok(matches
.into_iter()
.filter_map(|(s, p, o)| {
Some((
routed.dict.subject_term(s)?,
routed.dict.predicate_term(p)?,
routed.dict.object_term(o)?,
))
})
.collect())
}
pub fn route_pattern_ranged<R: RangeReader>(
reader: &R,
s: Option<&str>,
p: Option<&str>,
o: Option<&str>,
) -> Result<bool, FileError> {
Ok(route_pattern(reader, s, p, o)?.is_some())
}
}
struct RoutedPattern {
dict: Dictionary,
pattern: Pattern,
permutation: IndexPermutation,
header: Header,
section: ByteRange,
}
fn route_pattern<R: RangeReader>(
reader: &R,
s: Option<&str>,
p: Option<&str>,
o: Option<&str>,
) -> Result<Option<RoutedPattern>, FileError> {
let head = reader.read_at(0, HEADER_LEN as u64)?;
let header = Header::from_bytes(&head)?;
let dict_bytes = reader.read_at(header.dictionary_offset, header.dictionary_len)?;
let dict = decode_dictionary_container(&dict_bytes, header.dict_codec)?;
let Some(pattern) = resolve_query_pattern(&dict, s, p, o) else {
return Ok(None);
};
let permutation = GraphIndex::best_permutation_in(header.perms, pattern);
let section = locate_container_section_ranged(
reader,
header.root_dir_offset,
header.root_dir_len,
header
.perms
.position(permutation)
.ok_or(FileError::Container("routed to an absent permutation"))?,
header.perms.len() as u64,
)?;
Ok(Some(RoutedPattern {
dict,
pattern,
permutation,
header,
section,
}))
}
fn fetch_routed_matches<R: RangeReader>(
reader: &R,
routed: &RoutedPattern,
) -> Result<Vec<Triple>, FileError> {
let dir = read_tile_directory_ranged(reader, routed.section)?;
let [pa, _, _] = routed.permutation.order_pattern(routed.pattern);
let codec = routed.header.block_codec;
let mut out = Vec::new();
match pa {
Some(a) => {
for e in dir.iter().filter(|e| e.min_a <= a && a <= e.max_a) {
let bytes = reader.read_at(routed.section.offset + e.start, e.end - e.start)?;
let tile = decompress(codec, &bytes)?;
out.extend(GraphIndex::match_serialized_block(
&tile,
routed.permutation,
routed.pattern,
));
}
}
None => {
if let (Some(first), Some(last)) = (dir.first(), dir.last()) {
let base = first.start;
let body = reader.read_at(routed.section.offset + base, last.end - base)?;
for e in &dir {
let tile = decompress(
codec,
&body[(e.start - base) as usize..(e.end - base) as usize],
)?;
out.extend(GraphIndex::match_serialized_block(
&tile,
routed.permutation,
routed.pattern,
));
}
}
}
}
out.sort_unstable();
Ok(out)
}
fn resolve_query_pattern(
dict: &Dictionary,
s: Option<&str>,
p: Option<&str>,
o: Option<&str>,
) -> Option<Pattern> {
let sid = match s {
Some(t) => Some(dict.subject_id(t)?),
None => None,
};
let pid = match p {
Some(t) => Some(dict.predicate_id(t)?),
None => None,
};
let oid = match o {
Some(t) => Some(dict.object_id(t)?),
None => None,
};
Some((sid, pid, oid))
}
fn ranged_chunked_dictionary<R: RangeReader + Send + Sync + 'static>(
reader: &std::sync::Arc<R>,
header: &Header,
want: [bool; 4],
) -> Result<Dictionary, FileError> {
let cache = crate::chunk_cache::ChunkCache::unlimited_arc();
let mut dict_sections: Vec<crate::dict::ChunkedSection> = Vec::with_capacity(4);
for si in 0..4 {
if !want[si] {
dict_sections.push(crate::dict::ChunkedSection::from_parts(
crate::dict::SectionMeta {
term_count: 0,
restart_interval: 1,
restart_offsets: Vec::new(),
},
Vec::new(),
None,
cache.clone(),
si as u8,
));
continue;
}
let section = locate_container_section_ranged(
reader.as_ref(),
header.dictionary_offset,
header.dictionary_len,
si,
4,
)?;
let (meta, entries) = read_dict_dir_ranged(reader.as_ref(), section)?;
let ranges: Vec<ByteRange> = entries
.iter()
.map(|e| ByteRange {
offset: section.offset + e.start,
len: (e.end - e.start),
})
.collect();
let chunks: Vec<crate::dict::SectionChunk> = entries
.into_iter()
.map(|e| crate::dict::SectionChunk::new(e.first_run, e.key, e.body_start))
.collect();
let chunk_reader = reader.clone();
let codec = header.dict_codec;
let loader_ranges = ranges.clone();
let loader: crate::dict::ChunkLoader = Box::new(move |ci| {
let range = loader_ranges.get(ci)?;
let bytes = chunk_reader.read_at(range.offset, range.len).ok()?;
decompress(codec, &bytes).ok()
});
let bulk_reader = reader.clone();
let bulk: crate::dict::ChunkBulkLoader = Box::new(move |cis| {
let want: Option<Vec<ByteRange>> =
cis.iter().map(|&ci| ranges.get(ci).copied()).collect();
let blobs = read_coalesced(bulk_reader.as_ref(), &want?, DICT_COALESCE_GAP)?;
blobs.iter().map(|b| decompress(codec, b).ok()).collect()
});
dict_sections.push(
crate::dict::ChunkedSection::from_parts(
meta,
chunks,
Some(loader),
cache.clone(),
si as u8,
)
.with_bulk_loader(bulk),
);
}
let dict_arr: [crate::dict::ChunkedSection; 4] = dict_sections
.try_into()
.map_err(|_| FileError::Container("expected 4 dictionary sections"))?;
Ok(Dictionary::from_chunked_sections(dict_arr))
}
fn ranged_pyramid_slot<R: RangeReader + Send + Sync + 'static>(
reader: &std::sync::Arc<R>,
header: &Header,
) -> PyramidSlot {
if header.pyramid_meta_len == 0 {
return PyramidSlot::Resident(None);
}
let pyr_reader = reader.clone();
let pyr_off = header.pyramid_meta_offset;
let pyr_len = header.pyramid_meta_len;
PyramidSlot::Lazy {
loader: Box::new(move || {
let mb = pyr_reader.read_at(pyr_off, pyr_len).ok()?;
PyramidMeta::decode(&mb).ok()
}),
cell: std::sync::OnceLock::new(),
}
}
fn ranged_text_index_slot<R: RangeReader + Send + Sync + 'static>(
reader: &std::sync::Arc<R>,
header: &Header,
) -> TextIndexSlot {
if header.text_index_len == 0 {
return TextIndexSlot::Resident {
index: None,
token_table_len: None,
};
}
let ti_reader = reader.clone();
let probe_reader = reader.clone();
let ti_off = header.text_index_offset;
let ti_len = header.text_index_len;
let codec = header.block_codec;
TextIndexSlot::Lazy {
loader: Box::new(move || {
let prefix_len = read_token_table_len(&*ti_reader, ti_off, ti_len)?;
let prefix = ti_reader.read_at(ti_off, prefix_len).ok()?;
let postings_base = crate::text_index::TextIndex::postings_base(&prefix)? as u64;
let postings_abs = ti_off + postings_base;
let pr = ti_reader.clone();
let posting_loader =
Box::new(move |off: u64, len: u64| pr.read_at(postings_abs + off, len).ok());
crate::text_index::TextIndex::from_token_table(&prefix, codec, posting_loader).ok()
}),
cell: std::sync::OnceLock::new(),
token_table: Box::new(move || read_token_table_len(&*probe_reader, ti_off, ti_len)),
token_table_cell: std::sync::OnceLock::new(),
}
}
fn text_search_in(
ti: Option<&crate::text_index::TextIndex>,
dict: &Dictionary,
words: &[&str],
prefix: Option<&str>,
limit: usize,
) -> Vec<String> {
let Some(ti) = ti else {
return Vec::new();
};
let mut acc: Option<Vec<u32>> = None;
if let Some(p) = prefix {
acc = Some(ti.prefix(&p.to_lowercase()));
}
for w in words {
for tok in crate::text_index::tokenize(w) {
let posting = ti.lookup(&tok);
acc = Some(match acc {
Some(a) => intersect_sorted(&a, &posting),
None => posting,
});
if acc.as_ref().is_some_and(|a| a.is_empty()) {
return Vec::new();
}
}
}
let ids = acc.unwrap_or_default();
let mut out = Vec::with_capacity(if limit > 0 {
limit.min(ids.len())
} else {
ids.len()
});
for id in ids {
if let Some(iri) = dict.subject_term(id) {
out.push(iri);
if limit > 0 && out.len() >= limit {
break;
}
}
}
out
}
pub struct SearchView {
header: Header,
dict: Dictionary,
pyramid: PyramidSlot,
text_index: TextIndexSlot,
}
impl SearchView {
pub fn open_ranged<R: RangeReader + Send + Sync + 'static>(
reader: R,
) -> Result<Self, FileError> {
let head = reader.read_at(0, HEADER_LEN as u64)?;
let header = Header::from_bytes(&head)?;
let reader = std::sync::Arc::new(reader);
let dict = ranged_chunked_dictionary(&reader, &header, [true, true, false, false])?;
let pyramid = ranged_pyramid_slot(&reader, &header);
let text_index = ranged_text_index_slot(&reader, &header);
Ok(Self {
header,
dict,
pyramid,
text_index,
})
}
pub fn header(&self) -> &Header {
&self.header
}
pub fn has_text_index(&self) -> bool {
self.header.text_index_len > 0
}
pub fn has_pyramid(&self) -> bool {
self.header.pyramid_meta_len > 0
}
pub fn text_search(&self, words: &[&str], prefix: Option<&str>, limit: usize) -> Vec<String> {
text_search_in(self.text_index.index(), &self.dict, words, prefix, limit)
}
pub fn text_index_token_table_len(&self) -> Option<u64> {
self.text_index.token_table_len()
}
pub fn prefix_search(&self, prefix: &str, limit: usize) -> Vec<(String, String)> {
let pyr = match &self.pyramid {
PyramidSlot::Resident(p) => p.as_ref(),
PyramidSlot::Lazy { loader, cell } => cell.get_or_init(loader).as_ref(),
};
let Some(pyr) = pyr else {
return Vec::new();
};
pyr.prefix_search(prefix, limit)
.into_iter()
.filter_map(|e| {
self.dict
.subject_term(e.subject)
.map(|iri| (e.label.clone(), iri))
})
.collect()
}
}
#[must_use]
pub struct SummaryView {
pub round: u32,
pub summary: Vec<SuperEdge>,
pub class_hierarchy: Vec<ClassNode>,
pub level_rollups: Vec<LevelRollup>,
pub level_links: Vec<LevelLinks>,
pub descriptors: Vec<CommunityDescriptor>,
pub subclass_cycles: Vec<Vec<String>>,
pub disjoint_pairs: Vec<(String, String)>,
pub equivalent_pairs: Vec<(String, String)>,
dict: Dictionary,
}
impl SummaryView {
pub fn open_ranged<R: RangeReader>(reader: &R) -> Result<Option<Self>, FileError> {
let head = reader.read_at(0, HEADER_LEN as u64)?;
let header = Header::from_bytes(&head)?;
if header.pyramid_meta_len == 0 {
return Ok(None);
}
let dict_bytes = reader.read_at(header.dictionary_offset, header.dictionary_len)?;
let dict = decode_dictionary_container(&dict_bytes, header.dict_codec)?;
let mb = reader.read_at(header.pyramid_meta_offset, header.pyramid_meta_len)?;
let meta =
PyramidMeta::decode(&mb).map_err(|_| FileError::Container("malformed pyramid meta"))?;
Ok(Some(SummaryView {
round: meta.round,
summary: meta.summary,
class_hierarchy: meta.class_hierarchy,
level_rollups: meta.level_rollups,
level_links: meta.level_links,
descriptors: meta.descriptors,
subclass_cycles: meta.subclass_cycles,
disjoint_pairs: meta.disjoint_pairs,
equivalent_pairs: meta.equivalent_pairs,
dict,
}))
}
pub fn level_count(&self) -> usize {
self.level_rollups.len()
}
pub fn level_rollup(&self, k: usize) -> Option<&LevelRollup> {
self.level_rollups.get(k)
}
pub fn predicate_term(&self, id: u32) -> Option<String> {
self.dict.predicate_term(id)
}
pub fn predicate_total(&self, predicate: &str) -> u32 {
match self.dict.predicate_id(predicate) {
Some(pid) => self
.summary
.iter()
.filter(|e| e.predicate == pid)
.map(|e| e.count)
.sum(),
None => 0,
}
}
pub fn predicate_totals(&self) -> Vec<(String, u32)> {
let mut by_pred: std::collections::BTreeMap<u32, u32> = std::collections::BTreeMap::new();
for e in &self.summary {
*by_pred.entry(e.predicate).or_default() += e.count;
}
let mut out: Vec<(String, u32)> = by_pred
.into_iter()
.filter_map(|(pid, c)| self.dict.predicate_term(pid).map(|t| (t, c)))
.collect();
out.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
out
}
pub fn community_count(&self) -> usize {
let mut comms = std::collections::BTreeSet::new();
for e in &self.summary {
comms.insert(e.s_comm);
comms.insert(e.o_comm);
}
comms.len()
}
pub fn tbox_coherence(&self) -> Vec<crate::reason::Inconsistency> {
schema_coherence(
&self.class_hierarchy,
&self.subclass_cycles,
&self.disjoint_pairs,
&self.equivalent_pairs,
)
}
pub fn tbox_is_coherent(&self) -> bool {
self.tbox_coherence().is_empty()
}
}
pub fn schema_coherence(
class_hierarchy: &[ClassNode],
subclass_cycles: &[Vec<String>],
disjoint_pairs: &[(String, String)],
equivalent_pairs: &[(String, String)],
) -> Vec<crate::reason::Inconsistency> {
use crate::reason::Inconsistency;
use std::collections::{BTreeMap, BTreeSet, VecDeque};
const MAX_REACH: usize = 100_000;
let mut out: Vec<Inconsistency> = Vec::new();
for cyc in subclass_cycles {
let detail = if cyc.len() == 1 {
format!("{} is rdfs:subClassOf itself (a cycle)", cyc[0])
} else {
format!(
"classes {{{}}} are mutually rdfs:subClassOf (a cycle)",
cyc.join(", ")
)
};
out.push(Inconsistency {
kind: "subclass-cycle",
detail,
});
}
if !disjoint_pairs.is_empty() {
let mut adj: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
for n in class_hierarchy {
let e = adj.entry(n.class.as_str()).or_default();
for p in &n.parents {
e.push(p.as_str());
}
}
for (a, b) in equivalent_pairs {
adj.entry(a.as_str()).or_default().push(b.as_str());
adj.entry(b.as_str()).or_default().push(a.as_str());
}
let mut focuses: BTreeSet<&str> =
class_hierarchy.iter().map(|n| n.class.as_str()).collect();
for (a, b) in disjoint_pairs.iter().chain(equivalent_pairs) {
focuses.insert(a.as_str());
focuses.insert(b.as_str());
}
let mut seen: BTreeSet<&str> = BTreeSet::new();
for &c in &focuses {
let mut reach: BTreeSet<&str> = BTreeSet::new();
let mut q: VecDeque<&str> = VecDeque::new();
reach.insert(c);
q.push_back(c);
while let Some(x) = q.pop_front() {
if reach.len() > MAX_REACH {
break;
}
if let Some(ns) = adj.get(x) {
for &p in ns {
if reach.insert(p) {
q.push_back(p);
}
}
}
}
for (x, y) in disjoint_pairs {
if reach.contains(x.as_str()) && reach.contains(y.as_str()) && seen.insert(c) {
out.push(Inconsistency {
kind: "unsatisfiable-class",
detail: format!(
"{c} is a subclass of both {x} and {y}, which are \
owl:disjointWith — no individual can be a {c}"
),
});
break;
}
}
}
}
out.sort_by(|a, b| (a.kind, &a.detail).cmp(&(b.kind, &b.detail)));
out
}
pub fn read_schema_coherence_ranged<R: RangeReader>(
reader: &R,
) -> Result<Option<Vec<crate::reason::Inconsistency>>, FileError> {
let head = reader.read_at(0, HEADER_LEN as u64)?;
let header = Header::from_bytes(&head)?;
if header.pyramid_meta_len == 0 {
return Ok(None);
}
if header.schema_meta_len > 0 && (header.schema_meta_len as u64) <= header.pyramid_meta_len {
let off =
header.pyramid_meta_offset + header.pyramid_meta_len - header.schema_meta_len as u64;
let block = reader.read_at(off, header.schema_meta_len as u64)?;
let (hierarchy, cycles, disjoint, equivalent) = crate::meta::decode_schema_block(&block)
.map_err(|_| FileError::Container("malformed schema block"))?;
return Ok(Some(schema_coherence(
&hierarchy,
&cycles,
&disjoint,
&equivalent,
)));
}
let mb = reader.read_at(header.pyramid_meta_offset, header.pyramid_meta_len)?;
let meta =
PyramidMeta::decode(&mb).map_err(|_| FileError::Container("malformed pyramid meta"))?;
Ok(Some(schema_coherence(
&meta.class_hierarchy,
&meta.subclass_cycles,
&meta.disjoint_pairs,
&meta.equivalent_pairs,
)))
}
#[allow(clippy::type_complexity)]
pub fn read_schema_summary_ranged<R: RangeReader>(
reader: &R,
) -> Result<Option<(Vec<(String, u64)>, Vec<(String, String, String, u64)>)>, FileError> {
let head = reader.read_at(0, HEADER_LEN as u64)?;
let header = Header::from_bytes(&head)?;
if header.pyramid_meta_len == 0 {
return Ok(None);
}
if header.schema_meta_len > 0 && (header.schema_meta_len as u64) <= header.pyramid_meta_len {
let off =
header.pyramid_meta_offset + header.pyramid_meta_len - header.schema_meta_len as u64;
let block = reader.read_at(off, header.schema_meta_len as u64)?;
let summary = crate::meta::decode_schema_block_summary(&block)
.map_err(|_| FileError::Container("malformed schema block"))?;
return Ok(Some(summary));
}
let mb = reader.read_at(header.pyramid_meta_offset, header.pyramid_meta_len)?;
let meta =
PyramidMeta::decode(&mb).map_err(|_| FileError::Container("malformed pyramid meta"))?;
if meta.level_rollups.is_empty() && meta.level_links.is_empty() {
return Ok(None);
}
let classes = meta
.level_rollups
.iter()
.max_by_key(|r| r.depth)
.map(|r| r.classes.clone())
.unwrap_or_default();
let relations = meta
.level_links
.iter()
.max_by_key(|l| l.depth)
.map(|l| {
l.links
.iter()
.map(|c| {
(
c.s_class.clone(),
c.predicate.clone(),
c.o_class.clone(),
c.count,
)
})
.collect()
})
.unwrap_or_default();
Ok(Some((classes, relations)))
}
#[cfg(test)]
pub(crate) fn dict_chunk_keys_for_test(image: &[u8]) -> Vec<Vec<Vec<u8>>> {
let header = Header::from_bytes(&image[..HEADER_LEN]).unwrap();
let s = header.dictionary_offset as usize;
let e = s + header.dictionary_len as usize;
decode_container(&image[s..e], CODEC_NONE)
.unwrap()
.iter()
.map(|payload| {
let (_meta, entries) = parse_chunked_dict_dir(payload, payload.len() as u64).unwrap();
entries.into_iter().map(|c| c.key).collect()
})
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::dictionary::DictionaryBuilder;
use crate::index::GraphIndexBuilder;
#[test]
fn three_permutation_file_round_trips_every_open_path() {
use crate::index::PermSet;
use crate::reader::SliceReader;
let build = |perms: PermSet| {
let mut db = crate::DictionaryBuilder::new();
let mut triples = Vec::new();
for i in 0..60u32 {
let (s, p, o) = (
format!("<http://ex/s{}>", i % 11),
format!("<http://ex/p{}>", i % 3),
format!("<http://ex/o{}>", i % 7),
);
db.observe(&s, &p, &o);
triples.push((s, p, o));
}
let dict = db.build();
let ids: Vec<(u32, u32, u32)> = triples
.iter()
.map(|(s, p, o)| dict.encode(s, p, o).unwrap())
.collect();
let def = GraphIndexBuilder::from_triples(ids.clone())
.with_tile_budget(96)
.with_perms(perms)
.build();
let named = GraphIndexBuilder::from_triples(ids)
.with_tile_budget(96)
.with_perms(perms)
.build();
write_dataset(
&dict,
&def,
&[("<http://ex/g>".to_string(), named)],
true,
&[],
0,
)
};
let six = build(PermSet::ALL);
let three = build(PermSet::CORE);
assert!(three.len() < six.len());
assert_eq!(Header::from_bytes(&three).unwrap().perms, PermSet::CORE);
assert_eq!(Header::from_bytes(&six).unwrap().perms, PermSet::ALL);
let shapes: Vec<(Option<&str>, Option<&str>, Option<&str>)> = vec![
(None, None, None),
(Some("<http://ex/s3>"), None, None),
(None, Some("<http://ex/p1>"), None),
(None, None, Some("<http://ex/o5>")),
(Some("<http://ex/s3>"), Some("<http://ex/p1>"), None),
(Some("<http://ex/s3>"), None, Some("<http://ex/o5>")),
(None, Some("<http://ex/p1>"), Some("<http://ex/o5>")),
(
Some("<http://ex/s3>"),
Some("<http://ex/p1>"),
Some("<http://ex/o5>"),
),
];
for (image, tag) in [(&six, "six"), (&three, "three")] {
let resident = Rete::open(image).unwrap();
let leaked: &'static [u8] = Box::leak(image.clone().into_boxed_slice());
let ranged = Rete::open_ranged(&SliceReader::new(leaked)).unwrap();
let lazy = Rete::open_ranged_lazy(SliceReader::new(leaked)).unwrap();
for (s, p, o) in &shapes {
let want: Vec<_> = {
let mut v = Rete::open(&six).unwrap().query(*s, *p, *o);
v.sort();
v
};
for (got, path) in [
(resident.query(*s, *p, *o), "resident"),
(ranged.query(*s, *p, *o), "ranged"),
(lazy.query(*s, *p, *o), "lazy"),
] {
let mut got = got;
got.sort();
assert_eq!(got, want, "{tag}/{path} disagreed on {:?}", (s, p, o));
}
let mut g = resident.query_in_graph(Some("<http://ex/g>"), *s, *p, *o);
g.sort();
assert_eq!(g, want, "{tag}/named disagreed on {:?}", (s, p, o));
}
}
let leaked: &'static [u8] = Box::leak(three.clone().into_boxed_slice());
let reader = SliceReader::new(leaked);
let routed = Rete::query_ranged(&reader, None, None, Some("<http://ex/o5>")).unwrap();
let mut want = Rete::open(&six)
.unwrap()
.query(None, None, Some("<http://ex/o5>"));
want.sort();
let mut got = routed;
got.sort();
assert_eq!(got, want, "routed single-pattern read on a lean file");
}
#[test]
fn read_coalesced_merges_within_gap_and_splits_beyond() {
use crate::reader::{CountingReader, SliceReader};
let bytes = vec![0u8; 4096];
let ranges = [
ByteRange { offset: 0, len: 16 },
ByteRange {
offset: 48,
len: 16,
},
ByteRange {
offset: 1088,
len: 16,
},
];
let r = CountingReader::new(SliceReader::new(&bytes));
let out = read_coalesced(&r, &ranges, 16).unwrap();
assert_eq!(out.len(), 3);
assert_eq!(r.requests(), 3);
let r = CountingReader::new(SliceReader::new(&bytes));
read_coalesced(&r, &ranges, 64).unwrap();
assert_eq!(r.requests(), 2);
let r = CountingReader::new(SliceReader::new(&bytes));
read_coalesced(&r, &ranges, 4096).unwrap();
assert_eq!(r.requests(), 1);
}
fn build_image() -> Vec<u8> {
let triples = [
("Alice", "knows", "Bob"),
("Bob", "knows", "Carol"),
("Alice", "age", "30"),
];
let mut db = DictionaryBuilder::new();
for (s, p, o) in triples {
db.observe(s, p, o);
}
let dict = db.build();
let mut ib = GraphIndexBuilder::new();
for (s, p, o) in triples {
ib.push(dict.encode(s, p, o).unwrap());
}
let index = ib.build();
let (meta, levels) = build_pyramid_meta(&dict, &triples_ids(&dict), DEFAULT_TILE_BUDGET);
write_file(&dict, &index, false, &meta, levels)
}
fn triples_ids(dict: &Dictionary) -> Vec<(u32, u32, u32)> {
[
("Alice", "knows", "Bob"),
("Bob", "knows", "Carol"),
("Alice", "age", "30"),
]
.iter()
.map(|(s, p, o)| dict.encode(s, p, o).unwrap())
.collect()
}
#[test]
fn file_round_trips_header_and_counts() {
let bytes = build_image();
let rete = Rete::open(&bytes).unwrap();
assert_eq!(rete.header().quad_count, 3);
assert!(rete.header().term_count >= 5);
let expected_codec = writer_codec();
assert_eq!(rete.header().dict_codec, expected_codec);
assert_eq!(rete.header().block_codec, expected_codec);
assert_eq!(&bytes[bytes.len() - 4..], &MAGIC); }
#[test]
fn multi_tile_file_round_trips_and_routes() {
let triples: Vec<(String, String, String)> = (0..200)
.map(|i| {
(
format!("<http://ex/s/{i}>"),
format!("<http://ex/p/{}>", i % 5),
format!("<http://ex/o/{}>", i % 23),
)
})
.collect();
let mut db = DictionaryBuilder::new();
for (s, p, o) in &triples {
db.observe(s, p, o);
}
let dict = db.build();
let mut ib = GraphIndexBuilder::new().with_tile_budget(64);
for (s, p, o) in &triples {
ib.push(dict.encode(s, p, o).unwrap());
}
let index = ib.build();
assert!(
index.tile_sections()[0].len() > 3,
"tiny budget must force many tiles"
);
let bytes = write_file(&dict, &index, false, &[], 0);
let rete = Rete::open(&bytes).unwrap();
assert_eq!(rete.header().version, crate::header::CURRENT_FORMAT_VERSION);
assert_eq!(rete.query(None, None, None).len(), 200);
assert_eq!(rete.query(Some("<http://ex/s/7>"), None, None).len(), 1);
assert_eq!(
rete.query(None, Some("<http://ex/p/3>"), None).len(),
40,
"predicate extent spans tiles"
);
assert_eq!(
rete.query(None, None, Some("<http://ex/o/22>")).len(),
8 );
use crate::reader::SliceReader;
let reader = SliceReader::new(&bytes);
let routed = Rete::query_ranged(&reader, Some("<http://ex/s/7>"), None, None).unwrap();
assert_eq!(routed.len(), 1);
let routed = Rete::query_ranged(&reader, None, Some("<http://ex/p/3>"), None).unwrap();
assert_eq!(routed.len(), 40);
let routed = Rete::query_ranged(&reader, None, None, Some("<http://ex/o/22>")).unwrap();
assert_eq!(routed.len(), 8);
}
#[test]
fn tile_directory_offsets_survive_past_4gib() {
let mut dir = Vec::new();
write_uvarint(&mut dir, 2); write_uvarint(&mut dir, 5); write_uvarint(&mut dir, 0); write_uvarint(&mut dir, 3 << 30); write_uvarint(&mut dir, 1); write_uvarint(&mut dir, 0);
write_uvarint(&mut dir, 2 << 30); let total = dir.len() as u64 + (3u64 << 30) + (2u64 << 30) + 64;
let entries = parse_tile_directory(&dir, total).unwrap();
assert_eq!(entries.len(), 2);
assert_eq!(entries[1].start, dir.len() as u64 + (3u64 << 30));
assert!(
entries[1].end > u32::MAX as u64,
"tail tile sits past 4 GiB"
);
assert!(parse_tile_directory(&dir, 1 << 20).is_err());
}
#[test]
fn tile_synopsis_trailer_round_trips() {
let mut ib = GraphIndexBuilder::new().with_tile_budget(64);
for i in 0..200u32 {
ib.push((i, i % 7, i % 13));
}
let index = ib.build();
let tile_count = index.tile_sections()[0].len();
assert!(tile_count > 3, "tiny budget forces many tiles");
let payload = encode_tiled_section(&index, 0, CODEC_NONE);
let dir = parse_tile_directory(&payload, payload.len() as u64).unwrap();
assert_eq!(dir.len(), tile_count);
let trailer_start = dir.iter().map(|e| e.end).max().unwrap();
assert!(
trailer_start < payload.len() as u64,
"a trailer follows the tiles"
);
for e in &dir {
assert!(
e.end <= payload.len() as u64,
"tiles still located within the payload"
);
}
let syn = parse_tile_synopsis(&payload, trailer_start as usize, dir.len()).unwrap();
for (e, (min_b, max_b, min_c, max_c)) in dir.iter().zip(syn) {
let block = decompress(CODEC_NONE, &payload[e.start as usize..e.end as usize]).unwrap();
let z = *crate::triples::TripleBlock::parse(&block).unwrap().zone();
assert_eq!(
(min_b, max_b, min_c, max_c),
(z.min_b, z.max_b, z.min_c, z.max_c),
"synopsis equals the tile's own zone"
);
}
}
#[test]
fn tile_synopsis_lazy_matches_reference_every_shape() {
use crate::reader::{CountingReader, SliceReader};
let triples: Vec<(String, String, String)> = (0..200u32)
.map(|i| {
(
format!("<http://ex/s/{i:04}>"),
format!("<http://ex/p/{}>", i % 7),
format!("<http://ex/o/{:04}>", i % 13),
)
})
.collect();
let mut db = DictionaryBuilder::new();
for (s, p, o) in &triples {
db.observe(s, p, o);
}
let dict = db.build();
let mut ib = GraphIndexBuilder::new().with_tile_budget(64);
for (s, p, o) in &triples {
ib.push(dict.encode(s, p, o).unwrap());
}
let bytes = write_file(&dict, &ib.build(), false, &[], 0);
let eager = Rete::open(&bytes).unwrap();
assert!(
eager.header().has_tile_synopsis(),
"new files set the synopsis flag"
);
let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
let lazy = Rete::open_ranged_lazy(reader).unwrap();
let brute = |s: Option<&str>, p: Option<&str>, o: Option<&str>| {
let mut v: Vec<(String, String, String)> = triples
.iter()
.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)
})
.cloned()
.collect();
v.sort();
v
};
let sv = [
None,
Some("<http://ex/s/0007>"),
Some("<http://ex/s/0130>"),
Some("<http://ex/s/9999>"),
];
let pv = [
None,
Some("<http://ex/p/3>"),
Some("<http://ex/p/6>"),
Some("<http://ex/p/999>"),
];
let ov = [
None,
Some("<http://ex/o/0000>"),
Some("<http://ex/o/0012>"),
Some("<http://ex/o/9999>"),
];
for &s in &sv {
for &p in &pv {
for &o in &ov {
let mut e = eager.query(s, p, o);
e.sort();
let mut l = lazy.query(s, p, o);
l.sort();
let r = brute(s, p, o);
assert_eq!(e, r, "eager {s:?} {p:?} {o:?}");
assert_eq!(l, r, "lazy {s:?} {p:?} {o:?} — synopsis over-pruned");
}
}
}
assert!(!lazy.index_incomplete(), "no lazy fetch failed");
}
#[test]
fn polyglot_offset_reads_lazily() {
use crate::reader::{
detect_polyglot_base, CountingReader, OffsetReader, SliceReader, POLYGLOT_DIGITS,
POLYGLOT_MARKER,
};
let image = build_image();
let mut shell = Vec::new();
shell.extend_from_slice(b"<!DOCTYPE html><html><head><!--");
shell.extend_from_slice(POLYGLOT_MARKER);
let digits_at = shell.len();
shell.extend_from_slice(&[b'0'; POLYGLOT_DIGITS]); shell.extend_from_slice(b"--></head><body>a web page</body></html>\n");
shell.resize(50_000, b' '); let base = shell.len() as u64;
let digits = format!("{base:0width$}", width = POLYGLOT_DIGITS);
shell[digits_at..digits_at + POLYGLOT_DIGITS].copy_from_slice(digits.as_bytes());
let mut poly = shell;
poly.extend_from_slice(&image);
assert_ne!(
&poly[0..4],
b"RETE",
"polyglot must not start with the magic"
);
let detected = detect_polyglot_base(&poly[..crate::header::HEADER_LEN]).unwrap();
assert_eq!(detected, base);
let leaked: &'static [u8] = Box::leak(poly.into_boxed_slice());
let counting = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
let lazy = Rete::open_ranged_lazy(OffsetReader::new(counting.clone(), detected)).unwrap();
let plain = Rete::open(&image).unwrap();
let mut want = plain.query(None, None, None);
want.sort();
let mut got = lazy.query(None, None, None);
got.sort();
assert_eq!(
got, want,
"polyglot lazy query differs from the plain .rete"
);
assert!(!got.is_empty(), "expected some triples");
assert!(!lazy.index_incomplete(), "a lazy fetch failed");
let read = counting.bytes_read();
eprintln!(
"polyglot lazy read: HTML shell {base} B + .rete image {} B; \
the query touched only {read} B (never the shell).",
image.len()
);
assert!(
read < base,
"read {read} bytes; the HTML shell alone is {base} — the reader \
touched the prefix instead of only the embedded .rete"
);
}
#[cfg(test)]
fn build_text_indexed(triples: &[(String, String, String)]) -> Vec<u8> {
let mut db = DictionaryBuilder::new();
for (s, p, o) in triples {
db.observe(s, p, o);
}
let dict = db.build();
let mut ib = GraphIndexBuilder::new().with_tile_budget(64);
let mut id_triples: Vec<(u32, u32, u32)> = Vec::with_capacity(triples.len());
for (s, p, o) in triples {
let t = dict.encode(s, p, o).unwrap();
ib.push(t);
id_triples.push(t);
}
let index = ib.build();
let text_index = compute_text_index(&dict, &id_triples);
assert!(
!text_index.is_empty(),
"literals should produce a text index"
);
write_dataset_with_metadata(&dict, &index, &[], false, &[], 0, &[], &text_index)
}
#[test]
fn text_index_eager_matches_brute_force() {
let triples: Vec<(String, String, String)> = vec![
(
"<http://ex/s0>",
"<http://ex/label>",
"\"alpha glucose phosphate\"",
),
("<http://ex/s1>", "<http://ex/label>", "\"beta Glucose\""),
("<http://ex/s2>", "<http://ex/label>", "\"gamma fructose\""),
(
"<http://ex/s3>",
"<http://ex/note>",
"\"einstein relativity\"",
),
(
"<http://ex/s4>",
"<http://ex/ref>",
"<http://ex/not-a-literal>",
),
]
.into_iter()
.map(|(s, p, o)| (s.to_string(), p.to_string(), o.to_string()))
.collect();
let bytes = build_text_indexed(&triples);
let rete = Rete::open(&bytes).unwrap();
assert!(rete.has_text_index());
let brute = |words: &[&str]| -> Vec<String> {
let mut v: Vec<String> = triples
.iter()
.filter(|(_, _, o)| {
crate::terms::is_literal(o)
&& words.iter().all(|w| {
let wl = w.to_lowercase();
crate::terms::literal_lexical(o)
.unwrap()
.split(|c: char| !c.is_alphanumeric())
.any(|t| t.to_lowercase() == wl)
})
})
.map(|(s, _, _)| s.clone())
.collect();
v.sort();
v.dedup();
v
};
let mut got = rete.text_search(&["glucose"], None, 0);
got.sort();
assert_eq!(got, brute(&["glucose"]), "case-insensitive single word");
let mut got = rete.text_search(&["glucose", "phosphate"], None, 0);
got.sort();
assert_eq!(got, brute(&["glucose", "phosphate"]));
assert!(rete.text_search(&["zzznope"], None, 0).is_empty());
let got = rete.text_search(&[], Some("ein"), 0);
assert_eq!(got, vec!["<http://ex/s3>".to_string()]);
let mut db = DictionaryBuilder::new();
for (s, p, o) in &triples {
db.observe(s, p, o);
}
let dict = db.build();
let mut ib = GraphIndexBuilder::new();
for (s, p, o) in &triples {
ib.push(dict.encode(s, p, o).unwrap());
}
let plain = write_dataset(&dict, &ib.build(), &[], false, &[], 0);
let plain_rete = Rete::open(&plain).unwrap();
assert!(!plain_rete.has_text_index());
assert!(plain_rete.text_search(&["glucose"], None, 0).is_empty());
}
#[test]
fn text_index_lazy_faults_only_queried_postings() {
use crate::reader::{CountingReader, SliceReader};
let mut triples: Vec<(String, String, String)> = (0..300u32)
.map(|i| {
(
format!("<http://ex/s/{i:04}>"),
"<http://ex/label>".to_string(),
format!("\"common word number {i}\""),
)
})
.collect();
for i in [3u32, 77, 250] {
triples.push((
format!("<http://ex/s/{i:04}>"),
"<http://ex/tag>".to_string(),
"\"raretoken\"".to_string(),
));
}
let bytes = build_text_indexed(&triples);
let eager = Rete::open(&bytes).unwrap();
let mut want = eager.text_search(&["raretoken"], None, 0);
want.sort();
assert_eq!(want.len(), 3);
let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
let lazy = Rete::open_ranged_lazy(reader.clone()).unwrap();
let before = reader.bytes_read();
let mut got = lazy.text_search(&["raretoken"], None, 0);
got.sort();
assert_eq!(got, want, "lazy search matches eager");
let pulled = reader.bytes_read() - before;
let ti_len = eager.header().text_index_len;
assert!(
pulled < ti_len,
"search pulled {pulled} B but the section is {ti_len} B — faulted too much"
);
assert!(!lazy.index_incomplete());
}
#[test]
fn search_view_matches_full_open_for_far_fewer_bytes() {
use crate::reader::{CountingReader, SliceReader};
let filler = "lorem ipsum dolor sit amet consectetur adipiscing elit ".repeat(20);
let mut triples: Vec<(String, String, String)> = (0..300u32)
.map(|i| {
(
format!("<http://ex/s/{i:04}>"),
"<http://ex/abstract>".to_string(),
format!("\"{filler} number {i}\""),
)
})
.collect();
for i in [3u32, 77, 250] {
triples.push((
format!("<http://ex/s/{i:04}>"),
"<http://ex/tag>".to_string(),
"\"raretoken\"".to_string(),
));
}
let bytes = build_text_indexed(&triples);
let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
let full_reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
let full = Rete::open_ranged_lazy(full_reader.clone()).unwrap();
let mut want = full.text_search(&["raretoken"], None, 0);
want.sort();
assert_eq!(want.len(), 3);
let search_reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
let view = SearchView::open_ranged(search_reader.clone()).unwrap();
assert!(view.has_text_index());
let mut got = view.text_search(&["raretoken"], None, 0);
got.sort();
assert_eq!(got, want, "search view matches the full open");
assert!(
search_reader.bytes_read() < full_reader.bytes_read(),
"search read {} B, full open read {} B — the narrow open bought nothing",
search_reader.bytes_read(),
full_reader.bytes_read()
);
let mut prefix_hits = view.prefix_search("", 5);
prefix_hits.sort();
assert!(!prefix_hits.is_empty() || !view.has_pyramid());
}
#[test]
fn dict_chunk_directory_probe_costs_about_one_directory() {
use crate::reader::{CountingReader, SliceReader};
let head = "x".repeat(3000);
let triples: Vec<(String, String, String)> = (0..2000u32)
.map(|i| {
let tail = char::from(b'a' + (i % 26) as u8).to_string().repeat(1000);
(
format!("<http://ex/s/{i:04}>"),
"<http://ex/abstract>".to_string(),
format!("\"{head}{i:04}{tail}\""),
)
})
.collect();
let mut db = DictionaryBuilder::new();
for (s, p, o) in &triples {
db.observe(s, p, o);
}
let dict = db.build();
let mut ib = GraphIndexBuilder::new();
for (s, p, o) in &triples {
ib.push(dict.encode(s, p, o).unwrap());
}
let bytes = write_file(&dict, &ib.build(), false, &[], 0);
let reader = CountingReader::new(SliceReader::new(&bytes));
let header = Header::from_bytes(&reader.read_at(0, HEADER_LEN as u64).unwrap()).unwrap();
let section = locate_container_section_ranged(
&reader,
header.dictionary_offset,
header.dictionary_len,
2,
4,
)
.unwrap();
let before = reader.bytes_read();
let (_meta, entries) = read_dict_dir_ranged(&reader, section).unwrap();
let spent = reader.bytes_read() - before;
let dir_end = entries[0].start;
let head = reader.read_at(section.offset, 64).unwrap();
let (header_len, n0) = read_uvarint(&head).unwrap();
let dir_start = n0 as u64 + header_len;
let dir_len = dir_end - dir_start;
let dir_total = section.len - dir_start;
assert!(
dir_start + dir_len > 8192,
"directory fits the header prefix ({dir_len} B) — the probe never runs"
);
let mut old = 0u64;
let mut p = 4096u64.min(dir_total).max(1);
loop {
let got = p.min(dir_total);
old += got;
if got >= dir_len || p >= dir_total {
break;
}
p = p.saturating_mul(2).min(dir_total);
}
assert!(
spent <= dir_len * 2,
"probe read {spent} B for a {dir_len} B directory — past the append-only bound"
);
assert!(
spent < old,
"probe read {spent} B; the re-reading loop it replaced spent {old} B"
);
}
fn verbatim_keyed_twin(payload: &[u8], codec: u8) -> Vec<u8> {
let (_meta, entries) = parse_chunked_dict_dir(payload, payload.len() as u64).unwrap();
let (header_len, n0) = read_uvarint(payload).unwrap();
let mut out = Vec::new();
write_uvarint(&mut out, header_len);
out.extend_from_slice(&payload[n0..n0 + header_len as usize]);
write_uvarint(&mut out, entries.len() as u64);
let mut prev_run = 0usize;
for e in &entries {
let body = decompress(codec, &payload[e.start as usize..e.end as usize]).unwrap();
let first = crate::dict::run_first_term(&body, 0).unwrap();
write_uvarint(&mut out, (e.first_run - prev_run) as u64);
write_uvarint(&mut out, first.len() as u64);
out.extend_from_slice(&first);
write_uvarint(&mut out, e.end - e.start);
prev_run = e.first_run;
}
for e in &entries {
out.extend_from_slice(&payload[e.start as usize..e.end as usize]);
}
out
}
fn chunk_bounds_of(payload: &[u8], codec: u8) -> Vec<(Vec<u8>, Vec<u8>)> {
let (_meta, entries) = parse_chunked_dict_dir(payload, payload.len() as u64).unwrap();
entries
.iter()
.map(|e| {
let body = decompress(codec, &payload[e.start as usize..e.end as usize]).unwrap();
(
crate::dict::run_first_term(&body, 0).unwrap(),
crate::dict::run_last_term(&body, 0, body.len()).unwrap(),
)
})
.collect()
}
fn adversarial_terms() -> Vec<String> {
let mut terms: Vec<String> = Vec::new();
for i in 0..6_000 {
terms.push(format!("<http://example.org/work/{i:06}>"));
}
let shared = "X".repeat(900);
for i in 0..4_000 {
terms.push(format!("\"{shared}-{i:05}-tail\""));
}
let deep = "Y".repeat(3_000); for i in 0..120 {
terms.push(format!("\"{deep}{i:03}\""));
}
for i in 0..6 {
terms.push(format!("\"BIG-{i:02}-{}\"", "Z".repeat(90_000)));
}
for i in 0..800 {
terms.push(format!(
"\"h\u{e9}llo-{}-\u{1F600}{i:04}\"",
"\u{e9}".repeat(40)
));
}
for i in 0..2_000 {
terms.push(format!("\"{i:05}-{}\"", "T".repeat(1_800)));
}
terms.sort();
terms.dedup();
terms
}
fn chunked_section_of(terms: &[String]) -> Vec<u8> {
let mut b = crate::dict::DictSectionBuilder::new().with_restart_interval(16);
for t in terms {
b.push(t.clone());
}
b.build()
}
#[test]
fn chunk_directory_keys_are_separators_not_terms() {
let terms = adversarial_terms();
let raw = chunked_section_of(&terms);
let payload = encode_chunked_dict_section(&raw, CODEC_NONE);
let (_meta, entries) = parse_chunked_dict_dir(&payload, payload.len() as u64).unwrap();
let bounds = chunk_bounds_of(&payload, CODEC_NONE);
assert!(
entries.len() > 20,
"need a many-chunk section, got {}",
entries.len()
);
assert!(entries[0].key.is_empty(), "chunk 0 needs no separator");
let mut key_bytes = 0usize;
let mut term_bytes = 0usize;
let mut differ = 0usize;
for (i, e) in entries.iter().enumerate() {
key_bytes += e.key.len();
term_bytes += bounds[i].0.len();
if e.key != bounds[i].0 {
differ += 1;
}
assert!(
e.key <= bounds[i].0,
"chunk {i}: key must be <= its own first term"
);
if i > 0 {
assert!(
e.key.as_slice() > bounds[i - 1].1.as_slice(),
"chunk {i}: key must be > the previous chunk's last term"
);
assert!(
e.key > entries[i - 1].key,
"chunk {i}: keys must ascend for partition_point"
);
assert_eq!(
e.key,
crate::dict::shortest_separator(&bounds[i - 1].1, &bounds[i].0),
"chunk {i}: key is not the SHORTEST separator"
);
}
}
assert!(
differ > 0,
"every key equalled its first term — the writer stores terms again"
);
assert!(
key_bytes * 4 < term_bytes,
"separators bought almost nothing: {key_bytes} B vs {term_bytes} B of first terms"
);
eprintln!(
"chunks={} first-term keys={term_bytes} B separator keys={key_bytes} B ({:.1}x)",
entries.len(),
term_bytes as f64 / key_bytes.max(1) as f64
);
}
#[test]
fn separator_keys_route_identically_to_verbatim_first_terms() {
let terms = adversarial_terms();
let raw = chunked_section_of(&terms);
let sep_payload = encode_chunked_dict_section(&raw, CODEC_NONE);
let ver_payload = verbatim_keyed_twin(&sep_payload, CODEC_NONE);
let (_ms, es) = parse_chunked_dict_dir(&sep_payload, sep_payload.len() as u64).unwrap();
let (_mv, ev) = parse_chunked_dict_dir(&ver_payload, ver_payload.len() as u64).unwrap();
assert_eq!(es.len(), ev.len());
for (a, b) in es.iter().zip(ev.iter()) {
assert_eq!(a.first_run, b.first_run);
assert_eq!(
&sep_payload[a.start as usize..a.end as usize],
&ver_payload[b.start as usize..b.end as usize],
"chunk bodies must be identical"
);
}
assert!(
sep_payload.len() < ver_payload.len(),
"separator payload is not smaller"
);
let sec_s = decode_chunked_dict_section(
&sep_payload,
CODEC_NONE,
crate::chunk_cache::ChunkCache::unlimited_arc(),
0,
)
.unwrap();
let sec_v = decode_chunked_dict_section(
&ver_payload,
CODEC_NONE,
crate::chunk_cache::ChunkCache::unlimited_arc(),
0,
)
.unwrap();
for (i, t) in terms.iter().enumerate() {
let want = Some(i as u32 + 1);
assert_eq!(sec_v.id(t), want, "verbatim id() lost a term");
assert_eq!(sec_s.id(t), want, "separator id() lost a term");
}
for i in 0..terms.len() as u32 {
assert_eq!(sec_s.term(i + 1), sec_v.term(i + 1));
assert_eq!(
sec_s.term(i + 1).as_deref(),
Some(terms[i as usize].as_str())
);
}
let bounds = chunk_bounds_of(&sep_payload, CODEC_NONE);
let mut probes: Vec<String> = Vec::new();
let push = |bytes: &[u8], probes: &mut Vec<String>| {
if let Ok(s) = std::str::from_utf8(bytes) {
probes.push(s.to_string());
}
};
for (first, last) in &bounds {
for base in [first, last] {
push(base, &mut probes); let mut after = base.clone();
after.push(0x01); push(&after, &mut probes);
let mut far = base.clone();
far.push(b'~');
push(&far, &mut probes);
if !base.is_empty() {
push(&base[..base.len() - 1], &mut probes); let mut down = base.clone();
*down.last_mut().unwrap() = down.last().unwrap().wrapping_sub(1);
push(&down, &mut probes);
let mut up = base.clone();
*up.last_mut().unwrap() = up.last().unwrap().wrapping_add(1);
push(&up, &mut probes);
}
}
}
for e in &es {
push(&e.key, &mut probes);
let mut plus = e.key.clone();
plus.push(0);
push(&plus, &mut probes);
}
probes.push(String::new()); probes.push("\u{1}".to_string()); probes.push("\u{10FFFF}\u{10FFFF}".to_string()); probes.push(terms.first().unwrap().clone());
probes.push(terms.last().unwrap().clone());
let mut hits = 0usize;
for p in &probes {
let (a, b) = (sec_v.id(p), sec_s.id(p));
assert_eq!(
a,
b,
"probe diverged: {:?}",
p.get(..p.len().min(48)).unwrap_or(p.as_str())
);
hits += usize::from(a.is_some());
}
assert!(
probes.len() > 500 && hits > 50,
"probe set is too thin: {} probes, {hits} of them present",
probes.len()
);
eprintln!(
"{} boundary probes, {hits} present, all identical",
probes.len()
);
}
mod separator_props {
use super::*;
use proptest::prelude::*;
fn term_set() -> impl Strategy<Value = Vec<String>> {
prop::collection::vec(("[a-c]{1,6}", 0u8..3, 300usize..1200), 60..250).prop_map(
|specs| {
let mut terms: Vec<String> = specs
.into_iter()
.enumerate()
.map(|(i, (core, kind, pad))| {
let block = "z".repeat(pad);
match kind {
0 => format!("<http://ex/{core}/{i:05}>"),
1 => format!("\"{block}{core}{i:05}\""),
_ => format!("\"{core}{i:05}{block}\""),
}
})
.collect();
terms.sort();
terms.dedup();
terms
},
)
}
proptest! {
#[test]
fn prop_separator_keyed_section_answers_like_verbatim(
terms in term_set(),
extra in prop::collection::vec(any::<String>(), 0..8),
) {
let raw = chunked_section_of(&terms);
let sep = encode_chunked_dict_section(&raw, CODEC_NONE);
let ver = verbatim_keyed_twin(&sep, CODEC_NONE);
prop_assert!(sep.len() <= ver.len());
let (_m, es) = parse_chunked_dict_dir(&sep, sep.len() as u64).unwrap();
let bounds = chunk_bounds_of(&sep, CODEC_NONE);
prop_assert!(es[0].key.is_empty());
for i in 1..es.len() {
prop_assert!(es[i].key.as_slice() > bounds[i - 1].1.as_slice());
prop_assert!(es[i].key <= bounds[i].0);
prop_assert!(es[i].key > es[i - 1].key);
}
let sec_s = decode_chunked_dict_section(
&sep,
CODEC_NONE,
crate::chunk_cache::ChunkCache::unlimited_arc(),
0,
)
.unwrap();
let sec_v = decode_chunked_dict_section(
&ver,
CODEC_NONE,
crate::chunk_cache::ChunkCache::unlimited_arc(),
0,
)
.unwrap();
for (i, t) in terms.iter().enumerate() {
prop_assert_eq!(sec_s.id(t), Some(i as u32 + 1));
prop_assert_eq!(sec_v.id(t), Some(i as u32 + 1));
prop_assert_eq!(sec_s.term(i as u32 + 1), sec_v.term(i as u32 + 1));
}
let mut probes: Vec<String> = extra;
probes.push(String::new());
for (first, last) in &bounds {
for base in [first, last] {
if let Ok(s) = std::str::from_utf8(base) {
probes.push(s.to_string());
probes.push(format!("{s}\u{1}"));
probes.push(s[..s.len() - s.chars().next_back()
.map(char::len_utf8).unwrap_or(0)].to_string());
}
}
}
for e in &es {
if let Ok(s) = std::str::from_utf8(&e.key) {
probes.push(s.to_string());
}
}
for p in &probes {
prop_assert_eq!(sec_s.id(p), sec_v.id(p), "probe diverged: {:?}", p);
}
}
}
}
#[test]
fn text_index_is_tamper_evident_and_verifies() {
let triples: Vec<(String, String, String)> = vec![(
"<http://ex/s0>".to_string(),
"<http://ex/label>".to_string(),
"\"alpha glucose phosphate\"".to_string(),
)];
let bytes = build_text_indexed(&triples);
let header = Rete::open(&bytes).unwrap().header().clone();
assert!(header.text_index_len > 0);
assert!(verify(&bytes).unwrap(), "a text-indexed build must verify");
let mut tampered = bytes.clone();
tampered[header.text_index_offset as usize] ^= 0xff;
assert!(
!verify(&tampered).unwrap(),
"tampering with the text index must break verify()"
);
}
#[test]
fn synopsis_cuts_remote_fetch_bytes() {
use crate::header::FLAG_TILE_SYNOPSIS;
use crate::reader::{CountingReader, SliceReader};
let triples: Vec<(String, String, String)> = (0..400u32)
.map(|i| {
(
format!("<http://ex/s/{i:04}>"),
"<http://ex/p>".to_string(),
format!("<http://ex/o/{i:04}>"),
)
})
.collect();
let mut db = DictionaryBuilder::new();
for (s, p, o) in &triples {
db.observe(s, p, o);
}
let dict = db.build();
let mut ib = GraphIndexBuilder::new().with_tile_budget(64);
for (s, p, o) in &triples {
ib.push(dict.encode(s, p, o).unwrap());
}
let bytes = write_file(&dict, &ib.build(), false, &[], 0);
let q = (Some("<http://ex/s/0395>"), None, Some("<http://ex/o/0005>"));
let query_bytes = |image: &[u8]| -> (u64, usize) {
let leaked: &'static [u8] = Box::leak(image.to_vec().into_boxed_slice());
let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
let rete = Rete::open_ranged_lazy(reader.clone()).unwrap();
let before = reader.bytes_read(); let n = rete.query(q.0, q.1, q.2).len();
assert!(!rete.index_incomplete());
(reader.bytes_read() - before, n)
};
let (on_bytes, on_n) = query_bytes(&bytes);
let mut off = bytes.clone();
off[5] &= !FLAG_TILE_SYNOPSIS;
let (off_bytes, off_n) = query_bytes(&off);
assert_eq!(on_n, 0, "the pair never co-occurs");
assert_eq!(off_n, 0, "same answer without the synopsis");
assert!(
on_bytes < off_bytes,
"synopsis skips the routed tile fetch: {on_bytes} < {off_bytes}"
);
}
#[test]
fn double_bound_object_join_eager_matches_lazy() {
use crate::reader::SliceReader;
let occ = "<http://ex/occ>";
let phys = "<http://ex/physicist>";
let phil = "<http://ex/philosopher>";
let label = "<http://www.w3.org/2000/01/rdf-schema#label>";
let mut triples: Vec<(String, String, String)> = Vec::new();
for i in 0..20u32 {
triples.push((format!("<http://ex/p/{i:02}>"), occ.into(), phys.into()));
if i < 10 {
triples.push((format!("<http://ex/p/{i:02}>"), occ.into(), phil.into()));
}
triples.push((
format!("<http://ex/p/{i:02}>"),
label.into(),
format!("\"Name {i:02}\""),
));
}
let mut db = DictionaryBuilder::new();
for (s, p, o) in &triples {
db.observe(s, p, o);
}
let dict = db.build();
let mut ib = GraphIndexBuilder::new().with_tile_budget(16);
for (s, p, o) in &triples {
ib.push(dict.encode(s, p, o).unwrap());
}
let bytes = write_file(&dict, &ib.build(), false, &[], 0);
let q = "SELECT ?l WHERE { \
?p <http://ex/occ> <http://ex/physicist> ; \
<http://ex/occ> <http://ex/philosopher> ; \
<http://www.w3.org/2000/01/rdf-schema#label> ?l }";
let run = |rete: &Rete| -> Vec<String> {
let (_, sols) = crate::eval_sparql(rete, q).unwrap();
let mut v: Vec<String> = sols.iter().filter_map(|b| b.get("l").cloned()).collect();
v.sort();
v
};
let eager_rows = run(&Rete::open(&bytes).unwrap());
let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
let lazy = Rete::open_ranged_lazy(std::sync::Arc::new(SliceReader::new(leaked))).unwrap();
let lazy_rows = run(&lazy);
assert!(!lazy.index_incomplete());
assert_eq!(eager_rows.len(), 10, "the 10 physicist∩philosopher labels");
assert_eq!(eager_rows, lazy_rows, "eager and lazy must agree exactly");
}
#[test]
fn multi_chunk_dictionary_round_trips() {
let mut db = DictionaryBuilder::new();
let term = |i: u32| format!("<http://example.org/some/long/prefix/entity/{i:06}>");
for i in 0..6000u32 {
db.observe(&term(i), "<http://ex/p>", &term(i + 1));
}
let dict = db.build();
let mut ib = GraphIndexBuilder::new();
for i in 0..6000u32 {
ib.push(
dict.encode(&term(i), "<http://ex/p>", &term(i + 1))
.unwrap(),
);
}
let bytes = write_file(&dict, &ib.build(), false, &[], 0);
let rete = Rete::open(&bytes).unwrap();
let d = rete.dictionary();
assert_eq!(d.term_count(), dict.term_count());
for i in (0..6000).step_by(97).chain([0, 1, 5999, 6000]) {
let t = term(i);
let sid = dict.subject_id(&t);
assert_eq!(d.subject_id(&t), sid, "subject_id({t})");
if let Some(id) = sid {
assert_eq!(d.subject_term(id).as_deref(), Some(t.as_str()));
}
let oid = dict.object_id(&t);
assert_eq!(d.object_id(&t), oid, "object_id({t})");
}
assert_eq!(d.subject_id("<http://example.org/absent>"), None);
assert_eq!(d.predicate_id("<http://ex/p>"), Some(1));
assert_eq!(d.predicate_term(1).as_deref(), Some("<http://ex/p>"));
}
#[test]
#[cfg(feature = "compression")]
fn compression_shrinks_repetitive_data() {
let mut db = DictionaryBuilder::new();
let triples: Vec<(String, String, String)> = (0..500)
.map(|i| {
(
format!("<http://example.org/entity/{i}>"),
"<http://example.org/p/relatedTo>".to_string(),
format!("<http://example.org/entity/{}>", (i + 1) % 500),
)
})
.collect();
for (s, p, o) in &triples {
db.observe(s, p, o);
}
let dict = db.build();
let mut ib = GraphIndexBuilder::new();
for (s, p, o) in &triples {
ib.push(dict.encode(s, p, o).unwrap());
}
let bytes = write_file(&dict, &ib.build(), false, &[], 0);
let raw: usize = triples
.iter()
.map(|(s, p, o)| s.len() + p.len() + o.len())
.sum();
assert!(
bytes.len() < raw / 2,
"expected strong compression: file {} vs raw terms {raw}",
bytes.len()
);
let rete = Rete::open(&bytes).unwrap();
let r = rete.query(Some("<http://example.org/entity/0>"), None, None);
assert_eq!(r.len(), 1);
assert_eq!(r[0].2, "<http://example.org/entity/1>");
}
fn big_file_with_pyramid() -> Vec<u8> {
let triples: Vec<(String, String, String)> = (0..300)
.map(|i| {
(
format!("<http://ex/e{i}>"),
"<http://ex/next>".to_string(),
format!("<http://ex/e{}>", (i + 1) % 300),
)
})
.collect();
let mut db = DictionaryBuilder::new();
for (s, p, o) in &triples {
db.observe(s, p, o);
}
let dict = db.build();
let ids: Vec<_> = triples
.iter()
.map(|(s, p, o)| dict.encode(s, p, o).unwrap())
.collect();
let mut ib = GraphIndexBuilder::new();
for &t in &ids {
ib.push(t);
}
let (meta, levels) = build_pyramid_meta(&dict, &ids, DEFAULT_TILE_BUDGET);
write_file(&dict, &ib.build(), false, &meta, levels)
}
#[test]
fn ranged_open_is_minimal_and_correct() {
use crate::reader::{CountingReader, SliceReader};
let bytes = big_file_with_pyramid();
let full = CountingReader::new(SliceReader::new(&bytes));
let rete = Rete::open_ranged(&full).unwrap();
assert!(full.requests() <= 4, "requests = {}", full.requests());
assert_eq!(
rete.query(Some("<http://ex/e0>"), None, None)[0].2,
"<http://ex/e1>"
);
let summ_reader = CountingReader::new(SliceReader::new(&bytes));
let view = SummaryView::open_ranged(&summ_reader).unwrap().unwrap();
assert!(!view.summary.is_empty());
assert!(
summ_reader.bytes_read() < bytes.len() as u64,
"summary read {} of {} bytes",
summ_reader.bytes_read(),
bytes.len()
);
assert!(summ_reader.bytes_read() < full.bytes_read());
}
#[test]
fn content_hash_is_set_and_verifies() {
let bytes = build_image();
let rete = Rete::open(&bytes).unwrap();
assert_ne!(
rete.header().content_hash,
[0u8; 16],
"hash must be populated"
);
assert!(verify(&bytes).unwrap(), "freshly built file verifies");
assert_eq!(
Rete::open(&build_image()).unwrap().header().content_hash,
rete.header().content_hash
);
let mut tampered = bytes.clone();
let last = tampered.len() - 5; tampered[last] ^= 0xff;
assert!(!verify(&tampered).unwrap());
}
fn build_with_metadata(meta: &[u8]) -> Vec<u8> {
let triples = [
("Alice", "knows", "Bob"),
("Bob", "knows", "Carol"),
("Alice", "age", "30"),
];
let mut db = DictionaryBuilder::new();
for (s, p, o) in triples {
db.observe(s, p, o);
}
let dict = db.build();
let ids: Vec<_> = triples
.iter()
.map(|(s, p, o)| dict.encode(s, p, o).unwrap())
.collect();
let mut ib = GraphIndexBuilder::new();
for &t in &ids {
ib.push(t);
}
let (pmeta, levels) = build_pyramid_meta(&dict, &ids, DEFAULT_TILE_BUDGET);
write_dataset_with_metadata(&dict, &ib.build(), &[], false, &pmeta, levels, meta, &[])
}
#[test]
fn metadata_round_trips_and_shifts_offsets() {
let card = br#"{"title":"My Dataset"}"#;
let bytes = build_with_metadata(card);
let rete = Rete::open(&bytes).unwrap();
assert_eq!(rete.metadata(), Some(card.as_slice()));
let h = rete.header();
assert_eq!(h.metadata_offset, HEADER_LEN as u64);
assert_eq!(h.metadata_len, card.len() as u64);
assert_eq!(h.dictionary_offset, HEADER_LEN as u64 + card.len() as u64);
assert_eq!(
rete.query(Some("Bob"), Some("knows"), Some("Carol")).len(),
1
);
assert!(verify(&bytes).unwrap());
}
#[test]
fn empty_metadata_is_byte_identical_to_plain_writer() {
assert_eq!(
build_with_metadata(&[]),
build_image(),
"empty-metadata output must equal the plain writer byte-for-byte"
);
}
#[test]
fn replace_metadata_matches_a_direct_write() {
let first = br#"{"title":"My Dataset","queries":[1,2,3]}"#;
let bytes = build_with_metadata(first);
for target in [
br#"{"title":"My Dataset","queries":[1,2]}"#.as_slice(), br#"{"title":"My Dataset","queries":[1,2,3,4,5,6,7]}"#.as_slice(), b"".as_slice(), first.as_slice(), ] {
let spliced = replace_metadata(&bytes, target).unwrap();
assert_eq!(
spliced,
build_with_metadata(target),
"splicing {} bytes of metadata must equal a direct write",
target.len()
);
assert!(
verify(&spliced).unwrap(),
"the new hash covers the new card"
);
let rete = Rete::open(&spliced).unwrap();
assert_eq!(
rete.metadata(),
(!target.is_empty()).then_some(target),
"the new payload reads back verbatim"
);
assert_eq!(
rete.query(Some("Bob"), Some("knows"), Some("Carol")).len(),
1
);
}
}
#[test]
fn replace_metadata_keeps_build_info_adjacent() {
let with_info =
attach_build_info(&build_with_metadata(br#"{"a":1}"#), br#"{"builder":"x"}"#).unwrap();
let spliced = replace_metadata(&with_info, br#"{"a":1,"b":2}"#).unwrap();
let h = Header::from_bytes(&spliced).unwrap();
assert_eq!(h.metadata_offset, HEADER_LEN as u64);
assert_eq!(h.build_info_offset, HEADER_LEN as u64 + h.metadata_len);
assert_eq!(
read_build_info(&spliced).unwrap().as_deref(),
Some(br#"{"builder":"x"}"#.as_slice())
);
assert!(verify(&spliced).unwrap());
assert_eq!(
h.content_hash,
Header::from_bytes(&build_with_metadata(br#"{"a":1,"b":2}"#))
.unwrap()
.content_hash
);
}
#[test]
fn metadata_is_tamper_evident() {
let card = br#"{"title":"x"}"#;
let mut bytes = build_with_metadata(card);
assert!(verify(&bytes).unwrap());
bytes[HEADER_LEN + 2] ^= 0xff;
assert!(
!verify(&bytes).unwrap(),
"tampering with the card must break verify()"
);
}
#[test]
fn ranged_opens_do_not_fetch_metadata() {
use crate::reader::{CountingReader, SliceReader};
let card = vec![0xABu8; 512]; let bytes = build_with_metadata(&card);
let total = bytes.len() as u64;
let r = CountingReader::new(SliceReader::new(&bytes));
let rete = Rete::open_ranged(&r).unwrap();
assert!(
rete.metadata().is_none(),
"open_ranged must not load the card"
);
assert!(r.requests() <= 4, "requests = {}", r.requests());
assert!(
r.bytes_read() <= total - card.len() as u64,
"read {} of {} bytes; the {}-byte card must be skipped",
r.bytes_read(),
total,
card.len()
);
let rs = CountingReader::new(SliceReader::new(&bytes));
let view = SummaryView::open_ranged(&rs).unwrap().unwrap();
assert!(!view.summary.is_empty());
assert!(rs.bytes_read() <= total - card.len() as u64);
}
#[test]
fn metadata_ranged_fetches_only_header_and_card() {
use crate::reader::{CountingReader, SliceReader};
let card = vec![0xCDu8; 384];
let bytes = build_with_metadata(&card);
let r = CountingReader::new(SliceReader::new(&bytes));
let got = read_metadata_ranged(&r).unwrap().unwrap();
assert_eq!(got, card, "the card reads back verbatim");
assert_eq!(r.requests(), 2, "exactly header + metadata ranges");
assert_eq!(
r.bytes_read(),
HEADER_LEN as u64 + card.len() as u64,
"no dictionary/index/pyramid bytes are touched"
);
let plain = build_image();
let rp = CountingReader::new(SliceReader::new(&plain));
assert!(read_metadata_ranged(&rp).unwrap().is_none());
assert_eq!(rp.requests(), 1, "header only for a cardless file");
assert_eq!(rp.bytes_read(), HEADER_LEN as u64);
}
#[test]
fn build_info_attaches_outside_the_hash() {
let card = br#"{"title":"My Dataset"}"#;
let base = build_with_metadata(card);
let base_hash = Header::from_bytes(&base).unwrap().content_hash;
let info = br#"{"built_at":"2026-08-04T00:00:00Z","builder":"rete-cli 0.3.2"}"#;
let with = attach_build_info(&base, info).unwrap();
assert_eq!(read_build_info(&with).unwrap().unwrap(), info);
let h = Header::from_bytes(&with).unwrap();
assert_eq!(h.build_info_offset, HEADER_LEN as u64 + card.len() as u64);
assert_eq!(h.build_info_len, info.len() as u64);
assert_eq!(h.expected_file_len(), Some(with.len() as u64));
assert_eq!(h.content_hash, base_hash);
assert!(verify(&with).unwrap());
let other = attach_build_info(&base, br#"{"built_at":"1999-01-01T00:00:00Z"}"#).unwrap();
assert_eq!(Header::from_bytes(&other).unwrap().content_hash, base_hash);
let rete = Rete::open(&with).unwrap();
assert_eq!(
rete.query(Some("Bob"), Some("knows"), Some("Carol")).len(),
1
);
assert_eq!(rete.metadata(), Some(card.as_slice()));
let replaced = attach_build_info(&with, br#"{"builder":"other"}"#).unwrap();
assert_eq!(
read_build_info(&replaced).unwrap().unwrap(),
br#"{"builder":"other"}"#
);
assert!(verify(&replaced).unwrap());
let stripped = attach_build_info(&replaced, &[]).unwrap();
assert_eq!(stripped, base, "stripping build-info restores the image");
}
#[test]
fn build_info_attaches_on_a_metadata_free_file() {
let base = build_image();
let info = b"{\"builder\":\"x\"}";
let with = attach_build_info(&base, info).unwrap();
let h = Header::from_bytes(&with).unwrap();
assert_eq!(h.build_info_offset, HEADER_LEN as u64);
assert_eq!(h.metadata_len, 0);
assert_eq!(read_build_info(&with).unwrap().unwrap(), info);
assert!(verify(&with).unwrap());
assert_eq!(
h.content_hash,
Header::from_bytes(&base).unwrap().content_hash
);
Rete::open(&with).unwrap();
}
#[test]
fn a_plan_rebuilds_exactly_what_the_in_memory_splice_produces() {
let card = br#"{"title":"My Dataset"}"#;
let base = build_with_metadata(card);
let base_hash = Header::from_bytes(&base).unwrap().content_hash;
for start in [
base.clone(),
attach_build_info(&base, b"{\"a\":1}").unwrap(),
] {
for info in [
br#"{"builder":"rete-cli 0.3.2","query_costs":{}}"#.as_slice(), b"{}".as_slice(), b"".as_slice(), ] {
let plan = plan_build_info(&start, start.len() as u64, info.len() as u64).unwrap();
let mut streamed = Vec::new();
streamed.extend_from_slice(&plan.header);
streamed.extend_from_slice(&start[HEADER_LEN..plan.insert as usize]);
streamed.extend_from_slice(info);
streamed.extend_from_slice(&start[plan.tail_start as usize..]);
assert_eq!(streamed.len() as u64, plan.new_len);
assert_eq!(
streamed,
attach_build_info(&start, info).unwrap(),
"a {}-byte section streamed must equal the same section spliced",
info.len()
);
assert!(verify(&streamed).unwrap());
let h = Header::from_bytes(&streamed).unwrap();
assert_eq!(h.content_hash, base_hash);
assert_eq!(h.expected_file_len(), Some(plan.new_len));
let rete = Rete::open(&streamed).unwrap();
assert_eq!(rete.metadata(), Some(card.as_slice()));
assert_eq!(
rete.query(Some("Bob"), Some("knows"), Some("Carol")).len(),
1
);
}
}
}
#[test]
fn a_plan_refuses_a_header_that_overruns_the_file() {
let base = build_with_metadata(br#"{"title":"x"}"#);
assert!(plan_build_info(&base, 8, 16).is_err());
assert!(plan_build_info(&base[..HEADER_LEN], base.len() as u64, 16).is_ok());
}
#[test]
fn card_and_build_info_ranged_is_one_header_plus_one_range() {
use crate::reader::{CountingReader, SliceReader};
let card = vec![0xCDu8; 384];
let info = vec![0xEFu8; 200];
let bytes = attach_build_info(&build_with_metadata(&card), &info).unwrap();
let r = CountingReader::new(SliceReader::new(&bytes));
let (m, b) = read_card_and_build_info_ranged(&r).unwrap();
assert_eq!(m.unwrap(), card);
assert_eq!(b.unwrap(), info);
assert_eq!(r.requests(), 2, "header + one coalesced range");
assert_eq!(
r.bytes_read(),
HEADER_LEN as u64 + card.len() as u64 + info.len() as u64
);
let plain = build_with_metadata(&card);
let rp = CountingReader::new(SliceReader::new(&plain));
let (m, b) = read_card_and_build_info_ranged(&rp).unwrap();
assert_eq!(m.unwrap(), card);
assert!(b.is_none());
assert_eq!(rp.requests(), 2);
assert_eq!(
r.requests(),
rp.requests(),
"build info must not cost an extra request",
);
let none = build_image();
let rn = CountingReader::new(SliceReader::new(&none));
let (m, b) = read_card_and_build_info_ranged(&rn).unwrap();
assert!(m.is_none() && b.is_none());
assert_eq!(rn.requests(), 1);
}
#[test]
fn schema_summary_groups_by_type() {
let rt = RDF_TYPE;
let bytes = build_from(&[
("Alice", rt, "Person"),
("Bob", rt, "Person"),
("NYC", rt, "City"),
("Alice", "knows", "Bob"),
("Alice", "livesIn", "NYC"),
("Alice", "name", "\"Alice\""),
]);
let rete = Rete::open(&bytes).unwrap();
let summary = schema_summary(&rete);
assert!(summary.contains(&("Person".into(), "knows".into(), "Person".into(), 1)));
assert!(summary.contains(&("Person".into(), "livesIn".into(), "City".into(), 1)));
assert!(summary.contains(&("Person".into(), "name".into(), "(literal)".into(), 1)));
assert!(!summary.iter().any(|(_, p, _, _)| p == RDF_TYPE));
let classes = schema_classes(&rete);
assert_eq!(
classes,
vec![("Person".into(), 2u32), ("City".into(), 1u32)]
);
}
fn build_from(triples: &[(&str, &str, &str)]) -> Vec<u8> {
let mut db = DictionaryBuilder::new();
for (s, p, o) in triples {
db.observe(s, p, o);
}
let dict = db.build();
let mut ib = GraphIndexBuilder::new();
for (s, p, o) in triples {
ib.push(dict.encode(s, p, o).unwrap());
}
write_file(&dict, &ib.build(), false, &[], 0)
}
fn build_with_pyramid(triples: &[(&str, &str, &str)]) -> Vec<u8> {
let mut db = DictionaryBuilder::new();
for (s, p, o) in triples {
db.observe(s, p, o);
}
let dict = db.build();
let encoded: Vec<_> = triples
.iter()
.map(|(s, p, o)| dict.encode(s, p, o).unwrap())
.collect();
let mut ib = GraphIndexBuilder::new();
for t in &encoded {
ib.push(*t);
}
let (meta, levels) = build_pyramid_meta(&dict, &encoded, DEFAULT_TILE_BUDGET);
write_dataset(&dict, &ib.build(), &[], false, &meta, levels)
}
#[test]
fn tbox_coherence_flags_unsatisfiable_class_index_free() {
use crate::reader::{CountingReader, SliceReader};
let rt = RDF_TYPE;
let sub = "<http://www.w3.org/2000/01/rdf-schema#subClassOf>";
let disj = "<http://www.w3.org/2002/07/owl#disjointWith>";
let bytes = build_with_pyramid(&[
("<http://ex/C>", sub, "<http://ex/D>"),
("<http://ex/C>", sub, "<http://ex/E>"),
("<http://ex/D>", disj, "<http://ex/E>"),
("<http://ex/x>", rt, "<http://ex/C>"),
]);
let r = CountingReader::new(SliceReader::new(&bytes));
let view = SummaryView::open_ranged(&r).unwrap().unwrap();
let points = view.tbox_coherence();
assert!(
points
.iter()
.any(|i| i.kind == "unsatisfiable-class" && i.detail.contains("http://ex/C>")),
"expected C unsatisfiable from the schema alone, got {points:?}"
);
let header = Header::from_bytes(&bytes[..HEADER_LEN]).unwrap();
assert!(
r.bytes_read() <= bytes.len() as u64 - header.root_dir_len,
"tbox_coherence must not read the triple index"
);
}
#[test]
fn schema_coherence_reads_only_the_schema_block() {
use crate::reader::{CountingReader, SliceReader};
let rt = RDF_TYPE;
let sub = "<http://www.w3.org/2000/01/rdf-schema#subClassOf>";
let disj = "<http://www.w3.org/2002/07/owl#disjointWith>";
let mut triples: Vec<(String, String, String)> = vec![
("<http://ex/C>".into(), sub.into(), "<http://ex/D>".into()),
("<http://ex/C>".into(), sub.into(), "<http://ex/E>".into()),
("<http://ex/D>".into(), disj.into(), "<http://ex/E>".into()),
];
for i in 0..500 {
let s = format!("<http://ex/x{i}>");
triples.push((s.clone(), rt.into(), "<http://ex/C>".into()));
triples.push((
s,
"<http://ex/label>".into(),
format!("\"unique label {i}\""),
));
}
let trefs: Vec<(&str, &str, &str)> = triples
.iter()
.map(|(s, p, o)| (s.as_str(), p.as_str(), o.as_str()))
.collect();
let bytes = build_with_pyramid(&trefs);
let header = Header::from_bytes(&bytes[..HEADER_LEN]).unwrap();
assert!(
header.schema_meta_len > 0,
"the writer recorded a schema-block length"
);
assert!(
(header.schema_meta_len as u64) < header.pyramid_meta_len,
"schema block ({}) should be far smaller than the whole pyramid-meta ({})",
header.schema_meta_len,
header.pyramid_meta_len
);
let r = CountingReader::new(SliceReader::new(&bytes));
let points = read_schema_coherence_ranged(&r).unwrap().unwrap();
assert!(points.iter().any(|i| i.kind == "unsatisfiable-class"));
assert!(
r.bytes_read() <= HEADER_LEN as u64 + header.schema_meta_len as u64,
"read {} bytes; expected <= header + schema block ({})",
r.bytes_read(),
HEADER_LEN as u64 + header.schema_meta_len as u64
);
}
#[test]
fn tbox_coherence_clean_schema_is_coherent() {
use crate::reader::SliceReader;
let rt = RDF_TYPE;
let sub = "<http://www.w3.org/2000/01/rdf-schema#subClassOf>";
let bytes = build_with_pyramid(&[
("<http://ex/Dog>", sub, "<http://ex/Animal>"),
("<http://ex/x>", rt, "<http://ex/Dog>"),
]);
let view = SummaryView::open_ranged(&SliceReader::new(&bytes))
.unwrap()
.unwrap();
assert!(view.tbox_is_coherent(), "a plain hierarchy is coherent");
}
#[test]
fn named_graphs_round_trip() {
let all = [
("Alice", "knows", "Bob"), ("Bob", "age", "30"), ];
let mut db = DictionaryBuilder::new();
for (s, p, o) in all {
db.observe(s, p, o);
}
let dict = db.build();
let mut def = GraphIndexBuilder::new();
def.push(dict.encode("Alice", "knows", "Bob").unwrap());
let mut g1 = GraphIndexBuilder::new();
g1.push(dict.encode("Bob", "age", "30").unwrap());
let named = vec![("http://ex/g1".to_string(), g1.build())];
let bytes = write_dataset(&dict, &def.build(), &named, true, &[], 0);
assert!(verify(&bytes).unwrap());
let rete = Rete::open(&bytes).unwrap();
assert_eq!(rete.graph_names(), vec!["http://ex/g1"]);
let gi = rete.graph_index("http://ex/g1").unwrap();
assert_eq!(gi.triple_count(), 1);
assert!(rete.graph_index("http://ex/missing").is_none());
assert_eq!(rete.header().quad_count, 2);
assert_eq!(rete.query(Some("Alice"), None, None).len(), 1);
assert_eq!(
rete.dump(None),
vec![("Alice".into(), "knows".into(), "Bob".into())]
);
assert_eq!(
rete.dump(Some("http://ex/g1")),
vec![("Bob".into(), "age".into(), "30".into())]
);
}
#[test]
fn named_graph_container_opens_tile_lazily_like_the_root() {
let mut db = DictionaryBuilder::new();
let node = |n: u32| format!("<http://ex/n{n}>");
let p = "<http://ex/p>".to_string();
db.observe(&node(0), &p, &node(1));
for i in 0..500u32 {
db.observe(&node(1000 + i), &p, &node(2000 + i));
}
let dict = db.build();
let mut def = GraphIndexBuilder::new();
def.push(dict.encode(&node(0), &p, &node(1)).unwrap());
let mut g = GraphIndexBuilder::new().with_tile_budget(256); for i in 0..500u32 {
g.push(dict.encode(&node(1000 + i), &p, &node(2000 + i)).unwrap());
}
let named = vec![("<http://ex/g>".to_string(), g.build())];
let bytes = write_dataset(&dict, &def.build(), &named, true, &[], 0);
let rete = Rete::open(&bytes).unwrap();
let header = rete.header();
let soff = header.named_graphs_offset as usize;
let send = soff + header.named_graphs_len as usize;
let sec = &bytes[soff..send];
let (n, mut pos) = read_uvarint(sec).unwrap();
assert_eq!(n, 1);
let (ilen, u1) = read_uvarint(&sec[pos..]).unwrap();
pos += u1 + ilen as usize;
let (clen, u2) = read_uvarint(&sec[pos..]).unwrap();
pos += u2;
let container = ByteRange {
offset: (soff + pos) as u64,
len: clen,
};
use crate::reader::SliceReader;
let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
let reader: std::sync::Arc<dyn RangeReader + Send + Sync> =
std::sync::Arc::new(SliceReader::new(leaked));
let (lazy_idx, _, _) = open_index_container_lazy(
&reader,
container,
header.block_codec,
header.has_tile_synopsis(),
1,
header.perms,
)
.unwrap();
let mut want = rete
.graph_index("<http://ex/g>")
.unwrap()
.match_pattern((None, None, None));
let mut got = lazy_idx.match_pattern((None, None, None));
want.sort_unstable();
got.sort_unstable();
assert_eq!(got.len(), 500);
assert_eq!(got, want);
assert!(!lazy_idx.load_incomplete());
}
#[test]
fn dump_iter_matches_dump_and_stops_early() {
use crate::reader::{CountingReader, SliceReader};
let triples: Vec<(String, String, String)> = (0..2000u32)
.map(|i| {
(
format!("<http://ex/s/{i:04}>"),
"<http://ex/p>".to_string(),
format!("<http://ex/o/{i:04}>"),
)
})
.collect();
let mut db = DictionaryBuilder::new();
for (s, p, o) in &triples {
db.observe(s, p, o);
}
let dict = db.build();
let mut ib = GraphIndexBuilder::new().with_tile_budget(256);
for (s, p, o) in &triples {
ib.push(dict.encode(s, p, o).unwrap());
}
let bytes = write_file(&dict, &ib.build(), false, &[], 0);
let rete = Rete::open(&bytes).unwrap();
assert_eq!(rete.dump_iter(None).collect::<Vec<_>>(), rete.dump(None));
assert_eq!(rete.dump_iter(Some("http://ex/nope")).count(), 0);
let read_bytes = |take: usize| -> u64 {
let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
let rete = Rete::open_ranged_lazy(reader.clone()).unwrap();
let before = reader.bytes_read();
let n = rete.dump_iter(None).take(take).count();
assert_eq!(n, take.min(triples.len()));
assert!(!rete.index_incomplete());
reader.bytes_read() - before
};
let first_only = read_bytes(1);
let everything = read_bytes(triples.len());
assert!(
first_only < everything,
"dump_iter is not lazy: one triple read {first_only} B, a full scan {everything} B"
);
#[cfg(feature = "compression")]
assert!(
first_only * 2 < everything,
"dump_iter is not lazy: one triple read {first_only} B of the {everything} B a full scan reads"
);
let fresh = Rete::open(&bytes).unwrap();
let drain = |max_quads: usize| {
let (mut out, mut cursor, mut calls) = (Vec::new(), 0u32, 0);
loop {
let (batch, next, done) = fresh.dump_batch(None, cursor, max_quads);
out.extend(batch);
cursor = next;
calls += 1;
assert!(calls < 10_000, "cursor failed to advance");
if done {
break out;
}
}
};
let eager = fresh.dump(None);
for max in [1usize, 7, 128, 100_000] {
assert_eq!(drain(max), eager, "batch size {max} changed the dump");
}
let (t, _, done) = fresh.dump_batch(Some("http://ex/nope"), 0, 16);
assert!(t.is_empty() && done);
}
#[test]
fn query_batch_matches_query_in_graph_and_stops_early() {
use crate::reader::{CountingReader, SliceReader};
let triples: Vec<(String, String, String)> = (0..2000u32)
.map(|i| {
(
format!("<http://ex/s/{:04}>", i / 2),
format!("<http://ex/p/{}>", i % 2),
format!("<http://ex/o/{i:04}>"),
)
})
.collect();
let mut db = DictionaryBuilder::new();
for (s, p, o) in &triples {
db.observe(s, p, o);
}
let dict = db.build();
let mut ib = GraphIndexBuilder::new().with_tile_budget(256);
for (s, p, o) in &triples {
ib.push(dict.encode(s, p, o).unwrap());
}
let bytes = write_file(&dict, &ib.build(), false, &[], 0);
let rete = Rete::open(&bytes).unwrap();
let drain = |s: Option<&str>, p: Option<&str>, o: Option<&str>, max: usize| {
let (mut out, mut cursor, mut calls) = (Vec::new(), 0u64, 0);
loop {
let (batch, next, done) = rete.query_batch(None, s, p, o, cursor, max);
assert!(!batch.is_empty() || done, "empty non-final batch");
out.extend(batch);
cursor = next;
calls += 1;
assert!(calls < 20_000, "cursor failed to advance");
if done {
break out;
}
}
};
let shapes: [(Option<&str>, Option<&str>, Option<&str>); 6] = [
(None, None, None),
(Some("<http://ex/s/0007>"), None, None),
(None, Some("<http://ex/p/1>"), None),
(None, None, Some("<http://ex/o/0042>")),
(Some("<http://ex/s/0007>"), Some("<http://ex/p/0>"), None),
(
Some("<http://ex/s/0007>"),
Some("<http://ex/p/0>"),
Some("<http://ex/o/0014>"),
),
];
for (s, p, o) in shapes {
let eager = rete.query_in_graph(None, s, p, o);
assert!(
!eager.is_empty(),
"fixture shape {s:?} {p:?} {o:?} matches nothing"
);
assert_eq!(
rete.query_iter(None, s, p, o).collect::<Vec<_>>(),
eager,
"query_iter disagrees for {s:?} {p:?} {o:?}"
);
for max in [1usize, 3, 64, 100_000] {
let mut got = drain(s, p, o, max);
let mut want = eager.clone();
got.sort();
want.sort();
assert_eq!(got, want, "batch size {max} changed {s:?} {p:?} {o:?}");
}
if s.is_none() && p.is_none() && o.is_none() {
assert_eq!(drain(s, p, o, 7), eager, "unbound batching reordered rows");
}
}
let (t, _, done) = rete.query_batch(None, Some("<http://ex/nope>"), None, None, 0, 16);
assert!(t.is_empty() && done);
let (t, _, done) = rete.query_batch(Some("http://ex/nope"), None, None, None, 0, 16);
assert!(t.is_empty() && done);
assert_eq!(
rete.query_iter(None, Some("<http://ex/nope>"), None, None)
.count(),
0
);
let read_bytes = |take: usize| -> u64 {
let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
let rete = Rete::open_ranged_lazy(reader.clone()).unwrap();
let before = reader.bytes_read();
let (mut n, mut cursor) = (0usize, 0u64);
loop {
let (batch, next, done) = rete.query_batch(None, None, None, None, cursor, 1);
n += batch.len();
cursor = next;
if done || n >= take {
break;
}
}
assert!(n >= take.min(triples.len()));
assert!(!rete.index_incomplete());
reader.bytes_read() - before
};
let first_only = read_bytes(1);
let everything = read_bytes(triples.len());
assert!(
first_only < everything,
"query_batch is not lazy: one row read {first_only} B, a full scan {everything} B"
);
#[cfg(feature = "compression")]
assert!(
first_only * 2 < everything,
"query_batch is not lazy: one row read {first_only} B of the {everything} B a full scan reads"
);
}
#[test]
fn query_batch_is_graph_scoped() {
let mut db = DictionaryBuilder::new();
for (s, p, o) in [
("Alice", "knows", "Bob"),
("Alice", "knows", "Carol"),
("Alice", "knows", "Dave"),
("Erin", "knows", "Frank"),
("Alice", "knows", "Zoe"),
] {
db.observe(s, p, o);
}
let dict = db.build();
let mut def = GraphIndexBuilder::new();
def.push(dict.encode("Alice", "knows", "Bob").unwrap());
def.push(dict.encode("Alice", "knows", "Carol").unwrap());
let mut g1 = GraphIndexBuilder::new();
g1.push(dict.encode("Alice", "knows", "Dave").unwrap());
g1.push(dict.encode("Erin", "knows", "Frank").unwrap());
let mut g2 = GraphIndexBuilder::new();
g2.push(dict.encode("Alice", "knows", "Zoe").unwrap());
let named = vec![
("http://ex/g1".to_string(), g1.build()),
("http://ex/g2".to_string(), g2.build()),
];
let bytes = write_dataset(&dict, &def.build(), &named, true, &[], 0);
let rete = Rete::open(&bytes).unwrap();
for graph in [None, Some("http://ex/g1"), Some("http://ex/g2")] {
let eager = rete.query_in_graph(graph, None, None, None);
let (mut out, mut cursor) = (Vec::new(), 0u64);
loop {
let (batch, next, done) = rete.query_batch(graph, None, None, None, cursor, 1);
out.extend(batch);
cursor = next;
if done {
break;
}
}
assert_eq!(out, eager, "graph {graph:?} batched differently");
}
}
#[test]
fn query_in_graph_is_graph_scoped() {
let mut db = DictionaryBuilder::new();
for (s, p, o) in [
("Alice", "knows", "Bob"),
("Alice", "knows", "Carol"),
("Alice", "knows", "Dave"),
] {
db.observe(s, p, o);
}
let dict = db.build();
let mut def = GraphIndexBuilder::new();
def.push(dict.encode("Alice", "knows", "Bob").unwrap());
def.push(dict.encode("Alice", "knows", "Carol").unwrap());
let mut g1 = GraphIndexBuilder::new();
g1.push(dict.encode("Alice", "knows", "Dave").unwrap());
let named = vec![("http://ex/g1".to_string(), g1.build())];
let bytes = write_dataset(&dict, &def.build(), &named, true, &[], 0);
let rete = Rete::open(&bytes).unwrap();
let mut def_objs: Vec<String> = rete
.query_in_graph(None, Some("Alice"), Some("knows"), None)
.into_iter()
.map(|(_, _, o)| o)
.collect();
def_objs.sort();
assert_eq!(def_objs, vec!["Bob".to_string(), "Carol".to_string()]);
assert_eq!(
rete.query_in_graph(Some("http://ex/g1"), Some("Alice"), None, None),
vec![("Alice".into(), "knows".into(), "Dave".into())]
);
assert_eq!(rete.query_in_graph(None, None, None, None).len(), 2);
assert_eq!(
rete.query_in_graph(Some("http://ex/g1"), None, None, None)
.len(),
1
);
assert!(rete
.query_in_graph(Some("http://ex/missing"), None, None, None)
.is_empty());
}
#[test]
fn bounded_query_in_graph_does_not_fault_the_whole_dictionary() {
use crate::reader::{CountingReader, SliceReader};
let payload = |i: u32| -> String {
let mut s = String::with_capacity(800);
let mut x = i as u64 ^ 0x9e37_79b9_7f4a_7c15;
while s.len() < 800 {
x = x
.wrapping_mul(6364136223846793005)
.wrapping_add(1442695040888963407);
s.push_str(&format!("{:016x}", x ^ (x >> 29)));
}
s
};
let quads: Vec<(String, String, String, bool)> = (0..2000u32)
.map(|i| {
let named = i % 2 == 0;
let ns = if named { 'a' } else { 'b' };
(
format!("<http://ex/{ns}/s/{i:04}>"),
"<http://ex/abstract>".to_string(),
format!("\"{ns}{} number {i}\"", payload(i)),
named,
)
})
.collect();
let mut db = DictionaryBuilder::new();
for (s, p, o, _) in &quads {
db.observe(s, p, o);
}
let dict = db.build();
let mut def = GraphIndexBuilder::new().with_tile_budget(256);
let mut g1 = GraphIndexBuilder::new().with_tile_budget(256);
for (s, p, o, named) in &quads {
let t = dict.encode(s, p, o).unwrap();
if *named {
g1.push(t);
} else {
def.push(t);
}
}
let named = vec![("http://ex/g1".to_string(), g1.build())];
let bytes = write_dataset(&dict, &def.build(), &named, true, &[], 0);
let eager = Rete::open(&bytes).unwrap();
let dict_len = eager.header().dictionary_len;
let leaked: &'static [u8] = Box::leak(bytes.clone().into_boxed_slice());
let probe = |f: &dyn Fn(&Rete) -> Vec<TermTriple>| -> (Vec<TermTriple>, u64) {
let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
let lazy = Rete::open_ranged_lazy(reader.clone()).unwrap();
let before = reader.bytes_read();
let out = f(&lazy);
assert!(!lazy.index_incomplete(), "a chunk fetch failed");
(out, reader.bytes_read() - before)
};
let want = eager.query_in_graph(
Some("http://ex/g1"),
Some("<http://ex/a/s/0100>"),
None,
None,
);
assert_eq!(want.len(), 1, "fixture: expected exactly one match");
let (got, pulled) = probe(&|r| {
r.query_in_graph(
Some("http://ex/g1"),
Some("<http://ex/a/s/0100>"),
None,
None,
)
});
assert_eq!(got, want, "lazy query_in_graph disagrees with eager");
assert!(
pulled * 4 < dict_len,
"a 1-row query_in_graph faulted {pulled} B of a {dict_len} B dictionary — \
it is prefetching the whole thing again"
);
let (empty, q_pulled) =
probe(&|r| r.query_in_graph(Some("http://ex/nope"), None, None, None));
assert!(empty.is_empty());
let (_, d_pulled) = probe(&|r| {
let mut n = 0usize;
r.dump_each(Some("http://ex/nope"), |_, _, _| n += 1);
assert_eq!(n, 0, "an absent graph yielded triples");
Vec::new()
});
assert_eq!(
q_pulled, d_pulled,
"query_in_graph and dump_each disagree on what an absent graph costs"
);
assert!(
q_pulled * 4 < dict_len,
"an absent graph faulted {q_pulled} B of a {dict_len} B dictionary"
);
let want: Vec<TermTriple> = eager.dump(Some("http://ex/g1"));
assert_eq!(want.len(), 1000);
let (got, pulled) = probe(&|r| {
let mut out = Vec::new();
r.dump_each(Some("http://ex/g1"), |s, p, o| {
out.push((s.to_string(), p.to_string(), o.to_string()))
});
out
});
assert_eq!(got, want, "lazy dump_each disagrees with eager dump");
assert!(
pulled * 3 < dict_len * 2,
"dumping half the graph faulted {pulled} B of a {dict_len} B dictionary — \
the other graph's chunks are being pulled in too"
);
}
#[cfg(test)]
fn build_multichunk_image() -> Vec<u8> {
use crate::index::PermSet;
let mut db = DictionaryBuilder::new();
let mut triples: Vec<(String, String, String)> = (0..4000u32)
.map(|i| {
(
format!("<http://ex/s/{i:07}>"),
format!("<http://ex/p/{}>", i % 8),
format!("\"object literal number {i:07} with padding text to widen the term\""),
)
})
.collect();
triples.push((
"<http://ex/s/big>".to_string(),
"<http://ex/p/big>".to_string(),
format!("\"{}\"", "Z".repeat(8000)),
));
for (s, p, o) in &triples {
db.observe(s, p, o);
}
let dict = db.build();
let mut idx = GraphIndexBuilder::new()
.with_tile_budget(64)
.with_perms(PermSet::ALL);
for (s, p, o) in &triples {
idx.push(dict.encode(s, p, o).unwrap());
}
write_dataset(&dict, &idx.build(), &[], true, &[], 0)
}
#[test]
fn tiny_dict_cache_evicts_yet_dump_is_byte_identical() {
use crate::reader::SliceReader;
let leaked: &'static [u8] = Box::leak(build_multichunk_image().into_boxed_slice());
let dump = |cap: u64| -> (
Vec<(String, String, String)>,
crate::chunk_cache::CacheStats,
) {
let rete = Rete::open_ranged_lazy(SliceReader::new(leaked)).unwrap();
rete.dict.set_cache_cap(cap);
let mut out = Vec::new();
rete.dump_filtered_each(None, None, None, None, |s, p, o| {
out.push((s.to_string(), p.to_string(), o.to_string()))
});
assert!(
!rete.index_incomplete(),
"no lazy fault should fail at cap {cap}"
);
let stats = rete.dict_cache_stats();
(out, stats)
};
let (unlimited, us) = dump(u64::MAX);
assert_eq!(unlimited.len(), 4001, "every triple dumped");
assert_eq!(us.evictions, 0, "unlimited cap evicts nothing");
let (tiny, ts) = dump(128 * 1024);
assert_eq!(tiny, unlimited, "128 KiB-cap dump differs from unlimited");
assert!(ts.evictions > 0, "a tiny cap must evict (stats {ts:?})");
let (starved, ss) = dump(1);
assert_eq!(starved, unlimited, "1-byte-cap dump differs from unlimited");
assert!(
ss.evictions > 0,
"1-byte cap evicts every insert (stats {ss:?})"
);
}
#[test]
fn releasing_a_named_graph_drops_and_reopens_its_index() {
use crate::index::PermSet;
use crate::reader::SliceReader;
let mut db = DictionaryBuilder::new();
let all: Vec<(String, String, String)> = (0..40u32)
.map(|i| {
(
format!("<http://ex/s{}>", i % 7),
format!("<http://ex/p{}>", i % 3),
format!("<http://ex/o{}>", i % 5),
)
})
.collect();
for (s, p, o) in &all {
db.observe(s, p, o);
}
let dict = db.build();
let mk = |take: fn(u32) -> bool| {
let mut g = GraphIndexBuilder::new()
.with_tile_budget(16)
.with_perms(PermSet::ALL);
for (i, (s, p, o)) in all.iter().enumerate() {
if take(i as u32) {
g.push(dict.encode(s, p, o).unwrap());
}
}
g.build()
};
let def = mk(|_| true);
let g1 = mk(|i| i % 2 == 0);
let g2 = mk(|i| i % 3 == 0);
let image = write_dataset(
&dict,
&def,
&[
("<http://ex/g1>".to_string(), g1),
("<http://ex/g2>".to_string(), g2),
],
true,
&[],
0,
);
let leaked: &'static [u8] = Box::leak(image.into_boxed_slice());
let mut rete = Rete::open_ranged_lazy(SliceReader::new(leaked)).unwrap();
let opened = |r: &Rete| -> usize {
match &r.named_graphs {
NamedGraphsSlot::Lazy(l) => {
let mut n = 0;
l.for_each_opened(|_| n += 1);
n
}
_ => 0,
}
};
assert_eq!(opened(&rete), 0, "nothing opened before first touch");
assert!(rete.graph_index("<http://ex/g1>").is_some());
assert_eq!(opened(&rete), 1, "g1 opened");
rete.release_named_graph("<http://ex/g1>");
assert_eq!(opened(&rete), 0, "g1 released");
rete.release_named_graph("<http://ex/absent>");
rete.release_named_graph("<http://ex/g1>");
assert_eq!(opened(&rete), 0);
assert!(rete.graph_index("<http://ex/g1>").is_some());
assert_eq!(opened(&rete), 1, "g1 re-opened after release");
}
#[test]
fn filtered_dump_is_exactly_the_unfiltered_dump_filtered() {
use crate::index::PermSet;
use crate::reader::SliceReader;
let terms = |i: u32| {
(
format!("<http://ex/s{}>", i % 11),
format!("<http://ex/p{}>", i % 3),
if i.is_multiple_of(4) {
format!("\"lit {}\"@en", i % 5)
} else {
format!("<http://ex/o{}>", i % 7)
},
)
};
let build = |perms: PermSet| {
let mut db = DictionaryBuilder::new();
let all: Vec<_> = (0..90u32).map(terms).collect();
for (s, p, o) in &all {
db.observe(s, p, o);
}
let dict = db.build();
let mut def = GraphIndexBuilder::new()
.with_tile_budget(16)
.with_perms(perms);
let mut g1 = GraphIndexBuilder::new()
.with_tile_budget(16)
.with_perms(perms);
for (i, (s, p, o)) in all.iter().enumerate() {
let t = dict.encode(s, p, o).unwrap();
if i % 2 == 0 {
g1.push(t);
}
def.push(t);
}
write_dataset(
&dict,
&def.build(),
&[("<http://ex/g1>".to_string(), g1.build())],
true,
&[],
0,
)
};
let shapes: Vec<(Option<&str>, Option<&str>, Option<&str>)> = vec![
(None, None, None),
(Some("<http://ex/s3>"), None, None),
(None, Some("<http://ex/p1>"), None),
(None, None, Some("<http://ex/o5>")),
(None, None, Some("\"lit 0\"@en")),
(Some("<http://ex/s3>"), Some("<http://ex/p0>"), None),
(Some("<http://ex/s4>"), None, Some("<http://ex/o4>")),
(None, Some("<http://ex/p1>"), Some("<http://ex/o5>")),
(
Some("<http://ex/s3>"),
Some("<http://ex/p0>"),
Some("<http://ex/o3>"),
),
(Some("<http://ex/nope>"), None, None),
(None, Some("<http://ex/nope>"), None),
(None, None, Some("<http://ex/nope>")),
];
let mut any_pruned = false;
for perms in [PermSet::ALL, PermSet::CORE] {
let image = build(perms);
let leaked: &'static [u8] = Box::leak(image.clone().into_boxed_slice());
let opens: Vec<(&str, Rete)> = vec![
("resident", Rete::open(&image).unwrap()),
(
"ranged",
Rete::open_ranged(&SliceReader::new(leaked)).unwrap(),
),
(
"lazy",
Rete::open_ranged_lazy(SliceReader::new(leaked)).unwrap(),
),
];
for graph in [None, Some("<http://ex/g1>"), Some("<http://ex/absent>")] {
let full: Vec<TermTriple> = {
let mut v = Vec::new();
Rete::open(&image).unwrap().dump_each(graph, |s, p, o| {
v.push((s.to_string(), p.to_string(), o.to_string()))
});
v.sort();
v
};
for &(s, p, o) in &shapes {
let mut want = full.clone();
want.retain(|(ts, tp, to)| {
s.is_none_or(|x| x == ts)
&& p.is_none_or(|x| x == tp)
&& o.is_none_or(|x| x == to)
});
for (path, rete) in &opens {
let mut got = Vec::new();
rete.dump_filtered_each(graph, s, p, o, |s, p, o| {
got.push((s.to_string(), p.to_string(), o.to_string()))
});
got.sort();
assert_eq!(
got,
want,
"{path}/{perms:?} graph={graph:?} pattern={:?} — filtered dump \
is not the unfiltered dump filtered",
(s, p, o)
);
assert!(!rete.index_incomplete(), "{path}: a fetch failed");
}
let plan = opens[2].1.dump_plan(graph, s, p, o);
if graph == Some("<http://ex/absent>")
|| [s, p, o].contains(&Some("<http://ex/nope>"))
{
assert!(
plan.scan.is_none(),
"an unmatchable dump still planned a scan: {:?}",
(graph, s, p, o)
);
} else {
let scan = plan.scan.expect("a matchable dump plans a scan");
assert!(
PermSet::CORE.contains(scan.permutation),
"a dump routed to {:?}, outside PermSet::CORE — a \
--permutations 3 file has no such section",
scan.permutation
);
assert!(scan.tiles_admitted <= scan.tiles_routed);
assert!(scan.tiles_routed <= scan.tiles_total);
assert!(scan.tile_bytes <= scan.section_bytes);
if scan.tiles_admitted < scan.tiles_total {
any_pruned = true;
}
}
}
}
}
assert!(
any_pruned,
"no shape pruned a single tile — the fixture is too small to be \
measuring anything"
);
}
#[test]
fn a_predicate_scoped_dump_fetches_less_than_the_graph() {
use crate::reader::{CountingReader, SliceReader};
let mut db = DictionaryBuilder::new();
let mut all = Vec::new();
for s in 0..200u32 {
for p in 0..20u32 {
let t = (
format!("<http://ex/s/{s:04}>"),
format!("<http://ex/p/{p:02}>"),
format!("\"object {s:04}/{p:02}\""),
);
db.observe(&t.0, &t.1, &t.2);
all.push(t);
}
}
let dict = db.build();
let mut ib = GraphIndexBuilder::new().with_tile_budget(64);
for (s, p, o) in &all {
ib.push(dict.encode(s, p, o).unwrap());
}
let image = write_dataset(&dict, &ib.build(), &[], false, &[], 0);
let leaked: &'static [u8] = Box::leak(image.clone().into_boxed_slice());
let probe = |s: Option<&str>, p: Option<&str>, o: Option<&str>| -> (usize, u64) {
let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
let lazy = Rete::open_ranged_lazy(reader.clone()).unwrap();
let before = reader.bytes_read();
let mut n = 0usize;
lazy.dump_filtered_each(None, s, p, o, |_, _, _| n += 1);
assert!(!lazy.index_incomplete());
(n, reader.bytes_read() - before)
};
let (full_rows, full_bytes) = probe(None, None, None);
let (pred_rows, pred_bytes) = probe(None, Some("<http://ex/p/07>"), None);
let (subj_rows, subj_bytes) = probe(Some("<http://ex/s/0021>"), None, None);
assert_eq!(full_rows, 4000);
assert_eq!(pred_rows, 200, "one predicate covers every subject once");
assert_eq!(subj_rows, 20);
assert!(
pred_bytes < full_bytes && subj_bytes < full_bytes,
"a slice fetched {pred_bytes} / {subj_bytes} B where the whole graph fetched \
{full_bytes} B — the scan is not being pruned at all"
);
#[cfg(feature = "compression")]
assert!(
pred_bytes * 4 < full_bytes && subj_bytes * 4 < full_bytes,
"a 1-in-20 predicate slice fetched {pred_bytes} B and one subject \
{subj_bytes} B where the whole graph fetched {full_bytes} B"
);
let reader = std::sync::Arc::new(CountingReader::new(SliceReader::new(leaked)));
let lazy = Rete::open_ranged_lazy(reader.clone()).unwrap();
let before = reader.bytes_read();
let plan = lazy
.dump_plan(None, None, Some("<http://ex/p/07>"), None)
.scan
.expect("a known predicate plans a scan");
assert!(
reader.bytes_read() - before < 64 * 1024,
"planning fetched {} B — it is supposed to read directories, not tiles",
reader.bytes_read() - before
);
assert!(plan.tiles_admitted < plan.tiles_total);
assert!(
plan.tile_bytes * 5 < plan.section_bytes,
"one predicate of twenty planned {} B of a {} B section",
plan.tile_bytes,
plan.section_bytes
);
}
#[test]
fn query_quads_tags_every_graph() {
let mut db = DictionaryBuilder::new();
for (s, p, o) in [("Alice", "knows", "Bob"), ("Alice", "knows", "Dave")] {
db.observe(s, p, o);
}
let dict = db.build();
let mut def = GraphIndexBuilder::new();
def.push(dict.encode("Alice", "knows", "Bob").unwrap());
let mut g1 = GraphIndexBuilder::new();
g1.push(dict.encode("Alice", "knows", "Dave").unwrap());
let named = vec![("http://ex/g1".to_string(), g1.build())];
let bytes = write_dataset(&dict, &def.build(), &named, true, &[], 0);
let rete = Rete::open(&bytes).unwrap();
let quads = rete.query_quads(Some("Alice"), Some("knows"), None);
assert_eq!(quads.len(), 2);
assert_eq!(
quads[0],
(("Alice".into(), "knows".into(), "Bob".into()), None)
);
assert_eq!(
quads[1],
(
("Alice".into(), "knows".into(), "Dave".into()),
Some("http://ex/g1".to_string())
)
);
assert!(rete.query_quads(Some("Nobody"), None, None).is_empty());
}
#[test]
fn pyramid_meta_round_trips_in_file() {
let rete = Rete::open(&build_image()).unwrap();
let pyr = rete.pyramid().expect("file has a pyramid");
let total: u32 = pyr.summary.iter().map(|e| e.count).sum();
assert_eq!(total, 3);
assert!(!pyr.summary.is_empty());
assert!(pyr.tiles.is_empty());
}
#[test]
fn schema_pyramid_round_trips_through_file_index_free() {
use crate::reader::{CountingReader, SliceReader};
let sub = "<http://www.w3.org/2000/01/rdf-schema#subClassOf>";
let q = |s: &str, p: &str, o: &str| {
(s.to_string(), p.to_string(), o.to_string(), None::<String>)
};
let quads = vec![
q("<a>", RDF_TYPE, "<Astronomer>"),
q("<b>", RDF_TYPE, "<Astronomer>"),
q("<c>", RDF_TYPE, "<Person>"),
q("<Astronomer>", sub, "<Scientist>"),
q("<Scientist>", sub, "<Person>"),
q("<Person>", sub, "<Agent>"),
q("<a>", "<knows>", "<b>"),
q("<b>", "<knows>", "<c>"),
];
let (bytes, _) =
crate::ingest::assemble_dataset_with_opts(quads, true, false, None, |_, _| Vec::new());
let rete = Rete::open(&bytes).unwrap();
let pyr = rete.pyramid().expect("pyramid present");
assert!(!pyr.level_rollups.is_empty(), "schema pyramid shipped");
assert!(pyr
.class_hierarchy
.iter()
.any(|n| n.class == "<Agent>" && n.depth == 0));
let r = CountingReader::new(SliceReader::new(&bytes));
let view = SummaryView::open_ranged(&r).unwrap().unwrap();
assert!(view.level_count() >= 2, "multi-level pyramid");
let coarse = view.level_rollup(0).unwrap();
assert!(
coarse.classes.iter().any(|(c, _)| c == "<Agent>"),
"coarsest level rolls up to the root Agent"
);
let h = Header::from_bytes(&bytes).unwrap();
assert!(
r.bytes_read() <= bytes.len() as u64 - h.root_dir_len,
"summary read {} bytes; the {}-byte index section must be skipped",
r.bytes_read(),
h.root_dir_len
);
}
#[test]
fn predicate_totals_from_summary_only() {
use crate::reader::SliceReader;
let bytes = build_image();
let reader = SliceReader::new(&bytes);
let view = SummaryView::open_ranged(&reader).unwrap().unwrap();
assert_eq!(view.predicate_total("knows"), 2);
assert_eq!(view.predicate_total("age"), 1);
assert_eq!(view.predicate_total("missing"), 0);
let totals = view.predicate_totals();
assert_eq!(totals[0], ("knows".to_string(), 2)); }
#[test]
fn query_patterns_resolve_to_terms() {
let rete = Rete::open(&build_image()).unwrap();
assert_eq!(rete.query(None, None, None).len(), 3);
let mut alice = rete.query(Some("Alice"), None, None);
alice.sort();
assert_eq!(
alice,
vec![
("Alice".into(), "age".into(), "30".into()),
("Alice".into(), "knows".into(), "Bob".into()),
]
);
assert_eq!(rete.query(None, Some("knows"), None).len(), 2);
assert_eq!(
rete.query(Some("Bob"), Some("knows"), Some("Carol")),
vec![("Bob".into(), "knows".into(), "Carol".into())]
);
assert!(rete.query(Some("Nobody"), None, None).is_empty());
assert!(rete.query(None, Some("likes"), None).is_empty());
}
#[test]
fn query_provenance_reports_terms_ids_sections_and_index_choice() {
let bytes = build_image();
let rete = Rete::open(&bytes).unwrap();
let mut matches = rete.query_with_provenance(None, Some("knows"), None);
matches.sort_by(|a, b| a.terms.cmp(&b.terms));
assert_eq!(matches.len(), 2);
assert_eq!(
matches[0].terms,
("Alice".into(), "knows".into(), "Bob".into())
);
assert_eq!(
matches[0].ids,
rete.dictionary().encode("Alice", "knows", "Bob").unwrap()
);
assert_eq!(matches[0].graph.as_deref(), None);
assert_eq!(
matches[0].matched_pattern,
(None, Some(matches[0].ids.1), None)
);
assert_eq!(
matches[0].index_permutation,
crate::index::IndexPermutation::Pos
);
let h = rete.header();
assert_eq!(matches[0].dictionary_range.offset, h.dictionary_offset);
assert_eq!(matches[0].dictionary_range.len, h.dictionary_len);
assert_eq!(matches[0].index_range.offset, h.root_dir_offset);
assert_eq!(matches[0].index_range.len, h.root_dir_len);
assert!(
matches[0].index_section_range.offset > h.root_dir_offset,
"POS is section 1, so its payload starts after the container header and SPO payload"
);
assert!(matches[0].index_section_range.len > 0);
assert!(matches[0].index_section_range.end() <= matches[0].index_range.end());
assert!(matches[0].index_section_range.len < matches[0].index_range.len);
assert_eq!(
matches[0].pyramid_range.as_ref().map(|r| (r.offset, r.len)),
Some((h.pyramid_meta_offset, h.pyramid_meta_len))
);
let tile_range = matches[0].tile_range.expect("tiled file reports a tile");
assert!(matches[0]
.tile
.as_deref()
.unwrap()
.starts_with(matches[0].index_permutation.name()));
assert!(matches[0].index_section_range.offset <= tile_range.offset);
assert!(tile_range.end() <= matches[0].index_section_range.end());
}
fn build_labeled(n: usize) -> Vec<u8> {
const LABEL: &str = "<http://www.w3.org/2000/01/rdf-schema#label>";
const WORDS: &[&str] = &[
"alanine",
"benzene",
"glucose",
"dextrose",
"ethanol",
"formate",
"heptane",
"isoleucine",
];
let triples: Vec<(String, String, String)> = (0..n)
.flat_map(|i| {
let s = format!("<http://ex/e{i}>");
let w = WORDS[i % WORDS.len()];
[
(s.clone(), LABEL.to_string(), format!("\"{w}-{i:06}\"")),
(
s,
"<http://ex/p>".to_string(),
format!("<http://ex/c{}>", i % 64),
),
]
})
.collect();
let mut db = DictionaryBuilder::new();
for (s, p, o) in &triples {
db.observe(s, p, o);
}
let dict = db.build();
let ids: Vec<(u32, u32, u32)> = triples
.iter()
.map(|(s, p, o)| dict.encode(s, p, o).unwrap())
.collect();
let mut ib = GraphIndexBuilder::new();
for &t in &ids {
ib.push(t);
}
let (meta, levels) = build_pyramid_meta(&dict, &ids, DEFAULT_TILE_BUDGET);
write_file(&dict, &ib.build(), false, &meta, levels)
}
#[test]
fn prefix_search_matches_a_filter_scan() {
let bytes = build_labeled(800);
let rete = Rete::open(&bytes).unwrap();
let idx_subjects: std::collections::BTreeSet<String> = rete
.prefix_search("glucose", 10_000)
.into_iter()
.map(|(_label, subject)| subject)
.collect();
let q = "SELECT ?s WHERE { ?s <http://www.w3.org/2000/01/rdf-schema#label> ?l \
FILTER(STRSTARTS(LCASE(?l), \"glucose\")) }";
let crate::QueryOutput::Select(_, rows) = crate::eval_query(&rete, q).unwrap() else {
panic!("expected SELECT");
};
let scan_subjects: std::collections::BTreeSet<String> =
rows.iter().map(|r| r.get("s").cloned().unwrap()).collect();
assert_eq!(idx_subjects, scan_subjects, "index agrees with the scan");
assert_eq!(
idx_subjects.len(),
100,
"800/8 words = 100 glucose-* labels"
);
}
#[test]
#[ignore]
fn bench_prefix_search_vs_filter_scan() {
use std::time::Instant;
let n = 6000; let bytes = build_labeled(n);
let rete = Rete::open(&bytes).unwrap();
let q = "SELECT ?s WHERE { ?s <http://www.w3.org/2000/01/rdf-schema#label> ?l \
FILTER(STRSTARTS(LCASE(?l), \"glucose\")) }";
let reps = 200;
let idx_n = rete.prefix_search("glucose", 100_000).len();
let t = Instant::now();
for _ in 0..reps {
std::hint::black_box(rete.prefix_search("glucose", 100_000));
}
let idx_ms = t.elapsed().as_secs_f64() * 1000.0 / reps as f64;
let t = Instant::now();
for _ in 0..reps {
let _ = std::hint::black_box(crate::eval_query(&rete, q).unwrap());
}
let scan_ms = t.elapsed().as_secs_f64() * 1000.0 / reps as f64;
println!(
"label prefix search over {n} labeled subjects ({idx_n} matches): \
index {idx_ms:.4} ms vs FILTER scan {scan_ms:.3} ms ({:.0}× faster)",
scan_ms / idx_ms
);
}
#[test]
#[ignore]
fn bench_text_search_vs_contains_scan() {
use std::time::Instant;
const LABEL: &str = "<http://www.w3.org/2000/01/rdf-schema#label>";
const WORDS: &[&str] = &[
"alanine",
"benzene",
"glucose",
"dextrose",
"ethanol",
"formate",
"heptane",
"isoleucine",
];
let n = 6000;
let triples: Vec<(String, String, String)> = (0..n)
.map(|i| {
(
format!("<http://ex/e{i}>"),
LABEL.to_string(),
format!("\"{} sample number {i:06}\"", WORDS[i % WORDS.len()]),
)
})
.collect();
let mut db = DictionaryBuilder::new();
for (s, p, o) in &triples {
db.observe(s, p, o);
}
let dict = db.build();
let ids: Vec<(u32, u32, u32)> = triples
.iter()
.map(|(s, p, o)| dict.encode(s, p, o).unwrap())
.collect();
let mut ib = GraphIndexBuilder::new();
for &t in &ids {
ib.push(t);
}
let ti = compute_text_index(&dict, &ids);
let bytes = write_dataset_with_metadata(&dict, &ib.build(), &[], false, &[], 0, &[], &ti);
let rete = Rete::open(&bytes).unwrap();
let q = "SELECT ?s WHERE { ?s <http://www.w3.org/2000/01/rdf-schema#label> ?l \
FILTER(CONTAINS(LCASE(?l), \"glucose\")) }";
let reps = 200;
let idx_n = rete.text_search(&["glucose"], None, 100_000).len();
let t = Instant::now();
for _ in 0..reps {
std::hint::black_box(rete.text_search(&["glucose"], None, 100_000));
}
let idx_ms = t.elapsed().as_secs_f64() * 1000.0 / reps as f64;
let t = Instant::now();
for _ in 0..reps {
let _ = std::hint::black_box(crate::eval_query(&rete, q).unwrap());
}
let scan_ms = t.elapsed().as_secs_f64() * 1000.0 / reps as f64;
println!(
"text search over {n} literals ({idx_n} matches): \
index {idx_ms:.4} ms vs FILTER(CONTAINS) scan {scan_ms:.3} ms ({:.0}× faster)",
scan_ms / idx_ms
);
}
#[test]
#[ignore = "operational tool, driven by RETE_DEBUG_* env vars"]
fn debug_bound_po_routing() {
struct FR(std::fs::File);
impl crate::RangeReader for FR {
fn len(&self) -> u64 {
self.0.metadata().map(|m| m.len()).unwrap_or(0)
}
fn read_at(&self, offset: u64, len: u64) -> std::io::Result<Vec<u8>> {
use std::os::unix::fs::FileExt;
let mut buf = vec![0u8; len as usize];
self.0.read_exact_at(&mut buf, offset)?;
Ok(buf)
}
}
let path = std::env::var("RETE_DEBUG_FILE").expect("RETE_DEBUG_FILE");
let p_iri = std::env::var("RETE_DEBUG_P").expect("RETE_DEBUG_P");
let o_iri = std::env::var("RETE_DEBUG_O").expect("RETE_DEBUG_O");
let rete =
Rete::open_ranged_lazy(std::sync::Arc::new(FR(std::fs::File::open(&path).unwrap())))
.unwrap();
let pid = rete.dict.predicate_id(&p_iri).expect("p resolves");
let oid = rete.dict.object_id(&o_iri).expect("o resolves");
eprintln!("pid={pid} oid={oid}");
let pattern = (None, Some(pid), Some(oid));
let perm = GraphIndex::best_permutation(pattern);
eprintln!("best_permutation = {}", perm.name());
let si = perm.section_index();
let tiles = &rete.index.sections[si];
eprintln!("section {} tiles = {}", perm.name(), tiles.len());
let [pa, pb, pc] = perm.order_pattern(pattern);
eprintln!("permuted pattern pa={pa:?} pb={pb:?} pc={pc:?}");
let (start, end) = rete.index.tile_span(si, pa);
eprintln!("tile_span = [{start}, {end}) -> {} tiles", end - start);
let mut admitted = 0usize;
for (ti, t) in tiles.iter().enumerate().take(end).skip(start) {
if t.syn_admits(pb, pc) {
admitted += 1;
if admitted <= 10 {
let (lo, hi) = t.leading_range();
eprintln!(" admit tile {ti}: a=[{lo},{hi}] syn={:?}", t.syn);
}
}
}
eprintln!("admitted {admitted} tile(s) by synopsis");
let n = rete.index.scan_iter(pattern).count();
eprintln!("scan_iter matches = {n}");
let hi_res = rete.query(None, Some(&p_iri), Some(&o_iri));
eprintln!("high-level query matches = {}", hi_res.len());
}
#[test]
fn layout_header_label_matches_its_span() {
let mut db = DictionaryBuilder::new();
let mut triples = Vec::new();
for i in 0..24u32 {
let (s, p, o) = (
format!("<http://ex/s{}>", i % 5),
format!("<http://ex/p{}>", i % 3),
format!("<http://ex/o{i}>"),
);
db.observe(&s, &p, &o);
triples.push((s, p, o));
}
let dict = db.build();
let ids: Vec<(u32, u32, u32)> = triples
.iter()
.map(|(s, p, o)| dict.encode(s, p, o).unwrap())
.collect();
let index = GraphIndexBuilder::from_triples(ids).build();
let image = write_dataset_with_metadata(
&dict,
&index,
&[],
false,
&[],
0,
br#"{"name":"layout-label-fixture"}"#,
&[],
);
let rete = Rete::open(&image).unwrap();
let layout = rete.file_layout();
let header = layout
.iter()
.find(|s| s.kind == "header")
.expect("every layout starts with a header segment");
assert_eq!(
header.offset, 0,
"the header is the first thing in the file"
);
assert_eq!(
header.len, HEADER_LEN as u64,
"the header segment must span exactly HEADER_LEN"
);
let digits: String = header
.label
.chars()
.skip_while(|c| !c.is_ascii_digit())
.take_while(|c| c.is_ascii_digit())
.collect();
assert!(
!digits.is_empty(),
"header label {:?} states no size at all",
header.label
);
assert_eq!(
digits.parse::<u64>().unwrap(),
header.len,
"header label {:?} disagrees with the {}-byte span it describes",
header.label,
header.len
);
assert!(
layout.windows(2).all(|w| w[0].offset <= w[1].offset),
"file_layout must return segments sorted by offset"
);
}
}