#![cfg(not(target_arch = "wasm32"))]
use std::collections::BinaryHeap;
use std::fs::File;
use std::io::{BufReader, BufWriter, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use crate::build_progress::{self as progress, Stat, Ticker, Unit};
use crate::build_trace::{self as trace, note_merge, Phase};
use crate::dict::env_restart_interval;
use crate::extbuild_remap::{
remap_buf_bytes, role_only_id, shared_id, RemapStore, RemapWriter, Seg, SegCounts,
};
use crate::extbuild_sections::{ExtLabels, ExtSchema};
use crate::extbuild_spill::{
self as spill, RunFile, RunReader, RunSpiller, TermReader, TermWriter, ZReader, ZWriter,
};
use crate::index::GroupSizer;
use crate::ingest::{BuildStats, GraphLayout, IngestError, RawQuad};
use crate::triples::TripleBlockBuilder;
use crate::varint::write_uvarint;
use crate::DictionaryBuilder;
type IdTriple = (u32, u32, u32);
type Synopsis = (u32, u32, u32, u32);
pub type MetadataFn = Box<dyn FnOnce(&BuildStats) -> Vec<u8>>;
type TermCarriers = (String, Vec<(usize, u32)>);
type TileDirEntry = (u32, u32, u64, Synopsis);
type EncodedTile = (u32, u32, Vec<u8>, Synopsis);
pub struct ExternalBuildOptions {
pub memory_budget: u64,
pub tmp_dir: Option<PathBuf>,
pub metadata: MetadataFn,
pub build_info: Vec<u8>,
pub perms: crate::index::PermSet,
pub label_index: bool,
pub schema: bool,
pub type_override: Option<String>,
pub text_index: Option<crate::text_index::TextIndexScope>,
pub graphs: GraphLayout,
pub input_memory: u64,
}
impl Default for ExternalBuildOptions {
fn default() -> Self {
Self {
memory_budget: 4 << 30,
tmp_dir: None,
metadata: Box::new(|_| Vec::new()),
build_info: Vec::new(),
perms: crate::index::PermSet::ALL,
label_index: false,
schema: false,
type_override: None,
text_index: None,
graphs: GraphLayout::PerGraph,
input_memory: 0,
}
}
}
#[cfg(test)]
thread_local! {
pub(crate) static TEST_SPLIT: std::cell::Cell<Option<(u64, usize)>> = const { std::cell::Cell::new(None) };
pub(crate) static TEST_RUNS: std::cell::Cell<Option<(usize, usize)>> = const { std::cell::Cell::new(None) };
pub(crate) static TEST_WORKING: std::cell::Cell<Option<u64>> = const { std::cell::Cell::new(None) };
}
#[derive(Debug, thiserror::Error)]
pub enum ExtBuildError {
#[error("ingest: {0}")]
Ingest(#[from] IngestError),
#[error("io: {0}")]
Io(#[from] std::io::Error),
#[error("internal: {0}")]
Internal(&'static str),
}
const NAMED_GRAPH_BYTES_PER_TRIPLE: u64 = 160;
const NAMED_GRAPH_RAM_TRIPLE_CAP: usize = 30_000_000;
const CHUNK_BUDGET_FRACTION: f64 = 0.5;
const CHUNK_WORKING_FRACTION: f64 = 0.85;
const DICT_ENTRY_OVERHEAD: u64 = 96;
const PER_QUAD_OVERHEAD: u64 = 96;
const TILE_COMPRESS_BATCH: usize = 512;
pub fn build_external<S>(
mut stream: S,
output: &Path,
opts: ExternalBuildOptions,
) -> Result<BuildStats, ExtBuildError>
where
S: FnMut(&mut dyn FnMut(RawQuad) -> Result<(), ExtBuildError>) -> Result<(), ExtBuildError>,
{
let tmp_parent = opts
.tmp_dir
.clone()
.or_else(|| output.parent().map(|p| p.to_path_buf()))
.unwrap_or_else(|| PathBuf::from("."));
let tmp = TmpDir::create(&tmp_parent)?;
progress::set_spill_dir(Some(&tmp.dir));
let budget = opts.memory_budget.max(64 << 20);
let chunk_phase = budget.saturating_sub(opts.input_memory).max(32 << 20);
let chunk_budget = (chunk_phase as f64 * CHUNK_BUDGET_FRACTION) as u64;
let text_run_bytes = (budget / 4) as usize;
#[cfg(test)]
let (chunk_budget, text_run_bytes) = TEST_SPLIT
.with(|c| c.get())
.unwrap_or((chunk_budget, text_run_bytes));
let working_cap = (chunk_phase as f64 * CHUNK_WORKING_FRACTION) as u64;
#[cfg(test)]
let working_cap = TEST_WORKING.with(|c| c.get()).unwrap_or(working_cap);
if opts.input_memory > 0 {
eprintln!(
"extbuild: budget {} MiB, {} MiB of it for reading the input -> chunk target {} MiB, \
{} MiB with the chunk's dictionary",
budget >> 20,
opts.input_memory >> 20,
chunk_budget >> 20,
working_cap >> 20
);
} else {
eprintln!(
"extbuild: budget {} MiB -> chunk target {} MiB, {} MiB with the chunk's dictionary",
budget >> 20,
chunk_budget >> 20,
working_cap >> 20
);
}
let quad_layout = opts.graphs == GraphLayout::Quad;
if quad_layout {
crate::quad::check_perms(opts.perms).map_err(|_| {
ExtBuildError::Internal("the quad index needs the 3- or 6-permutation set")
})?;
}
let mut text = opts
.text_index
.clone()
.map(crate::text_index_spill::TextSpill::new);
let promote = crate::write_opts::frequent_objects();
let mut chunker = Chunker::new(&tmp, chunk_budget, opts.graphs)
.with_working_cap(working_cap)
.with_object_counts(promote > 0);
chunker.text = text.as_mut();
let mut ext_labels = opts
.label_index
.then(|| ExtLabels::new(&tmp.dir, (budget / 8) as usize));
let mut ext_schema = opts.schema.then(|| {
ExtSchema::new(
&tmp.dir,
(budget / 8) as usize,
opts.type_override.as_deref(),
)
});
progress::begin(Phase::ParseChunk, None, Unit::None);
stream(&mut |q: RawQuad| chunker.push(q))?;
let chunks = chunker.finish()?;
let statements: u64 = chunks
.iter()
.map(|c| c.triple_count + c.quad_count)
.sum::<u64>();
progress::set_stat(Stat::Statements, statements);
let has_named = chunks.iter().any(|c| c.quad_count > 0);
eprintln!(
"extbuild: {} chunk(s), {} statement(s) spilled",
chunks.len(),
statements
);
trace::add("chunks", chunks.len() as u64);
note_scratch(&tmp);
trace::lap(Phase::ParseChunk);
let promoted = if promote > 0 {
let carriers = chunks.iter().map(|c| {
let [shared, subj, obj, _] = c.section_terms;
shared as u64 + subj as u64 + obj as u64
});
progress::begin(Phase::FrequentObjects, Some(carriers.sum()), Unit::Terms);
let p = select_frequent_objects(&tmp, &chunks, promote)?;
trace::lap(Phase::FrequentObjects);
p
} else {
Vec::new()
};
let carriers = chunks.iter().map(|c| {
let [shared, subj, obj, pred] = c.section_terms;
2 * shared as u64 + subj as u64 + obj as u64 + pred as u64 + c.graph_terms as u64
});
progress::begin(Phase::Dictionary, Some(carriers.sum()), Unit::Terms);
let schema_preds = crate::extbuild_sections::schema_predicates(opts.type_override.as_deref());
let (merged, remaps) =
merge_dictionaries_with(&tmp, &chunks, &schema_preds, budget, &promoted)?;
drop(promoted);
if let Some(x) = ext_schema.as_mut() {
x.set_merge_facts(&merged.schema_facts);
}
if let Some(l) = ext_labels.as_mut() {
l.set_predicates(merged.schema_facts.label_preds.clone());
}
note_scratch(&tmp);
trace::lap(Phase::Dictionary);
progress::set_stat(Stat::Terms, merged.term_count);
eprintln!(
"extbuild: merged dictionary — {} term(s)",
merged.term_count
);
if merged.graph_count > 0 {
eprintln!("extbuild: {} named graph(s)", merged.graph_count);
}
let run_len = ((budget / 2) / 24).clamp(1 << 16, u32::MAX as u64) as usize;
let quad_run_len = ((budget / 2) / 32).clamp(1 << 16, u32::MAX as u64) as usize;
#[cfg(test)]
let (run_len, quad_run_len) = TEST_RUNS
.with(|c| c.get())
.unwrap_or((run_len, quad_run_len));
let first = opts
.perms
.iter()
.next()
.ok_or(ExtBuildError::Internal("no permutation to index"))?;
if let Some(text) = text.as_mut() {
text.begin_remap(tmp.path("global.tok"))?;
}
progress::begin(Phase::Encode, Some(statements), Unit::Statements);
let (remap_run_len, remap_quad_run_len) = if has_named {
(run_len / 2, quad_run_len / 2)
} else {
(run_len, quad_run_len)
};
let (tri_runs, quad_runs) = remap_chunks(
&tmp,
&chunks,
&remaps,
quad_layout,
first,
remap_run_len,
remap_quad_run_len,
text.as_mut(),
ext_schema.as_mut(),
ext_labels.as_mut(),
)?;
note_scratch(&tmp);
trace::lap(Phase::Encode);
let codec = crate::file::writer_codec();
let text_section = match text {
Some(t) => {
progress::begin(Phase::TextIndex, None, Unit::None);
let s = t.finish(&|name| tmp.path(name), text_run_bytes, codec)?;
note_scratch(&tmp);
trace::lap(Phase::TextIndex);
s
}
None => None,
};
if let Some(x) = ext_schema.as_mut() {
x.begin_spo()?;
}
let spilled: u64 = chunks.iter().map(|c| c.triple_count).sum();
let perm_count = opts.perms.len() as u64;
progress::begin(
Phase::IndexPermutations,
Some(perm_count * spilled),
Unit::Records,
);
let mut perms_done = 0u64;
let (perm_sections, default_count) = index_permutations(
&tmp,
tri_runs,
opts.perms,
run_len,
codec,
"",
&mut |perm, t| match (ext_schema.as_mut(), perm) {
(Some(x), crate::index::IndexPermutation::Spo) => x.observe_spo(t),
(Some(x), crate::index::IndexPermutation::Pos) => {
x.observe_pos(t);
Ok(())
}
_ => Ok(()),
},
&mut |perm, n| {
eprintln!(
"extbuild: permutation {} indexed ({} unique triple(s))",
perm.name(),
n
);
trace::lap(Phase::IndexPermutations);
perms_done += 1;
progress::set_done(perms_done * spilled);
},
)?;
if let Some(x) = ext_schema.as_mut() {
x.finish_links()?;
}
if has_named {
let quads: u64 = chunks.iter().map(|c| c.quad_count).sum();
let passes = if quad_layout { perm_count } else { 1 };
progress::begin(Phase::NamedGraphs, Some(passes * quads), Unit::Records);
}
let (quad, named) = if !has_named {
(None, None)
} else if quad_layout {
let q = build_quad_section(&tmp, quad_runs, quad_run_len, opts.perms, codec)?;
eprintln!(
"extbuild: quad index — {} named graph(s), {} unique quad(s), {} byte(s)",
q.graph_count, q.quad_count, q.len
);
(Some(q), None)
} else {
let ram_triples = (((budget / 2) / NAMED_GRAPH_BYTES_PER_TRIPLE).max(1) as usize)
.min(NAMED_GRAPH_RAM_TRIPLE_CAP);
let section = build_named_graphs(
&tmp,
quad_runs,
merged
.graph_names
.as_deref()
.ok_or(ExtBuildError::Internal("named quads without graph names"))?,
merged.graph_count,
run_len,
ram_triples,
opts.perms,
codec,
)?;
eprintln!(
"extbuild: {} named graph(s) indexed ({} unique quad(s))",
section.count, section.triples
);
(None, Some(section))
};
if has_named {
note_scratch(&tmp);
trace::lap(Phase::NamedGraphs);
}
let named_triples = named.as_ref().map(|n| n.triples).unwrap_or(0)
+ quad.as_ref().map(|q| q.quad_count).unwrap_or(0);
let named_graphs = named.as_ref().map(|n| n.count).unwrap_or(0)
+ quad.as_ref().map(|q| q.graph_count).unwrap_or(0);
let quad_count = default_count + named_triples;
let mut stats = BuildStats {
statements: quad_count as usize,
default_triples: default_count as usize,
named_graphs: named_graphs as usize,
terms: merged.term_count as usize,
pyramid_levels: 0,
};
let metadata = (opts.metadata)(&stats);
if !metadata.is_empty() {
trace::lap(Phase::CardDerive);
}
note_scratch(&tmp);
progress::begin(Phase::Write, None, Unit::None);
write_final_file(
output,
&metadata,
&opts.build_info,
&merged,
&perm_sections,
text_section.as_ref(),
named.as_ref(),
quad.as_ref(),
quad_count,
codec,
opts.perms,
true,
)?;
stats.pyramid_levels = 0;
note_scratch(&tmp);
trace::lap(Phase::Write);
if ext_labels.is_some() || ext_schema.is_some() {
progress::begin(Phase::LabelsSchema, None, Unit::None);
}
let label_section = match ext_labels {
Some(l) => {
let s = l
.finish(output, (budget / 4) as usize, budget / 8, codec)
.map_err(|e| ExtBuildError::Io(std::io::Error::other(e.to_string())))?;
note_scratch(&tmp);
s
}
None => None,
};
let schema_section = match ext_schema {
Some(x) => x
.finish(output)
.map_err(|e| ExtBuildError::Io(std::io::Error::other(e.to_string())))?,
None => Vec::new(),
};
if label_section.is_some() || !schema_section.is_empty() {
use crate::file::PayloadPart;
use crate::header::SectionKind;
let mut sections: Vec<(SectionKind, Vec<PayloadPart<'_>>)> = Vec::new();
if let Some(l) = &label_section {
sections.push((
SectionKind::LabelIndex,
vec![
PayloadPart::Bytes(&l.digest_and_head),
PayloadPart::File(&l.body),
],
));
eprintln!("extbuild: label index — {} entries", l.entries);
}
if !schema_section.is_empty() {
sections.push((
SectionKind::Schema,
vec![PayloadPart::Bytes(&schema_section)],
));
eprintln!("extbuild: schema section — {} bytes", schema_section.len());
}
crate::file::append_sections_to_file(output, §ions)
.map_err(|e| ExtBuildError::Io(std::io::Error::other(e.to_string())))?;
trace::lap(Phase::LabelsSchema);
}
progress::set_spill_dir(None);
Ok(stats)
}
fn note_scratch(tmp: &TmpDir) {
if !trace::active() {
return;
}
fn walk(dir: &Path) -> u64 {
let Ok(rd) = std::fs::read_dir(dir) else {
return 0;
};
rd.flatten()
.map(|e| match e.metadata() {
Ok(m) if m.is_dir() => walk(&e.path()),
Ok(m) => m.len(),
Err(_) => 0,
})
.sum()
}
trace::max("scratch_peak_bytes", walk(&tmp.dir));
}
struct TmpDir {
dir: PathBuf,
}
impl TmpDir {
fn create(parent: &Path) -> Result<Self, std::io::Error> {
static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
let seq = SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let dir = parent.join(format!(".rete-extbuild-{}-{}", std::process::id(), seq));
std::fs::create_dir_all(&dir)?;
Ok(TmpDir { dir })
}
fn path(&self, name: &str) -> PathBuf {
self.dir.join(name)
}
}
impl Drop for TmpDir {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.dir);
}
}
struct ChunkInfo {
triple_count: u64,
quad_count: u64,
section_terms: [u32; 4],
graph_terms: u32,
tri: RunFile,
qtri: RunFile,
}
impl ChunkInfo {
fn seg_counts(&self) -> SegCounts {
let [shared, subj, obj, pred] = self.section_terms;
[shared, subj, obj, pred, self.graph_terms]
}
}
struct Chunker<'a> {
tmp: &'a TmpDir,
chunk_budget: u64,
acc_bytes: u64,
working_cap: u64,
db: DictionaryBuilder,
term_sealed: u64,
count_objects: bool,
quads: Vec<(String, String, String)>,
named: Vec<(u32, String, String, String)>,
graph_ids: std::collections::BTreeMap<String, u32>,
chunks: Vec<ChunkInfo>,
has_quoted: bool,
text: Option<&'a mut crate::text_index_spill::TextSpill>,
quad: bool,
}
impl<'a> Chunker<'a> {
fn new(tmp: &'a TmpDir, chunk_budget: u64, graphs: GraphLayout) -> Self {
Chunker {
tmp,
chunk_budget,
acc_bytes: 0,
working_cap: u64::MAX,
db: Self::new_builder(false),
term_sealed: 0,
count_objects: false,
quads: Vec::new(),
named: Vec::new(),
graph_ids: std::collections::BTreeMap::new(),
chunks: Vec::new(),
has_quoted: false,
text: None,
quad: graphs == GraphLayout::Quad,
}
}
fn with_working_cap(mut self, cap: u64) -> Self {
self.working_cap = cap;
self
}
fn with_object_counts(mut self, on: bool) -> Self {
self.count_objects = on;
self.db = Self::new_builder(on);
self
}
fn new_builder(count_objects: bool) -> DictionaryBuilder {
let b = DictionaryBuilder::default();
if count_objects {
b.with_object_counts()
} else {
b
}
}
fn dict_charge(&self) -> u64 {
let (entries, bytes) = self.db.footprint();
let per_entry = DICT_ENTRY_OVERHEAD + if self.count_objects { 16 } else { 0 };
2 * bytes + per_entry * entries
}
fn push(&mut self, q: RawQuad) -> Result<(), ExtBuildError> {
let (s, p, o, g) = q;
self.acc_bytes += (s.len() + p.len() + o.len()) as u64 + PER_QUAD_OVERHEAD;
self.db.observe(&s, &p, &o);
match g {
None => self.quads.push((s, p, o)),
Some(graph) => {
let next = self.graph_ids.len() as u32 + 1;
let gid = match self.graph_ids.get(&graph) {
Some(&id) => id,
None => {
self.acc_bytes += graph.len() as u64 + PER_QUAD_OVERHEAD;
if self.quad {
self.db.observe_graph(&graph);
}
self.graph_ids.insert(graph, next);
next
}
};
self.named.push((gid, s, p, o));
}
}
if self.acc_bytes >= self.chunk_budget {
self.seal()?;
} else if self.acc_bytes + self.dict_charge() >= self.working_cap {
self.term_sealed += 1;
self.seal()?;
}
Ok(())
}
fn seal(&mut self) -> Result<(), ExtBuildError> {
if self.quads.is_empty() && self.named.is_empty() {
return Ok(());
}
let ci = self.chunks.len();
let quads = std::mem::take(&mut self.quads);
let named = std::mem::take(&mut self.named);
let graph_ids = std::mem::take(&mut self.graph_ids);
self.acc_bytes = 0;
let db = std::mem::replace(&mut self.db, Self::new_builder(self.count_objects));
let (dict, object_counts) = db.build_counted();
if self.count_objects {
if object_counts.len() != dict.object_only_count() as usize {
return Err(ExtBuildError::Internal(
"object counts do not match the object-only section",
));
}
let mut w = ZWriter::create(&self.tmp.path(&format!("c{ci}.ocnt")))?;
let mut rec = Vec::with_capacity(10);
for n in object_counts {
rec.clear();
write_uvarint(&mut rec, n);
w.write_all(&rec)?;
}
w.finish()?;
}
if dict.has_quoted_triples() {
self.has_quoted = true;
}
if let Some(text) = self.text.as_mut() {
text.spill_chunk(&self.tmp.path(&format!("c{ci}.tok")), &dict, &quads, &named)?;
}
let triple_count = quads.len() as u64;
let mut ids: Vec<IdTriple> = Vec::with_capacity(quads.len());
for (s, p, o) in quads {
ids.push(
dict.encode(&s, &p, &o)
.ok_or(ExtBuildError::Internal("chunk term missing from own dict"))?,
);
}
let tri = spill::spill_sorted_run(&self.tmp.path(&format!("c{ci}.tri")), &mut ids)?;
drop(ids);
let quad_count = named.len() as u64;
let graphs_seen = graph_ids.len();
let graph_terms = if self.quad { 0 } else { graph_ids.len() as u32 };
let mut gcol = vec![0u32; graph_ids.len() + 1];
if !named.is_empty() && self.quad {
for (name, insertion_id) in &graph_ids {
gcol[*insertion_id as usize] = dict
.subject_id(name)
.ok_or(ExtBuildError::Internal("graph name missing from own dict"))?;
}
} else if !named.is_empty() {
let mut gw = TermWriter::create(&self.tmp.path(&format!("c{ci}.graph")))?;
for (rank, (name, insertion_id)) in graph_ids.iter().enumerate() {
gcol[*insertion_id as usize] = rank as u32 + 1;
gw.push(name)?;
}
gw.finish()?;
}
let mut qids: Vec<IdQuad> = Vec::with_capacity(named.len());
for (gid, s, p, o) in named {
let (si, pi, oi) = dict
.encode(&s, &p, &o)
.ok_or(ExtBuildError::Internal("chunk term missing from own dict"))?;
qids.push((gcol[gid as usize], si, pi, oi));
}
let qtri = spill::spill_sorted_run(&self.tmp.path(&format!("c{ci}.qtri")), &mut qids)?;
drop(qids);
let shared = dict.shared_count();
let subj_only = dict.subject_only_count();
let obj_only = dict.object_only_count();
let preds = {
let mut n = 1u32;
while dict.predicate_term(n).is_some() {
n += 1;
}
n - 1
};
let write_terms = |name: &str,
count: u32,
term_of: &dyn Fn(u32) -> Option<String>|
-> Result<(), ExtBuildError> {
let mut w = TermWriter::create(&self.tmp.path(name))?;
for i in 1..=count {
let t = term_of(i).ok_or(ExtBuildError::Internal("dict id out of range"))?;
w.push(&t)?;
}
w.finish()?;
Ok(())
};
write_terms(&format!("c{ci}.shared"), shared, &|i| dict.subject_term(i))?;
write_terms(&format!("c{ci}.subj"), subj_only, &|i| {
dict.subject_term(shared + i)
})?;
write_terms(&format!("c{ci}.obj"), obj_only, &|i| {
dict.object_term(shared + i)
})?;
write_terms(&format!("c{ci}.pred"), preds, &|i| dict.predicate_term(i))?;
progress::set_stat(Stat::Chunks, ci as u64 + 1);
eprintln!(
"extbuild: chunk {ci} sealed — {} statement(s), {} term(s), {graphs_seen} graph(s)",
triple_count + quad_count,
shared + subj_only + obj_only + preds
);
self.chunks.push(ChunkInfo {
triple_count,
quad_count,
section_terms: [shared, subj_only, obj_only, preds],
graph_terms,
tri,
qtri,
});
Ok(())
}
fn finish(mut self) -> Result<Vec<ChunkInfo>, ExtBuildError> {
self.seal()?;
if self.term_sealed > 0 {
eprintln!(
"extbuild: {} of {} chunk(s) sealed by their dictionary's size before their quads' \
(a term-dense stretch)",
self.term_sealed,
self.chunks.len()
);
trace::add("chunks_term_sealed", self.term_sealed);
}
if self.chunks.is_empty() {
self.chunks.push(ChunkInfo {
triple_count: 0,
quad_count: 0,
section_terms: [0, 0, 0, 0],
graph_terms: 0,
tri: spill::write_run::<IdTriple>(&self.tmp.path("c0.tri"), &[])?,
qtri: spill::write_run::<IdQuad>(&self.tmp.path("c0.qtri"), &[])?,
});
for name in ["c0.shared", "c0.subj", "c0.obj", "c0.pred"] {
File::create(self.tmp.path(name))?;
}
}
Ok(self.chunks)
}
}
struct MergedDict {
section_files: [SectionFile; 4],
term_count: u64,
has_quoted: bool,
graph_names: Option<PathBuf>,
graph_count: u32,
schema_facts: crate::extbuild_sections::MergeFacts,
}
struct SectionFile {
path: PathBuf,
len: u64,
}
struct SpaceStream {
a: TermFileReader, b: TermFileReader, a_base: u32, b_base: u32, a_next: Option<String>,
b_next: Option<String>,
a_rank: u32,
b_rank: u32,
}
impl SpaceStream {
fn new(a: TermFileReader, b: TermFileReader, shared: u32) -> Result<Self, std::io::Error> {
let mut s = SpaceStream {
a,
b,
a_base: 0,
b_base: shared,
a_next: None,
b_next: None,
a_rank: 0,
b_rank: 0,
};
s.a_next = s.a.next()?;
s.b_next = s.b.next()?;
Ok(s)
}
fn next(&mut self) -> Result<Option<(String, u32)>, std::io::Error> {
let take_a = match (&self.a_next, &self.b_next) {
(None, None) => return Ok(None),
(Some(_), None) => true,
(None, Some(_)) => false,
(Some(x), Some(y)) => x < y,
};
if take_a {
let t = self.a_next.take().unwrap();
self.a_rank += 1;
let id = self.a_base + self.a_rank;
self.a_next = self.a.next()?;
Ok(Some((t, id)))
} else {
let t = self.b_next.take().unwrap();
self.b_rank += 1;
let id = self.b_base + self.b_rank;
self.b_next = self.b.next()?;
Ok(Some((t, id)))
}
}
}
struct TermFileReader(TermReader);
impl TermFileReader {
fn open(path: &Path) -> Result<Self, std::io::Error> {
TermReader::open(path).map(TermFileReader)
}
fn open_consuming(path: &Path) -> Result<Self, std::io::Error> {
TermReader::open_consuming(path).map(TermFileReader)
}
fn next(&mut self) -> Result<Option<String>, std::io::Error> {
self.0.next()
}
}
struct HeapEntry {
term: String,
chunk: usize,
local_id: u32,
}
impl PartialEq for HeapEntry {
fn eq(&self, other: &Self) -> bool {
self.term == other.term && self.chunk == other.chunk
}
}
impl Eq for HeapEntry {}
impl PartialOrd for HeapEntry {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
impl Ord for HeapEntry {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
other
.term
.cmp(&self.term)
.then_with(|| other.chunk.cmp(&self.chunk))
}
}
struct KWayTerms {
heap: BinaryHeap<HeapEntry>,
streams: Vec<SpaceStream>,
ticker: Ticker,
}
impl KWayTerms {
fn new(mut streams: Vec<SpaceStream>) -> Result<Self, std::io::Error> {
note_merge(streams.len());
let mut heap = BinaryHeap::new();
for (ci, s) in streams.iter_mut().enumerate() {
if let Some((term, id)) = s.next()? {
heap.push(HeapEntry {
term,
chunk: ci,
local_id: id,
});
}
}
Ok(KWayTerms {
heap,
streams,
ticker: Ticker::new(),
})
}
fn next(&mut self) -> Result<Option<TermCarriers>, std::io::Error> {
let first = match self.heap.pop() {
Some(e) => e,
None => return Ok(None),
};
let term = first.term;
let mut carriers = vec![(first.chunk, first.local_id)];
if let Some((t, id)) = self.streams[first.chunk].next()? {
self.heap.push(HeapEntry {
term: t,
chunk: first.chunk,
local_id: id,
});
}
while let Some(top) = self.heap.peek() {
if top.term != term {
break;
}
let e = self.heap.pop().unwrap();
carriers.push((e.chunk, e.local_id));
if let Some((t, id)) = self.streams[e.chunk].next()? {
self.heap.push(HeapEntry {
term: t,
chunk: e.chunk,
local_id: id,
});
}
}
self.ticker.tick(carriers.len() as u64);
Ok(Some((term, carriers)))
}
}
#[cfg(test)]
fn merge_dictionaries(
tmp: &TmpDir,
chunks: &[ChunkInfo],
) -> Result<(MergedDict, RemapStore), ExtBuildError> {
merge_dictionaries_with(tmp, chunks, &[], 64 << 20, &[])
}
struct Frequent {
count: u64,
term: String,
}
impl Frequent {
fn is_beaten_by(&self, count: u64, term: &str) -> bool {
count > self.count || (count == self.count && term < self.term.as_str())
}
}
impl PartialEq for Frequent {
fn eq(&self, other: &Self) -> bool {
self.count == other.count && self.term == other.term
}
}
impl Eq for Frequent {}
impl PartialOrd for Frequent {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
impl Ord for Frequent {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
other
.count
.cmp(&self.count)
.then_with(|| self.term.cmp(&other.term))
}
}
fn select_frequent_objects(
tmp: &TmpDir,
chunks: &[ChunkInfo],
n: usize,
) -> Result<Vec<String>, ExtBuildError> {
let empty = tmp.path("ocnt.empty");
File::create(&empty)?;
let mut subj_streams = Vec::with_capacity(chunks.len());
let mut obj_streams = Vec::with_capacity(chunks.len());
let mut counts = Vec::with_capacity(chunks.len());
for (ci, c) in chunks.iter().enumerate() {
let open = |suffix: &str| TermFileReader::open(&tmp.path(&format!("c{ci}.{suffix}")));
subj_streams.push(SpaceStream::new(
open("shared")?,
open("subj")?,
c.section_terms[0],
)?);
obj_streams.push(SpaceStream::new(
open("obj")?,
TermFileReader::open(&empty)?,
c.section_terms[2],
)?);
let path = tmp.path(&format!("c{ci}.ocnt"));
counts.push(if c.section_terms[2] > 0 {
Some(ZReader::open_consuming(&path)?)
} else {
let _ = std::fs::remove_file(&path);
None
});
}
let mut next_local = vec![1u32; chunks.len()];
let mut subjects = KWayTerms::new(subj_streams)?;
let mut objects = KWayTerms::new(obj_streams)?;
let mut subject = subjects.next()?;
let mut top: BinaryHeap<Frequent> = BinaryHeap::with_capacity(n + 1);
let mut candidates = 0u64;
while let Some((term, carriers)) = objects.next()? {
let mut count = 0u64;
for (ci, local) in carriers {
let r = counts[ci]
.as_mut()
.ok_or(ExtBuildError::Internal("object counts missing for a chunk"))?;
let c = spill::read_stream_varint(r)?.ok_or(ExtBuildError::Internal(
"object counts end before the object-only terms",
))?;
if local != next_local[ci] {
return Err(ExtBuildError::Internal(
"object-only terms out of step with their counts",
));
}
next_local[ci] += 1;
count += c;
}
while subject
.as_ref()
.is_some_and(|(s, _)| s.as_str() < term.as_str())
{
subject = subjects.next()?;
}
if subject.as_ref().is_some_and(|(s, _)| *s == term) {
continue; }
candidates += 1;
if top.len() < n {
top.push(Frequent { count, term });
} else if top
.peek()
.is_some_and(|worst| worst.is_beaten_by(count, &term))
{
top.pop();
top.push(Frequent { count, term });
}
}
for r in counts.iter_mut().flatten() {
if spill::read_stream_varint(r)?.is_some() {
return Err(ExtBuildError::Internal(
"object counts outlast the object-only terms",
));
}
}
let _ = std::fs::remove_file(&empty);
let mut promoted: Vec<Frequent> = top.into_vec();
promoted.sort_unstable_by(|a, b| a.term.cmp(&b.term));
let least = promoted.iter().map(|f| f.count).min().unwrap_or(0);
eprintln!(
"extbuild: frequent objects — {} of {candidates} object-only term(s) promoted to the \
shared section (each in at least {least} statement(s))",
promoted.len()
);
Ok(promoted.into_iter().map(|f| f.term).collect())
}
fn merge_dictionaries_with(
tmp: &TmpDir,
chunks: &[ChunkInfo],
schema_preds: &[String],
budget: u64,
promoted: &[String],
) -> Result<(MergedDict, RemapStore), ExtBuildError> {
let k = chunks.len();
let mut next_promoted = promoted.iter().peekable();
let mut schema_facts = crate::extbuild_sections::MergeFacts {
pred_ids: vec![None; schema_preds.len()],
..Default::default()
};
let counts: Vec<SegCounts> = chunks.iter().map(ChunkInfo::seg_counts).collect();
let mut remaps = RemapWriter::new(&tmp.dir, &counts, remap_buf_bytes(budget, k));
let shared_of = |ci: usize| chunks[ci].section_terms[0];
let mut subj_streams = Vec::with_capacity(k);
let mut obj_streams = Vec::with_capacity(k);
for (ci, c) in chunks.iter().enumerate() {
let open =
|suffix: &str| TermFileReader::open_consuming(&tmp.path(&format!("c{ci}.{suffix}")));
let (shared_s, shared_o) = (open("shared")?, open("shared")?);
subj_streams.push(SpaceStream::new(
shared_s,
open("subj")?,
c.section_terms[0],
)?);
obj_streams.push(SpaceStream::new(
shared_o,
open("obj")?,
c.section_terms[0],
)?);
}
let mut subjects = KWayTerms::new(subj_streams)?;
let mut objects = KWayTerms::new(obj_streams)?;
let mut shared_sec = RawSectionWriter::create(tmp.path("g.shared.raw"))?;
let mut subj_sec = RawSectionWriter::create(tmp.path("g.subj.raw"))?;
let mut obj_sec = RawSectionWriter::create(tmp.path("g.obj.raw"))?;
let mut has_quoted = false;
let mut ranks = [0u32; 3]; let mut s_item = subjects.next()?;
let mut o_item = objects.next()?;
loop {
enum Class {
Shared,
SubjOnly,
ObjOnly,
}
let class = match (&s_item, &o_item) {
(None, None) => break,
(Some(_), None) => Class::SubjOnly,
(None, Some(_)) => Class::ObjOnly,
(Some((st, _)), Some((ot, _))) => match st.cmp(ot) {
std::cmp::Ordering::Less => Class::SubjOnly,
std::cmp::Ordering::Greater => Class::ObjOnly,
std::cmp::Ordering::Equal => Class::Shared,
},
};
match class {
Class::Shared => {
let (term, s_carriers) = s_item.take().unwrap();
let (_, o_carriers) = o_item.take().unwrap();
let id = shared_id(ranks[0])?;
for (ci, lid) in s_carriers {
match lid.checked_sub(shared_of(ci)) {
None | Some(0) => remaps.push(ci, Seg::Shared, lid, id)?,
Some(rel) => remaps.push(ci, Seg::SubjOnly, rel, id)?,
}
}
for (ci, lid) in o_carriers {
match lid.checked_sub(shared_of(ci)) {
None | Some(0) => {}
Some(rel) => remaps.push(ci, Seg::ObjOnly, rel, id)?,
}
}
has_quoted |= term.starts_with("<<");
if term.starts_with('"') {
schema_facts.shared_literals += 1;
}
shared_sec.push(&term)?;
ranks[0] += 1;
s_item = subjects.next()?;
o_item = objects.next()?;
}
Class::SubjOnly => {
let (term, carriers) = s_item.take().unwrap();
let id = role_only_id(ranks[1])?;
for (ci, lid) in carriers {
match lid.checked_sub(shared_of(ci)) {
Some(rel) if rel > 0 => remaps.push(ci, Seg::SubjOnly, rel, id)?,
_ => {
return Err(ExtBuildError::Internal(
"a chunk-shared term merged as subject-only",
))
}
}
}
has_quoted |= term.starts_with("<<");
subj_sec.push(&term)?;
ranks[1] += 1;
s_item = subjects.next()?;
}
Class::ObjOnly => {
let (term, carriers) = o_item.take().unwrap();
let promote = next_promoted.next_if(|p| **p == term).is_some();
let id = if promote {
shared_id(ranks[0])?
} else {
role_only_id(ranks[2])?
};
for (ci, lid) in carriers {
match lid.checked_sub(shared_of(ci)) {
Some(rel) if rel > 0 => remaps.push(ci, Seg::ObjOnly, rel, id)?,
_ => {
return Err(ExtBuildError::Internal(
"a chunk-shared term merged as object-only",
))
}
}
}
has_quoted |= term.starts_with("<<");
if promote {
if term.starts_with('"') {
schema_facts.shared_literals += 1;
}
shared_sec.push(&term)?;
ranks[0] += 1;
} else {
if term.starts_with('"') {
schema_facts.objonly_literals += 1;
}
obj_sec.push(&term)?;
ranks[2] += 1;
}
o_item = objects.next()?;
}
}
}
if next_promoted.next().is_some() {
return Err(ExtBuildError::Internal(
"a promoted frequent object was not met as an object-only term",
));
}
drop((subjects, objects));
for seg in [Seg::Shared, Seg::SubjOnly, Seg::ObjOnly] {
remaps.flush(seg)?;
}
let (n_shared, n_subj, n_obj) = (ranks[0], ranks[1], ranks[2]);
if n_shared.checked_add(n_subj.max(n_obj)).is_none() {
return Err(ExtBuildError::Internal(
"more than 2^32 subject or object terms: the dictionary is too large for one file",
));
}
schema_facts.n_shared = n_shared;
let mut pred_sec = RawSectionWriter::create(tmp.path("g.pred.raw"))?;
{
let mut streams = Vec::with_capacity(k);
for (ci, c) in chunks.iter().enumerate() {
let empty = tmp.path(&format!("c{ci}.pred.empty"));
File::create(&empty)?;
streams.push(SpaceStream::new(
TermFileReader::open_consuming(&tmp.path(&format!("c{ci}.pred")))?,
TermFileReader::open_consuming(&empty)?,
c.section_terms[3],
)?);
}
let mut kway = KWayTerms::new(streams)?;
let mut rank = 0u32;
while let Some((term, carriers)) = kway.next()? {
for (ci, lid) in carriers {
remaps.push(ci, Seg::Pred, lid, rank + 1)?; }
if let Some(i) = schema_preds.iter().position(|w| *w == term) {
schema_facts.pred_ids[i] = Some(rank + 1);
}
if let Some(slot) = crate::label_index::slot_of_predicate(&term) {
schema_facts.label_preds.push((rank + 1, slot));
}
pred_sec.push(&term)?;
rank += 1;
}
}
remaps.flush(Seg::Pred)?;
let mut graph_count = 0u32;
let graph_names = if chunks.iter().any(|c| c.graph_terms > 0) {
let path = tmp.path("g.graphs");
let mut w = TermWriter::create(&path)?;
let mut streams = Vec::with_capacity(k);
for (ci, c) in chunks.iter().enumerate() {
let real = tmp.path(&format!("c{ci}.graph"));
if !real.exists() {
File::create(&real)?;
}
let empty = tmp.path(&format!("c{ci}.graph.empty"));
File::create(&empty)?;
streams.push(SpaceStream::new(
TermFileReader::open_consuming(&real)?,
TermFileReader::open_consuming(&empty)?,
c.graph_terms,
)?);
}
let mut kway = KWayTerms::new(streams)?;
while let Some((name, carriers)) = kway.next()? {
graph_count += 1;
for (ci, lid) in carriers {
remaps.push(ci, Seg::Graph, lid, graph_count)?;
}
w.push(&name)?;
}
w.finish()?;
Some(path)
} else {
None
};
let remaps = remaps.finish(n_shared)?;
let codec = crate::file::writer_codec();
let section_files = [
shared_sec.finish_chunked(tmp, "g.shared.sec", codec)?,
subj_sec.finish_chunked(tmp, "g.subj.sec", codec)?,
obj_sec.finish_chunked(tmp, "g.obj.sec", codec)?,
pred_sec.finish_chunked(tmp, "g.pred.sec", codec)?,
];
for ci in 0..k {
for suffix in [
"shared",
"subj",
"obj",
"pred",
"pred.empty",
"graph",
"graph.empty",
] {
let _ = std::fs::remove_file(tmp.path(&format!("c{ci}.{suffix}")));
}
}
let term_count =
n_shared as u64 + n_subj as u64 + n_obj as u64 + section_files[3].term_count as u64;
Ok((
MergedDict {
section_files: section_files.map(|s| s.file),
term_count,
has_quoted,
graph_names,
graph_count,
schema_facts,
},
remaps,
))
}
struct RawSectionWriter {
w: ZWriter,
path: PathBuf,
prev: String,
n: u64,
body_len: u64,
restarts: BufWriter<File>,
restarts_path: PathBuf,
n_restarts: u64,
restart_varint_bytes: u64,
}
struct FinishedSection {
file: SectionFile,
term_count: u32,
}
struct OffsetReader {
rd: BufReader<File>,
}
impl OffsetReader {
fn open(path: &Path) -> Result<Self, std::io::Error> {
Ok(OffsetReader {
rd: BufReader::with_capacity(1 << 16, File::open(path)?),
})
}
fn next(&mut self) -> Result<Option<u64>, std::io::Error> {
let mut b = [0u8; 8];
match self.rd.read_exact(&mut b) {
Ok(()) => Ok(Some(u64::from_le_bytes(b))),
Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => Ok(None),
Err(e) => Err(e),
}
}
}
impl RawSectionWriter {
fn create(path: PathBuf) -> Result<Self, std::io::Error> {
let restarts_path = path.with_extension("restarts");
Ok(RawSectionWriter {
w: ZWriter::create(&path)?,
path,
prev: String::new(),
n: 0,
body_len: 0,
restarts: BufWriter::with_capacity(1 << 16, File::create(&restarts_path)?),
restarts_path,
n_restarts: 0,
restart_varint_bytes: 0,
})
}
fn push(&mut self, term: &str) -> Result<(), std::io::Error> {
let r = env_restart_interval() as u64;
let mut entry = Vec::with_capacity(term.len() + 8);
if self.n.is_multiple_of(r) {
self.restarts.write_all(&self.body_len.to_le_bytes())?;
self.n_restarts += 1;
self.restart_varint_bytes += uvarint_len(self.body_len);
write_uvarint(&mut entry, 0);
write_uvarint(&mut entry, term.len() as u64);
entry.extend_from_slice(term.as_bytes());
} else {
let shared = common_prefix_len(&self.prev, term);
let suffix = &term.as_bytes()[shared..];
write_uvarint(&mut entry, shared as u64);
write_uvarint(&mut entry, suffix.len() as u64);
entry.extend_from_slice(suffix);
}
self.w.write_all(&entry)?;
self.body_len += entry.len() as u64;
self.prev.clear();
self.prev.push_str(term);
self.n += 1;
Ok(())
}
fn finish_chunked(
mut self,
tmp: &TmpDir,
out_name: &str,
codec: u8,
) -> Result<FinishedSection, ExtBuildError> {
self.w.finish()?;
self.restarts.flush()?;
drop(self.restarts);
let term_count = self.n as u32;
let interval = env_restart_interval() as u64;
let header_len = uvarint_len(self.n)
+ uvarint_len(interval)
+ uvarint_len(self.n_restarts)
+ self.restart_varint_bytes;
let budget = crate::write_opts::dict_chunk_budget() as u64;
let mut offs = OffsetReader::open(&self.restarts_path)?;
let mut body = ZReader::open(&self.path)?;
let mut at = 0u64;
let comp_path = tmp.path(&format!("{out_name}.chunks"));
let mut comp_out = BufWriter::with_capacity(1 << 20, File::create(&comp_path)?);
let dir_path = tmp.path(&format!("{out_name}.dir"));
let mut dir_out = BufWriter::with_capacity(1 << 16, File::create(&dir_path)?);
let mut dir_len = 0u64;
let mut num_chunks = 0u64;
let mut prev_run = 0u64;
let mut prev_last: Option<Vec<u8>> = None;
let mut entry = Vec::new();
let mut run = 0u64;
let mut next = offs.next()?;
while let Some(start) = next {
let first_run = run;
let mut last_run_start = start;
run += 1;
next = offs.next()?;
while let Some(o) = next {
if o - start >= budget {
break;
}
last_run_start = o;
run += 1;
next = offs.next()?;
}
let end = next.unwrap_or(self.body_len);
if start != at {
return Err(ExtBuildError::Internal(
"dictionary chunks do not tile the body",
));
}
let mut raw = vec![0u8; (end - start) as usize];
body.read_exact(&mut raw)?;
at = end;
let first_term = read_restart_term(&raw)
.ok_or(ExtBuildError::Internal("restart entry unreadable"))?;
let key = if num_chunks == 0 {
Vec::new()
} else {
match &prev_last {
None => first_term,
Some(pl) => crate::dict::shortest_separator(pl, &first_term),
}
};
if next.is_some() {
let last_run_rel = (last_run_start - start) as usize;
prev_last = crate::dict::run_last_term(&raw, last_run_rel, raw.len());
}
let comp = crate::file::compress(codec, &raw);
comp_out.write_all(&comp)?;
entry.clear();
write_uvarint(&mut entry, first_run - prev_run);
write_uvarint(&mut entry, key.len() as u64);
entry.extend_from_slice(&key);
write_uvarint(&mut entry, comp.len() as u64);
dir_out.write_all(&entry)?;
dir_len += entry.len() as u64;
prev_run = first_run;
num_chunks += 1;
}
comp_out.flush()?;
drop(comp_out);
dir_out.flush()?;
drop(dir_out);
drop(body);
let _ = std::fs::remove_file(&self.path);
let out_path = tmp.path(out_name);
let mut out = BufWriter::with_capacity(1 << 20, File::create(&out_path)?);
let mut head = Vec::new();
write_uvarint(&mut head, header_len);
write_uvarint(&mut head, self.n);
write_uvarint(&mut head, interval);
write_uvarint(&mut head, self.n_restarts);
out.write_all(&head)?;
let mut written = head.len() as u64;
let mut offs = OffsetReader::open(&self.restarts_path)?;
while let Some(off) = offs.next()? {
head.clear();
write_uvarint(&mut head, off);
out.write_all(&head)?;
written += head.len() as u64;
}
drop(offs);
let _ = std::fs::remove_file(&self.restarts_path);
if written != uvarint_len(header_len) + header_len {
return Err(ExtBuildError::Internal(
"dictionary section header length mismatch",
));
}
head.clear();
write_uvarint(&mut head, num_chunks);
out.write_all(&head)?;
written += head.len() as u64;
let mut dir_in = File::open(&dir_path)?;
written += std::io::copy(&mut dir_in, &mut out)?;
drop(dir_in);
let _ = std::fs::remove_file(&dir_path);
let mut comp_in = File::open(&comp_path)?;
let copied = std::io::copy(&mut comp_in, &mut out)?;
drop(comp_in);
out.flush()?;
let _ = std::fs::remove_file(&comp_path);
let len = written + copied;
debug_assert_eq!(
len,
uvarint_len(header_len) + header_len + uvarint_len(num_chunks) + dir_len + copied
);
Ok(FinishedSection {
file: SectionFile {
path: out_path,
len,
},
term_count,
})
}
}
fn common_prefix_len(a: &str, b: &str) -> usize {
a.as_bytes()
.iter()
.zip(b.as_bytes())
.take_while(|(x, y)| x == y)
.count()
}
fn read_restart_term(raw: &[u8]) -> Option<Vec<u8>> {
let (shared, n1) = crate::varint::read_uvarint(raw)?;
if shared != 0 {
return None;
}
let (len, n2) = crate::varint::read_uvarint(&raw[n1..])?;
let start = n1 + n2;
raw.get(start..start + len as usize).map(|s| s.to_vec())
}
#[allow(clippy::too_many_arguments)]
fn remap_chunks(
tmp: &TmpDir,
chunks: &[ChunkInfo],
remaps: &RemapStore,
quad_layout: bool,
first: crate::index::IndexPermutation,
run_len: usize,
quad_run_len: usize,
mut text: Option<&mut crate::text_index_spill::TextSpill>,
mut schema: Option<&mut ExtSchema>,
mut labels: Option<&mut ExtLabels>,
) -> Result<(Vec<RunFile>, Vec<RunFile>), ExtBuildError> {
let mut tri = RunSpiller::<IdTriple>::new(&tmp.dir, first.name(), run_len);
let mut quads: Option<RunSpiller<IdQuad>> = None;
let mut ticker = Ticker::new();
for (ci, chunk) in chunks.iter().enumerate() {
let maps = remaps.load(ci)?;
if let Some(text) = text.as_deref_mut() {
text.remap_chunk(&tmp.path(&format!("c{ci}.tok")), &maps.subj)?;
}
let mut rd = RunReader::<IdTriple>::open(chunk.tri.clone());
while let Some((s, p, o)) = rd.next()? {
let gs = maps.subj[(s - 1) as usize];
let gp = maps.pred[(p - 1) as usize];
let go = maps.obj[(o - 1) as usize];
tri.push(first.forward((gs, gp, go)))?;
ticker.tick(1);
if let Some(x) = schema.as_deref_mut() {
x.tee(gs, gp, go)?;
}
if let Some(x) = labels.as_deref_mut() {
x.tee(gs, gp, go)?;
}
}
drop(rd);
if chunk.quad_count > 0 {
let q = quads.get_or_insert_with(|| {
RunSpiller::new(&tmp.dir, &format!("q{}", QUAD_FIRST.name()), quad_run_len)
});
let gmap = if quad_layout { &maps.subj } else { &maps.graph };
let mut rd = RunReader::<IdQuad>::open(chunk.qtri.clone());
while let Some((g, s, p, o)) = rd.next()? {
q.push((
gmap[(g - 1) as usize],
maps.subj[(s - 1) as usize],
maps.pred[(p - 1) as usize],
maps.obj[(o - 1) as usize],
))?;
ticker.tick(1);
}
} else {
chunk.qtri.remove();
}
note_scratch(tmp);
}
let total = remaps.total_table_bytes();
trace::add("remap_total_bytes", total);
eprintln!(
"extbuild: remapped {} chunk(s) one at a time — {} B of id tables in all, \
at most {} B resident",
chunks.len(),
total,
remaps.largest_table_bytes()
);
let tri = tri.finish()?;
let quads = match quads {
Some(q) => q.finish()?,
None => Vec::new(),
};
Ok((tri, quads))
}
const QUAD_FIRST: crate::index::IndexPermutation = crate::index::IndexPermutation::Spo;
#[cfg(not(test))]
const SCRATCH_SAMPLE_EVERY: u64 = 1 << 24;
#[cfg(test)]
const SCRATCH_SAMPLE_EVERY: u64 = 1 << 9;
#[allow(clippy::too_many_arguments)]
fn index_permutations(
tmp: &TmpDir,
runs: Vec<RunFile>,
perms: crate::index::PermSet,
run_len: usize,
codec: u8,
prefix: &str,
observe: &mut dyn FnMut(crate::index::IndexPermutation, IdTriple) -> std::io::Result<()>,
done: &mut dyn FnMut(crate::index::IndexPermutation, u64),
) -> Result<(Vec<TiledSection>, u64), ExtBuildError> {
let order: Vec<crate::index::IndexPermutation> = perms.iter().collect();
index_chain(tmp, runs, &order, run_len, codec, prefix, observe, done)
}
#[allow(clippy::too_many_arguments)]
fn index_chain(
tmp: &TmpDir,
runs: Vec<RunFile>,
order: &[crate::index::IndexPermutation],
run_len: usize,
codec: u8,
prefix: &str,
observe: &mut dyn FnMut(crate::index::IndexPermutation, IdTriple) -> std::io::Result<()>,
done: &mut dyn FnMut(crate::index::IndexPermutation, u64),
) -> Result<(Vec<TiledSection>, u64), ExtBuildError> {
let mut runs = runs;
let mut sections = Vec::with_capacity(order.len());
let mut count: Option<u64> = None;
for (k, &perm) in order.iter().enumerate() {
let next = order.get(k + 1).copied();
let mut spill = next.map(|np| {
RunSpiller::<IdTriple>::new(&tmp.dir, &format!("{prefix}{}", np.name()), run_len)
});
let mut tiler = StreamingTiler::new(tmp, &format!("{prefix}{}", perm.name()), codec)?;
note_scratch(tmp);
let mut seen = 0u64;
let mut ticker = prefix.is_empty().then(Ticker::new);
let n = spill::merge_runs(std::mem::take(&mut runs), |t: IdTriple| {
observe(perm, t)?;
tiler.push(t)?;
if let Some(tk) = ticker.as_mut() {
tk.tick(1);
}
if let (Some(sp), Some(np)) = (spill.as_mut(), next) {
sp.push(np.forward(perm.back(t)))?;
}
seen += 1;
if seen.is_multiple_of(SCRATCH_SAMPLE_EVERY) {
note_scratch(tmp);
}
Ok::<(), ExtBuildError>(())
})?;
note_scratch(tmp);
let section = tiler.finish()?;
if let Some(sp) = spill {
runs = sp.finish()?;
}
if count.is_some_and(|c| c != n) {
return Err(ExtBuildError::Internal("permutation dedup counts diverge"));
}
count = Some(n);
sections.push(section);
drop(ticker);
done(perm, n);
}
for r in &runs {
r.remove(); }
Ok((sections, count.unwrap_or(0)))
}
type IdQuad = (u32, u32, u32, u32);
struct NamedSection {
body: PathBuf,
body_len: u64,
count: u64,
triples: u64,
}
fn uvarint_len(mut v: u64) -> u64 {
let mut n = 1;
while v >= 0x80 {
v >>= 7;
n += 1;
}
n
}
#[allow(clippy::too_many_arguments)]
fn build_named_graphs(
tmp: &TmpDir,
quad_runs: Vec<RunFile>,
graph_names: &Path,
graph_count: u32,
tri_run_len: usize,
ram_triples: usize,
perms: crate::index::PermSet,
codec: u8,
) -> Result<NamedSection, ExtBuildError> {
note_scratch(tmp);
let body_path = tmp.path("named.body");
let mut body = BufWriter::with_capacity(1 << 20, File::create(&body_path)?);
let mut names = TermFileReader::open(graph_names)?;
let mut emitted = 0u32;
let mut total: u64 = 0;
let mut body_len: u64 = 0;
let mut cur = GraphRun::new();
let mut ticker = Ticker::new();
spill::merge_runs(quad_runs, |q: IdQuad| {
ticker.tick(1);
if cur.graph != 0 && cur.graph != q.0 {
if cur.graph != emitted + 1 {
return Err(ExtBuildError::Internal(
"graph ordinals are not dense and ascending",
));
}
let done = std::mem::replace(&mut cur, GraphRun::new());
body_len += done.finish(
tmp,
&mut body,
&mut names,
tri_run_len,
perms,
codec,
&mut total,
)?;
emitted += 1;
}
cur.graph = q.0;
cur.push(tmp, (q.1, q.2, q.3), ram_triples, perms, tri_run_len)
})?;
if cur.graph != 0 {
if cur.graph != emitted + 1 {
return Err(ExtBuildError::Internal(
"graph ordinals are not dense and ascending",
));
}
body_len += cur.finish(
tmp,
&mut body,
&mut names,
tri_run_len,
perms,
codec,
&mut total,
)?;
emitted += 1;
}
body.flush()?;
drop(body);
drop(names);
let _ = std::fs::remove_file(graph_names);
if emitted != graph_count {
return Err(ExtBuildError::Internal(
"named-graph count diverges from the merged graph names",
));
}
Ok(NamedSection {
body: body_path,
body_len,
count: emitted as u64,
triples: total,
})
}
struct GraphRun {
graph: u32,
resident: Vec<IdTriple>,
spill: Option<RunSpiller<IdTriple>>,
n: u64,
}
impl GraphRun {
fn new() -> Self {
GraphRun {
graph: 0,
resident: Vec::new(),
spill: None,
n: 0,
}
}
fn push(
&mut self,
tmp: &TmpDir,
t: IdTriple,
ram_triples: usize,
perms: crate::index::PermSet,
run_len: usize,
) -> Result<(), ExtBuildError> {
self.n += 1;
if self.spill.is_none() && self.resident.len() < ram_triples {
self.resident.push(t);
return Ok(());
}
let first = perms
.iter()
.next()
.ok_or(ExtBuildError::Internal("no permutation to index"))?;
if self.spill.is_none() {
let name = format!("ng{}.{}", self.graph, first.name());
let mut sp = RunSpiller::new(&tmp.dir, &name, run_len);
for &r in &self.resident {
sp.push(first.forward(r))?;
}
self.resident = Vec::new();
self.spill = Some(sp);
}
let sp = self.spill.as_mut().expect("just opened");
sp.push(first.forward(t))?;
Ok(())
}
#[allow(clippy::too_many_arguments)]
fn finish(
self,
tmp: &TmpDir,
body: &mut BufWriter<File>,
names: &mut TermFileReader,
tri_run_len: usize,
perms: crate::index::PermSet,
codec: u8,
total: &mut u64,
) -> Result<u64, ExtBuildError> {
let iri = names
.next()?
.ok_or(ExtBuildError::Internal("graph name file exhausted"))?;
*total += self.n;
let mut head = Vec::new();
write_uvarint(&mut head, iri.len() as u64);
head.extend_from_slice(iri.as_bytes());
match self.spill {
None => {
let index = crate::GraphIndexBuilder::from_triples(self.resident)
.with_perms(perms)
.build();
let container = crate::file::encode_index_container(&index, codec);
write_uvarint(&mut head, container.len() as u64);
body.write_all(&head)?;
body.write_all(&container)?;
Ok(head.len() as u64 + container.len() as u64)
}
Some(sp) => {
let prefix = format!("ng{}.", self.graph);
let (sections, _) = index_permutations(
tmp,
sp.finish()?,
perms,
tri_run_len,
codec,
&prefix,
&mut |_, _| Ok(()),
&mut |_, _| {},
)?;
let mut clen = uvarint_len(sections.len() as u64);
for s in §ions {
clen += uvarint_len(s.len()) + s.len();
}
write_uvarint(&mut head, clen);
body.write_all(&head)?;
write_uvarint_to(body, sections.len() as u64)?;
for s in §ions {
write_uvarint_to(body, s.len())?;
s.write_to(body, None, true)?;
}
Ok(head.len() as u64 + clen)
}
}
}
}
struct QuadSectionFile {
path: PathBuf,
len: u64,
digest: [u8; 16],
graph_count: u64,
quad_count: u64,
}
fn drain_quad_tiles(
pending: &mut Vec<Vec<crate::quad::QuadRow>>,
codec: u8,
perm: crate::index::IndexPermutation,
dir: &mut crate::quad::PermDirWriter,
payload: &mut BufWriter<File>,
) -> Result<(), ExtBuildError> {
for t in crate::quad::encode_tiles(pending, codec, perm) {
dir.add(&t);
payload.write_all(&t.bytes)?;
}
pending.clear();
Ok(())
}
fn build_quad_section(
tmp: &TmpDir,
runs: Vec<RunFile>,
run_len: usize,
perms: crate::index::PermSet,
codec: u8,
) -> Result<QuadSectionFile, ExtBuildError> {
let order: Vec<crate::index::IndexPermutation> = perms.iter().collect();
if order.first() != Some(&QUAD_FIRST) {
return Err(ExtBuildError::Internal(
"the quad index's first order must be G+SPO",
));
}
let gblocks_path = tmp.path("quad.gblocks");
let mut gblocks = BufWriter::with_capacity(1 << 20, File::create(&gblocks_path)?);
let mut gdir = crate::quad::GraphDirWriter::new(codec);
let mut dirs: Vec<Vec<u8>> = Vec::with_capacity(order.len());
let mut payloads: Vec<(PathBuf, u64)> = Vec::with_capacity(order.len());
let mut quad_count: Option<u64> = None;
let mut runs = runs;
for (k, &perm) in order.iter().enumerate() {
let spo = perm == crate::index::IndexPermutation::Spo;
let next = order.get(k + 1).copied();
let mut spill =
next.map(|np| RunSpiller::<IdQuad>::new(&tmp.dir, &format!("q{}", np.name()), run_len));
let ppath = tmp.path(&format!("q{}.tiles", perm.name()));
let mut payload = BufWriter::with_capacity(1 << 20, File::create(&ppath)?);
let mut dir = crate::quad::PermDirWriter::default();
let mut tiler = crate::quad::QuadTiler::new();
let mut pending: Vec<Vec<crate::quad::QuadRow>> = Vec::new();
let mut cur: Option<(u32, u64)> = None;
note_scratch(tmp);
let mut ticker = Ticker::new();
spill::merge_runs(std::mem::take(&mut runs), |q: IdQuad| {
ticker.tick(1);
tiler.push(q, &mut |t| pending.push(t));
if pending.len() >= TILE_COMPRESS_BATCH {
drain_quad_tiles(&mut pending, codec, perm, &mut dir, &mut payload)?;
}
if spo {
cur = match cur {
Some((g, n)) if g == q.0 => Some((g, n + 1)),
Some((g, n)) => {
gdir.push(g, n, &mut gblocks)?;
Some((q.0, 1))
}
None => Some((q.0, 1)),
};
}
if let (Some(sp), Some(np)) = (spill.as_mut(), next) {
let (a, b, c) = np.forward(perm.back((q.1, q.2, q.3)));
sp.push((q.0, a, b, c))?;
}
Ok::<(), ExtBuildError>(())
})?;
note_scratch(tmp);
let n = tiler.finish(&mut |t| pending.push(t));
drain_quad_tiles(&mut pending, codec, perm, &mut dir, &mut payload)?;
if let (true, Some((g, k))) = (spo, cur) {
gdir.push(g, k, &mut gblocks)?;
}
payload.flush()?;
drop(payload);
if let Some(sp) = spill {
runs = sp.finish()?;
}
if quad_count.is_some_and(|prev| prev != n) {
return Err(ExtBuildError::Internal("quad orders' dedup counts diverge"));
}
quad_count = Some(n);
payloads.push((ppath, dir.payload_len()));
dirs.push(dir.finish());
eprintln!(
"extbuild: quad order G+{} indexed ({n} unique quad(s))",
perm.name()
);
}
let (gindex, gblocks_len, graph_count) = gdir.finish(&mut gblocks)?;
gblocks.flush()?;
drop(gblocks);
let quad_count = quad_count.unwrap_or(0);
let mut part_lens = vec![gindex.len() as u64];
part_lens.extend(dirs.iter().map(|d| d.len() as u64));
part_lens.push(gblocks_len);
part_lens.extend(payloads.iter().map(|(_, l)| *l));
let preamble = crate::quad::encode_preamble(perms, graph_count, quad_count, &part_lens);
let path = tmp.path("quad.sec");
let mut out = BufWriter::with_capacity(1 << 20, File::create(&path)?);
let mut hasher = blake3::Hasher::new();
let mut len = 0u64;
for bytes in std::iter::once(&preamble)
.chain(std::iter::once(&gindex))
.chain(dirs.iter())
{
out.write_all(bytes)?;
hasher.update(bytes);
len += bytes.len() as u64;
}
for p in std::iter::once(&gblocks_path).chain(payloads.iter().map(|(p, _)| p)) {
len += copy_hashed(p, &mut out, Some(&mut hasher))?;
let _ = std::fs::remove_file(p);
}
out.flush()?;
let mut digest = [0u8; 16];
digest.copy_from_slice(&hasher.finalize().as_bytes()[..16]);
Ok(QuadSectionFile {
path,
len,
digest,
graph_count,
quad_count,
})
}
fn write_uvarint_to(w: &mut BufWriter<File>, v: u64) -> Result<(), std::io::Error> {
let mut b = Vec::new();
write_uvarint(&mut b, v);
w.write_all(&b)
}
struct StreamingTiler {
tile_budget: usize,
codec: u8,
tile: Vec<(u32, u32, u32)>,
tile_size: usize,
group: Vec<(u32, u32, u32)>,
sizer: GroupSizer,
gtotal: usize,
prev_a: u32,
pending: Vec<Vec<(u32, u32, u32)>>,
comp_out: BufWriter<File>,
comp_path: PathBuf,
dir: Vec<TileDirEntry>,
count: u64,
}
impl StreamingTiler {
fn new(tmp: &TmpDir, name: &str, codec: u8) -> Result<Self, ExtBuildError> {
let comp_path = tmp.path(&format!("{name}.tiles"));
Ok(StreamingTiler {
tile_budget: crate::write_opts::tile_budget(),
codec,
tile: Vec::new(),
tile_size: 0,
group: Vec::new(),
sizer: GroupSizer::start(0, 0),
gtotal: 0,
prev_a: 0,
pending: Vec::new(),
comp_out: BufWriter::with_capacity(1 << 20, File::create(&comp_path)?),
comp_path,
dir: Vec::new(),
count: 0,
})
}
fn push(&mut self, t: (u32, u32, u32)) -> Result<(), ExtBuildError> {
self.count += 1;
if let Some(&(ga, _, _)) = self.group.first() {
if t.0 != ga {
self.close_group()?;
}
}
if self.group.is_empty() {
self.sizer = GroupSizer::start(t.0, self.prev_a);
}
self.group.push(t);
self.gtotal = self.sizer.push(t.1, t.2);
if self.gtotal > self.tile_budget {
self.tile.append(&mut self.group);
self.flush_tile()?;
self.tile_size = 0;
self.prev_a = t.0;
self.gtotal = 0;
}
Ok(())
}
fn close_group(&mut self) -> Result<(), ExtBuildError> {
if self.group.is_empty() {
return Ok(());
}
let a = self.group[0].0;
let gsize = self.sizer.total();
if !self.tile.is_empty() && self.tile_size + gsize > self.tile_budget {
self.flush_tile()?;
}
self.tile_size += gsize;
self.prev_a = a;
self.tile.append(&mut self.group);
Ok(())
}
fn flush_tile(&mut self) -> Result<(), ExtBuildError> {
if self.tile.is_empty() {
return Ok(());
}
self.pending.push(std::mem::take(&mut self.tile));
self.tile_size = 0;
if self.pending.len() >= TILE_COMPRESS_BATCH {
self.compress_pending()?;
}
Ok(())
}
fn compress_pending(&mut self) -> Result<(), ExtBuildError> {
if self.pending.is_empty() {
return Ok(());
}
let batch = std::mem::take(&mut self.pending);
let codec = self.codec;
let encode_one = |run: &Vec<(u32, u32, u32)>| -> (u32, u32, Vec<u8>, (u32, u32, u32, u32)) {
let mut b = TripleBlockBuilder::new();
for &t in run {
b.push(t);
}
let bytes = b.build();
let syn = match crate::triples::TripleBlock::parse(&bytes) {
Ok(blk) => {
let z = blk.zone();
(z.min_b, z.max_b, z.min_c, z.max_c)
}
Err(_) => (0, u32::MAX, 0, u32::MAX),
};
let comp = crate::file::compress(codec, &bytes);
(run[0].0, run[run.len() - 1].0, comp, syn)
};
#[cfg(feature = "parallel")]
let encoded: Vec<EncodedTile> = {
use rayon::prelude::*;
batch.par_iter().map(encode_one).collect()
};
#[cfg(not(feature = "parallel"))]
let encoded: Vec<EncodedTile> = batch.iter().map(encode_one).collect();
for (min_a, max_a, comp, syn) in encoded {
self.comp_out.write_all(&comp)?;
self.dir.push((min_a, max_a, comp.len() as u64, syn));
}
Ok(())
}
fn finish(mut self) -> Result<TiledSection, ExtBuildError> {
self.close_group()?;
self.flush_tile()?;
self.compress_pending()?;
self.comp_out.flush()?;
drop(self.comp_out);
Ok(TiledSection::from_dir(&self.dir, self.comp_path))
}
}
struct TiledSection {
head: Vec<u8>,
tiles: PathBuf,
tiles_len: u64,
trailer: Vec<u8>,
}
impl TiledSection {
fn from_dir(dir: &[TileDirEntry], tiles: PathBuf) -> Self {
let mut head = Vec::new();
write_uvarint(&mut head, dir.len() as u64);
let mut prev_min = 0u32;
let mut tiles_len = 0u64;
for &(min_a, max_a, comp_len, _) in dir {
write_uvarint(&mut head, (min_a - prev_min) as u64);
write_uvarint(&mut head, (max_a - min_a) as u64);
write_uvarint(&mut head, comp_len);
prev_min = min_a;
tiles_len += comp_len;
}
let mut trailer = Vec::new();
for &(_, _, _, (min_b, max_b, min_c, max_c)) in dir {
write_uvarint(&mut trailer, min_b as u64);
write_uvarint(&mut trailer, (max_b - min_b) as u64);
write_uvarint(&mut trailer, min_c as u64);
write_uvarint(&mut trailer, (max_c - min_c) as u64);
}
TiledSection {
head,
tiles,
tiles_len,
trailer,
}
}
fn len(&self) -> u64 {
self.head.len() as u64 + self.tiles_len + self.trailer.len() as u64
}
fn write_to(
&self,
out: &mut BufWriter<File>,
mut hasher: Option<&mut blake3::Hasher>,
consume: bool,
) -> Result<(), std::io::Error> {
out.write_all(&self.head)?;
if let Some(h) = hasher.as_deref_mut() {
h.update(&self.head);
}
let n = copy_hashed(&self.tiles, out, hasher.as_deref_mut())?;
if n != self.tiles_len {
return Err(std::io::Error::other(
"a spilled tile file has the wrong length",
));
}
out.write_all(&self.trailer)?;
if let Some(h) = hasher {
h.update(&self.trailer);
}
if consume {
let _ = std::fs::remove_file(&self.tiles);
}
Ok(())
}
}
#[allow(clippy::too_many_arguments)]
fn write_final_file(
output: &Path,
metadata: &[u8],
build_info: &[u8],
dict: &MergedDict,
perm_sections: &[TiledSection],
text: Option<&crate::text_index_spill::TextSection>,
named: Option<&NamedSection>,
quad: Option<&QuadSectionFile>,
quad_count: u64,
codec: u8,
perms: crate::index::PermSet,
consume: bool,
) -> Result<(), ExtBuildError> {
let gone = |p: &Path| {
if consume {
let _ = std::fs::remove_file(p);
}
};
use crate::header::{
Header, FLAG_HAS_QUADS, FLAG_HAS_QUOTED_TRIPLES, FLAG_TEXT_INDEX_NAMED_GRAPHS,
FLAG_TILE_SYNOPSIS, HEADER_LEN,
};
let seal = quad.map(|q| crate::quad::encode_seal(q.digest));
let mut dict_frame = Vec::new();
write_uvarint(&mut dict_frame, 4);
let mut dict_len = dict_frame.len() as u64;
let mut dict_section_heads = Vec::with_capacity(4);
for s in &dict.section_files {
let mut h = Vec::new();
write_uvarint(&mut h, s.len);
dict_len += h.len() as u64 + s.len;
dict_section_heads.push(h);
}
let mut index_frame = Vec::new();
write_uvarint(
&mut index_frame,
(perm_sections.len() + usize::from(seal.is_some())) as u64,
);
let mut index_len = index_frame.len() as u64;
let mut index_section_heads = Vec::with_capacity(perm_sections.len());
for s in perm_sections {
let mut h = Vec::new();
write_uvarint(&mut h, s.len());
index_len += h.len() as u64 + s.len();
index_section_heads.push(h);
}
let mut seal_frame = Vec::new();
if let Some(seal) = &seal {
write_uvarint(&mut seal_frame, seal.len() as u64);
seal_frame.extend_from_slice(seal);
index_len += seal_frame.len() as u64;
}
let mut named_frame = Vec::new();
let named_len = match named {
Some(n) => {
write_uvarint(&mut named_frame, n.count);
named_frame.len() as u64 + n.body_len
}
None => 0,
};
let meta_len = metadata.len() as u64;
let build_len = build_info.len() as u64;
let dict_offset = HEADER_LEN as u64 + meta_len + build_len;
let index_offset = dict_offset + dict_len;
let text_offset = index_offset + index_len;
let text_len = text.map_or(0, |t| t.len());
let named_offset = text_offset + text_len;
let mut out = BufWriter::with_capacity(1 << 20, File::create(output)?);
let mut hasher = blake3::Hasher::new();
out.write_all(&[0u8; HEADER_LEN])?;
if meta_len > 0 {
out.write_all(metadata)?;
hasher.update(metadata);
}
if build_len > 0 {
out.write_all(build_info)?;
}
let write_hashed = |out: &mut BufWriter<File>,
hasher: &mut blake3::Hasher,
bytes: &[u8]|
-> Result<(), std::io::Error> {
out.write_all(bytes)?;
hasher.update(bytes);
Ok(())
};
write_hashed(&mut out, &mut hasher, &dict_frame)?;
for (s, head) in dict.section_files.iter().zip(&dict_section_heads) {
write_hashed(&mut out, &mut hasher, head)?;
copy_hashed(&s.path, &mut out, Some(&mut hasher))?;
gone(&s.path);
}
write_hashed(&mut out, &mut hasher, &index_frame)?;
for (s, head) in perm_sections.iter().zip(&index_section_heads) {
write_hashed(&mut out, &mut hasher, head)?;
s.write_to(&mut out, Some(&mut hasher), consume)?;
}
write_hashed(&mut out, &mut hasher, &seal_frame)?;
let mut text_digest: Option<blake3::Hash> = None;
if let Some(t) = text {
if t.partial {
out.write_all(&[0u8; crate::header::TEXT_INDEX_PARTIAL_DIGEST_LEN as usize])?;
let mut digest = blake3::Hasher::new();
write_hashed(&mut out, &mut digest, &t.head)?;
copy_hashed(&t.table, &mut out, Some(&mut digest))?;
copy_hashed(&t.postings, &mut out, Some(&mut digest))?;
text_digest = Some(digest.finalize());
} else {
write_hashed(&mut out, &mut hasher, &t.head)?;
copy_hashed(&t.table, &mut out, Some(&mut hasher))?;
copy_hashed(&t.postings, &mut out, Some(&mut hasher))?;
}
gone(&t.table);
gone(&t.postings);
}
if let Some(n) = named {
write_hashed(&mut out, &mut hasher, &named_frame)?;
copy_hashed(&n.body, &mut out, Some(&mut hasher))?;
gone(&n.body);
}
if let Some(q) = quad {
copy_hashed(&q.path, &mut out, None)?;
gone(&q.path);
}
out.write_all(&crate::header::MAGIC)?; out.flush()?;
let mut hash = [0u8; 16];
hash.copy_from_slice(&hasher.finalize().as_bytes()[..16]);
let header = Header {
version: crate::header::CURRENT_FORMAT_VERSION,
flags: FLAG_TILE_SYNOPSIS
| if named_len > 0 || quad.is_some() {
FLAG_HAS_QUADS
} else {
0
}
| if quad.is_some() {
crate::header::FLAG_QUAD_INDEX
} else {
0
}
| if dict.has_quoted {
FLAG_HAS_QUOTED_TRIPLES
} else {
0
}
| if text.is_some_and(|t| t.named) {
FLAG_TEXT_INDEX_NAMED_GRAPHS
} else {
0
},
metadata_offset: HEADER_LEN as u64,
metadata_len: meta_len,
dictionary_offset: dict_offset,
dictionary_len: dict_len,
root_dir_offset: index_offset,
root_dir_len: index_len,
pyramid_meta_offset: 0,
pyramid_meta_len: 0,
dict_codec: codec,
block_codec: codec,
pyramid_levels: 0,
perms,
quad_count,
term_count: dict.term_count,
content_hash: hash,
named_graphs_offset: if named_len > 0 { named_offset } else { 0 },
named_graphs_len: named_len,
schema_meta_len: 0,
text_index_offset: if text_len > 0 { text_offset } else { 0 },
text_index_len: text_len,
text_index_partial: text.is_some_and(|t| t.partial),
build_info_offset: if build_len > 0 {
HEADER_LEN as u64 + meta_len
} else {
0
},
build_info_len: build_len,
extra_sections: match quad {
Some(q) => vec![crate::header::Section {
kind: crate::header::SectionKind::QuadIndex,
flags: 0,
offset: named_offset,
length: q.len,
}],
None => Vec::new(),
},
};
let mut f = out.into_inner().map_err(|e| e.into_error())?;
f.seek(SeekFrom::Start(0))?;
f.write_all(&header.to_bytes())?;
if let Some(d) = text_digest {
f.seek(SeekFrom::Start(text_offset))?;
f.write_all(&d.as_bytes()[..crate::header::TEXT_INDEX_PARTIAL_DIGEST_LEN as usize])?;
}
f.flush()?;
Ok(())
}
fn copy_hashed(
path: &Path,
out: &mut BufWriter<File>,
mut hasher: Option<&mut blake3::Hasher>,
) -> Result<u64, std::io::Error> {
let mut rd = File::open(path)?;
let mut buf = vec![0u8; 1 << 20];
let mut len = 0u64;
loop {
let n = rd.read(&mut buf)?;
if n == 0 {
break;
}
out.write_all(&buf[..n])?;
if let Some(h) = hasher.as_deref_mut() {
h.update(&buf[..n]);
}
len += n as u64;
}
Ok(len)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ingest;
fn test_quads(n: usize) -> Vec<RawQuad> {
let mut quads = Vec::new();
for i in 0..n {
let s = format!("<http://ex/s{}>", i % (n / 7 + 1));
let p = format!("<http://ex/p{}>", i % 13);
let o = match i % 5 {
0 => format!("<http://ex/s{}>", (i + 3) % (n / 7 + 1)), 1 => format!("\"lit {i}\""),
2 => format!("\"{}\"^^<http://www.w3.org/2001/XMLSchema#integer>", i),
3 => format!("\"v{}\"@en", i % 50),
_ => format!("_:b{}", i % 97),
};
quads.push((s, p, o, None));
}
for i in 0..(n / 20) {
let j = i * 17 % n;
quads.push(quads[j].clone());
}
quads
}
fn build_reference(quads: Vec<RawQuad>) -> Vec<u8> {
let qs = quads.clone();
let (bytes, _stats) = ingest::assemble_dataset_streaming_algo(
move |visit: &mut dyn FnMut(RawQuad)| {
for q in qs.iter().cloned() {
visit(q);
}
Ok(())
},
false,
false,
None,
crate::PyramidAlgo::Louvain,
|_, _, _| Vec::new(),
)
.unwrap();
bytes
}
fn build_ext(quads: Vec<RawQuad>, budget: u64) -> (Vec<u8>, BuildStats) {
static TEST_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
let seq = TEST_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let dir = std::env::temp_dir().join(format!(
"rete-extbuild-test-{}-{budget}-{seq}",
std::process::id()
));
std::fs::create_dir_all(&dir).unwrap();
let out = dir.join("out.rete");
let stats = build_external(
|visit| {
for q in quads.iter().cloned() {
visit(q)?;
}
Ok(())
},
&out,
ExternalBuildOptions {
memory_budget: budget,
tmp_dir: Some(dir.clone()),
metadata: Box::new(|_| Vec::new()),
build_info: Vec::new(),
perms: crate::index::PermSet::ALL,
..Default::default()
},
)
.unwrap();
let bytes = std::fs::read(&out).unwrap();
let _ = std::fs::remove_dir_all(&dir);
(bytes, stats)
}
#[test]
fn external_build_info_is_outside_the_hash() {
let quads = test_quads(500);
let (plain, _) = build_ext(quads.clone(), 0);
let dir = std::env::temp_dir().join(format!("rete-extbuild-bi-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let out = dir.join("out.rete");
let info = br#"{"built_at":"2026-08-04T00:00:00Z"}"#.to_vec();
build_external(
|visit| {
for q in quads.iter().cloned() {
visit(q)?;
}
Ok(())
},
&out,
ExternalBuildOptions {
memory_budget: 0,
tmp_dir: Some(dir.clone()),
metadata: Box::new(|_| Vec::new()),
build_info: info.clone(),
perms: crate::index::PermSet::ALL,
..Default::default()
},
)
.unwrap();
let with = std::fs::read(&out).unwrap();
let _ = std::fs::remove_dir_all(&dir);
assert_eq!(crate::read_build_info(&with).unwrap().unwrap(), info);
let h_with = crate::Header::from_bytes(&with).unwrap();
let h_plain = crate::Header::from_bytes(&plain).unwrap();
assert_eq!(h_with.content_hash, h_plain.content_hash);
assert!(crate::verify(&with).unwrap());
assert_eq!(crate::attach_build_info(&with, &[]).unwrap(), plain);
crate::Rete::open(&with).unwrap();
}
#[test]
fn external_build_is_byte_identical_to_streaming() {
let quads = test_quads(3000);
let reference = build_reference(quads.clone());
let (bytes_floor, stats) = build_ext(quads.clone(), 0);
let header = crate::Header::from_bytes(&bytes_floor).unwrap();
assert_eq!(stats.statements as u64, header.quad_count);
assert!(
stats.statements < quads.len(),
"fixture must contain duplicates for this to mean anything"
);
assert_eq!(
bytes_floor, reference,
"single-chunk external build must be byte-identical"
);
}
#[test]
fn skewed_external_build_is_byte_identical() {
let mut quads: Vec<RawQuad> = Vec::new();
for i in 0..30_000usize {
quads.push((
format!("<http://ex/s{}>", i % 500),
"<http://ex/cites>".to_string(),
format!("<http://ex/o{i}>"),
None,
));
}
for i in 0..500usize {
quads.push((
format!("<http://ex/s{i}>"),
format!("<http://ex/p{}>", i % 7),
format!("\"lit {i}\""),
None,
));
}
let reference = build_reference(quads.clone());
let (bytes, stats) = build_ext(quads.clone(), 0);
assert_eq!(stats.statements, quads.len());
assert_eq!(
bytes, reference,
"mega-group external build must be byte-identical"
);
}
#[test]
fn multi_chunk_external_build_is_byte_identical() {
let quads = test_quads(3000);
let reference = build_reference(quads.clone());
let dir = std::env::temp_dir().join(format!("rete-extbuild-mc-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let out = dir.join("out.rete");
let tmp = TmpDir::create(&dir).unwrap();
let mut chunker = Chunker::new(&tmp, 4 * 1024, GraphLayout::PerGraph); for q in quads.iter().cloned() {
chunker.push(q).unwrap();
}
let chunks = chunker.finish().unwrap();
assert!(
chunks.len() >= 4,
"expected many chunks, got {}",
chunks.len()
);
let statements: u64 = chunks.iter().map(|c| c.triple_count).sum();
let (merged, remaps) = merge_dictionaries(&tmp, &chunks).unwrap();
let codec = crate::file::writer_codec();
let (runs, _) = remap_chunks(
&tmp,
&chunks,
&remaps,
false,
crate::index::IndexPermutation::Spo,
256,
256,
None,
None,
None,
)
.unwrap();
assert!(runs.len() >= 4, "expected many runs, got {}", runs.len());
let (sections, count) = index_permutations(
&tmp,
runs,
crate::index::PermSet::ALL,
256,
codec,
"",
&mut |_, _| Ok(()),
&mut |_, _| {},
)
.unwrap();
write_final_file(
&out,
&[],
&[],
&merged,
§ions,
None,
None,
None,
count,
codec,
crate::index::PermSet::ALL,
true,
)
.unwrap();
let bytes = std::fs::read(&out).unwrap();
assert_eq!(statements as usize, quads.len());
assert_eq!(
bytes, reference,
"multi-chunk external build must be byte-identical"
);
drop(tmp);
let _ = std::fs::remove_dir_all(&dir);
}
fn build_ext_split(quads: &[RawQuad], chunk: u64) -> Vec<u8> {
TEST_SPLIT.with(|c| c.set(Some((chunk, 1 << 10))));
let (bytes, _) = build_ext(quads.to_vec(), 0);
TEST_SPLIT.with(|c| c.set(None));
bytes
}
#[test]
fn remap_memory_does_not_grow_with_the_chunk_count() {
const CHUNK: u64 = 4 * 1024;
let run = |n: usize| -> (u64, u64, u64) {
let quads = test_quads(n);
let reference = build_reference(quads.clone());
let collector = crate::build_trace::Collector::start();
let bytes = build_ext_split(&quads, CHUNK);
let t = collector.snapshot();
drop(collector);
assert_eq!(bytes, reference, "{n} quads: external build drifted");
(
t.counter("chunks").unwrap(),
t.counter("remap_total_bytes").unwrap(),
t.counter("remap_peak_bytes").unwrap(),
)
};
let (k1, total1, peak1) = run(1_000);
let (k4, total4, peak4) = run(4_000);
assert!(k1 >= 20 && k4 >= 3 * k1, "chunks: {k1}, then {k4}");
assert!(total4 >= 3 * total1, "tables: {total1} B, then {total4} B");
assert!(
total4 >= 50 * peak4,
"{total4} B in all, {peak4} B resident"
);
assert!(
peak4 <= peak1 * 3 / 2,
"resident: {peak1} B, then {peak4} B"
);
assert!(
peak4 <= (CHUNK + 1024) / 4,
"{peak4} B for a {CHUNK} B chunk"
);
}
#[test]
fn term_dense_tail_seals_by_dictionary_and_stays_identical() {
let mut quads = test_quads(2_000);
for i in 0..3_000usize {
quads.push((
format!("<http://ex/s{}>", i % 300),
"<https://w3id.org/oc/ontology/doiLink>".to_string(),
format!("<https://doi.org/10.{}/x{i}>", 1000 + i % 7),
None,
));
}
for i in 0..3_000usize {
quads.push((
format!("<http://ex/id/{i}>"),
"<http://ex/hasLiteralValue>".to_string(),
format!("\"10.{}/x{i}\"", 1000 + i % 7),
None,
));
}
let reference = build_reference(quads.clone());
let (one_chunk, _) = build_ext(quads.clone(), 0);
assert_eq!(one_chunk, reference, "the one-chunk build drifted");
let chunked = |working: Option<u64>| -> (Vec<u8>, u64, u64) {
TEST_SPLIT.with(|c| c.set(Some((64 << 10, 1 << 16))));
TEST_WORKING.with(|c| c.set(working));
let collector = crate::build_trace::Collector::start();
let (bytes, _) = build_ext(quads.clone(), 0);
let t = collector.snapshot();
drop(collector);
TEST_SPLIT.with(|c| c.set(None));
TEST_WORKING.with(|c| c.set(None));
(
bytes,
t.counter("chunks").unwrap(),
t.counter("chunks_term_sealed").unwrap_or(0),
)
};
let (by_bytes, k_bytes, sealed_bytes) = chunked(Some(u64::MAX));
assert_eq!(sealed_bytes, 0);
let (by_terms, k_terms, sealed_terms) = chunked(Some(154 << 10));
eprintln!("{k_bytes} chunks by bytes; {k_terms} with the cap, {sealed_terms} sealed by it");
assert!(k_bytes >= 4, "{k_bytes} chunk(s) by bytes");
assert!(
sealed_terms >= 8 && k_terms - sealed_terms >= 4,
"{sealed_terms} of {k_terms} chunk(s) sealed by the dictionary"
);
assert!(k_terms > k_bytes, "{k_terms} chunks, {k_bytes} by bytes");
assert_eq!(by_bytes, reference, "the byte-chunked build drifted");
assert_eq!(by_terms, reference, "the term-chunked build drifted");
}
#[test]
fn compressed_spill_peak_stays_under_one_raw_copy() {
let quads = test_quads(40_000);
let n = quads.len() as u64;
let reference = build_reference(quads.clone());
TEST_SPLIT.with(|c| c.set(Some((256 << 10, 1 << 16))));
TEST_RUNS.with(|c| c.set(Some((6_000, 6_000))));
let collector = crate::build_trace::Collector::start();
let (bytes, _) = build_ext(quads, 0);
let t = collector.snapshot();
drop(collector);
TEST_SPLIT.with(|c| c.set(None));
TEST_RUNS.with(|c| c.set(None));
assert_eq!(bytes, reference, "the compressed spill changed the file");
let peak = t.counter("scratch_peak_bytes").unwrap();
let chunks = t.counter("chunks").unwrap();
let out = bytes.len() as u64;
eprintln!(
"{n} statements, {chunks} chunks: spill peak {peak} B = {:.2} B per statement (output {out} B = {:.2})",
peak as f64 / n as f64,
out as f64 / n as f64
);
assert!(chunks >= 4, "{chunks} chunk(s)");
assert!(
peak <= out + 6 * n,
"spill peak {peak} B for {n} statements and a {out} B file"
);
}
#[test]
fn separator_keyed_dictionary_is_byte_identical_across_writers() {
let blob = "w".repeat(1200);
let quads: Vec<RawQuad> = (0..900usize)
.map(|i| {
(
format!("<http://ex/s{i:04}>"),
"<http://ex/note>".to_string(),
if i % 2 == 0 {
format!("\"{blob}{i:04}\"")
} else {
format!("\"{i:04}{blob}\"")
},
None,
)
})
.collect();
let reference = build_reference(quads.clone());
let (bytes, _) = build_ext(quads, 0);
assert_eq!(
bytes, reference,
"the external builder's chunk directory drifted from the in-RAM one"
);
let keys = crate::file::dict_chunk_keys_for_test(&bytes);
let obj = &keys[2]; assert!(
obj.len() > 4,
"object-only section has {} chunks",
obj.len()
);
assert!(obj[0].is_empty(), "chunk 0 must carry no separator");
let longest = obj.iter().map(Vec::len).max().unwrap();
assert!(
longest < 1_200,
"longest key is {longest} B — that is a stored term, not a separator"
);
let rete = crate::Rete::open(&bytes).unwrap();
for i in (0..900usize).step_by(7) {
let o = if i % 2 == 0 {
format!("\"{blob}{i:04}\"")
} else {
format!("\"{i:04}{blob}\"")
};
assert!(
rete.dictionary().object_id(&o).is_some(),
"object {i} lost by the chunk directory"
);
}
}
#[test]
fn separator_keyed_dictionary_survives_named_graphs() {
let blob = "w".repeat(1200);
let quads: Vec<RawQuad> = (0..900usize)
.map(|i| {
(
format!("<http://ex/s{i:04}>"),
"<http://ex/note>".to_string(),
if i % 2 == 0 {
format!("\"{blob}{i:04}\"")
} else {
format!("\"{i:04}{blob}\"")
},
Some(format!("<http://ex/g{:02}>", (i * 7) % 30)),
)
})
.collect();
let reference = build_reference(quads.clone());
let (bytes, stats) = build_ext(quads, 0);
assert_eq!(
bytes, reference,
"the external builder's chunk directory drifted from the in-RAM one \
once named graphs were in the input"
);
assert_eq!(stats.named_graphs, 30);
let keys = crate::file::dict_chunk_keys_for_test(&bytes);
let obj = &keys[2]; assert!(
obj.len() > 4,
"object-only section has {} chunks",
obj.len()
);
assert!(obj[0].is_empty(), "chunk 0 must carry no separator");
let longest = obj.iter().map(Vec::len).max().unwrap();
assert!(
longest < 1_200,
"longest key is {longest} B — that is a stored term, not a separator"
);
}
#[test]
fn external_build_output_is_queryable() {
let quads = test_quads(1200);
let (bytes, _) = build_ext(quads.clone(), 0);
let rete = crate::Rete::open(&bytes).unwrap();
let out = crate::eval_query(
&rete,
"SELECT (COUNT(*) AS ?n) WHERE { ?s <http://ex/p1> ?o }",
)
.unwrap();
match out {
crate::QueryOutput::Select(_, rows) => assert_eq!(rows.len(), 1),
other => panic!("expected select, got {other:?}"),
}
}
#[cfg(feature = "compression")]
fn section_from_tiles(
path: &Path,
codec: u8,
f: &mut dyn FnMut(IdTriple) -> Result<(), ExtBuildError>,
) -> Result<TiledSection, ExtBuildError> {
let mut file = File::open(path)?;
let (mut buf, mut pos, mut eof) = (Vec::new(), 0usize, false);
let mut dir: Vec<TileDirEntry> = Vec::new();
loop {
if !eof && buf.len() - pos < (4 << 20) {
buf.drain(..pos);
pos = 0;
let at = buf.len();
buf.resize(at + (16 << 20), 0);
let n = file.read(&mut buf[at..])?;
buf.truncate(at + n);
eof = n == 0;
continue;
}
if pos == buf.len() {
break;
}
let len = zstd::zstd_safe::find_frame_compressed_size(&buf[pos..])
.map_err(|_| ExtBuildError::Internal("a spilled tile is cut short"))?;
let raw = crate::file::decompress(codec, &buf[pos..pos + len])
.map_err(|_| ExtBuildError::Internal("a spilled tile does not decode"))?;
let blk = crate::triples::TripleBlock::parse(&raw)
.map_err(|_| ExtBuildError::Internal("a spilled tile does not parse"))?;
let triples = blk.triples();
let z = blk.zone();
let (Some(first), Some(last)) = (triples.first(), triples.last()) else {
return Err(ExtBuildError::Internal("an empty spilled tile"));
};
dir.push((
first.0,
last.0,
len as u64,
(z.min_b, z.max_b, z.min_c, z.max_c),
));
for t in triples {
f(t)?;
}
pos += len;
}
Ok(TiledSection::from_dir(&dir, path.to_path_buf()))
}
#[cfg(feature = "compression")]
fn resume_chain(
tmp: &TmpDir,
order: &[crate::index::IndexPermutation],
statements: u64,
run_len: usize,
codec: u8,
) -> Result<Vec<TiledSection>, ExtBuildError> {
let tiles = |p: crate::index::IndexPermutation| tmp.path(&format!("{}.tiles", p.name()));
let started = order.iter().take_while(|p| tiles(**p).exists()).count();
if order.iter().skip(started).any(|p| tiles(*p).exists()) {
return Err(ExtBuildError::Internal(
"the permutations' tiles files have gaps",
));
}
if started < 2 {
return Err(ExtBuildError::Internal(
"killed in the first permutation: its runs are half consumed, rerun the build",
));
}
let cut = started - 1;
for entry in std::fs::read_dir(&tmp.dir)?.flatten() {
let n = entry.file_name().to_string_lossy().into_owned();
let doomed = order[cut..]
.iter()
.take(2)
.any(|p| n.starts_with(&format!("{}.", p.name())));
if doomed {
let _ = std::fs::remove_file(entry.path());
}
}
let mut sections = Vec::with_capacity(order.len());
for &p in &order[..cut - 1] {
sections.push(section_from_tiles(&tiles(p), codec, &mut |_| Ok(()))?);
}
let (prev, next) = (order[cut - 1], order[cut]);
let mut spill = RunSpiller::<IdTriple>::new(&tmp.dir, next.name(), run_len);
let mut n = 0u64;
sections.push(section_from_tiles(&tiles(prev), codec, &mut |t| {
n += 1;
spill.push(next.forward(prev.back(t)))?;
Ok(())
})?);
if n != statements {
return Err(ExtBuildError::Internal(
"the last finished section does not hold every statement",
));
}
let (rest, count) = index_chain(
tmp,
spill.finish()?,
&order[cut..],
run_len,
codec,
"",
&mut |_, _| Ok(()),
&mut |p, _| eprintln!("resume: {} done", p.name()),
)?;
if count != statements {
return Err(ExtBuildError::Internal("permutation dedup counts diverge"));
}
sections.extend(rest);
Ok(sections)
}
#[test]
#[ignore = "operational tool, driven by RETE_RESUME_* env vars"]
#[cfg(feature = "compression")]
fn resume_from_spill() {
let spill = PathBuf::from(std::env::var("RETE_RESUME_SPILL").expect("RETE_RESUME_SPILL"));
let names: Vec<String> = std::fs::read_dir(&spill)
.unwrap()
.flatten()
.map(|e| e.file_name().to_string_lossy().into_owned())
.collect();
assert!(
!names
.iter()
.any(|n| n.starts_with('q') || n == "g.graphs" || n.starts_with("named.")),
"this spill has named quads; resuming it here would write a file with \
the named graphs silently missing — rerun the build instead"
);
assert!(
!names
.iter()
.any(|n| n == "global.tok" || n.starts_with("tok.run") || n.ends_with(".tok")),
"this spill has an unfinished text index (global.tok / tok.run* / c<i>.tok); \
resuming it here would write a file without it — rerun the build instead"
);
let out = PathBuf::from(std::env::var("RETE_RESUME_OUT").expect("RETE_RESUME_OUT"));
let term_count: u64 = std::env::var("RETE_RESUME_TERMS").unwrap().parse().unwrap();
let quad_count: u64 = std::env::var("RETE_RESUME_QUADS").unwrap().parse().unwrap();
let metadata: Vec<u8> = std::env::var("RETE_RESUME_CARD")
.ok()
.map(|p| std::fs::read(p).unwrap())
.unwrap_or_default();
let perms = match std::env::var("RETE_RESUME_PERMS").as_deref() {
Ok("3") => crate::index::PermSet::CORE,
_ => crate::index::PermSet::ALL,
};
let budget_mb: u64 = std::env::var("RETE_RESUME_BUDGET_MB")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(16384);
let run_len = ((budget_mb << 20) / 2 / 24) as usize;
let sec = |name: &str| -> SectionFile {
let path = spill.join(name);
let len = std::fs::metadata(&path)
.unwrap_or_else(|_| panic!("{name} is gone: the final write had begun"))
.len();
SectionFile { path, len }
};
let merged = MergedDict {
section_files: [
sec("g.shared.sec"),
sec("g.subj.sec"),
sec("g.obj.sec"),
sec("g.pred.sec"),
],
term_count,
has_quoted: false,
graph_names: None,
graph_count: 0,
schema_facts: Default::default(),
};
let tmp = TmpDir { dir: spill.clone() };
let codec = crate::file::writer_codec();
let order: Vec<_> = perms.iter().collect();
let sections = resume_chain(&tmp, &order, quad_count, run_len, codec).unwrap();
let (table, postings) = (spill.join("tok.table"), spill.join("tok.postings"));
let text = if table.exists() && postings.exists() {
let table_len = std::fs::metadata(&table).unwrap().len();
let mut head = Vec::new();
write_uvarint(&mut head, table_len);
let mut prefix = head.clone();
prefix.extend(std::fs::read(&table).unwrap());
let scope = crate::text_index::TextIndex::from_token_table(
&prefix,
codec,
Box::new(|_, _| None),
)
.expect("tok.table parses")
.scope()
.clone();
eprintln!("resume: reusing the text index");
Some(crate::text_index_spill::TextSection {
head,
table,
table_len,
postings_len: std::fs::metadata(&postings).unwrap().len(),
postings,
partial: scope.restricts_predicates(),
named: scope.covers_named_graphs(),
})
} else {
None
};
write_final_file(
&out,
&metadata,
&[],
&merged,
§ions,
text.as_ref(),
None,
None,
quad_count,
codec,
perms,
false, )
.unwrap();
eprintln!("resume: wrote {}", out.display());
std::mem::forget(tmp);
}
#[test]
#[cfg(feature = "compression")]
fn a_build_killed_mid_chain_resumes_to_the_same_file() {
let quads = test_quads(5000);
let reference = build_reference(quads.clone());
let dir = std::env::temp_dir().join(format!("rete-extbuild-resume-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let out = dir.join("out.rete");
let tmp = TmpDir::create(&dir).unwrap();
let mut chunker = Chunker::new(&tmp, 16 * 1024, GraphLayout::PerGraph);
for q in quads.iter().cloned() {
chunker.push(q).unwrap();
}
let chunks = chunker.finish().unwrap();
let (merged, remaps) = merge_dictionaries(&tmp, &chunks).unwrap();
let order: Vec<_> = crate::index::PermSet::ALL.iter().collect();
let codec = crate::file::writer_codec();
let (runs, _) = remap_chunks(
&tmp, &chunks, &remaps, false, order[0], 700, 700, None, None, None,
)
.unwrap();
let (_, n) = index_chain(
&tmp,
runs,
&order[..3],
700,
codec,
"",
&mut |_, _| Ok(()),
&mut |_, _| {},
)
.unwrap();
std::fs::write(
tmp.path(&format!("{}.tiles", order[3].name())),
[0x28, 0xb5],
)
.unwrap();
let sections = resume_chain(&tmp, &order, n, 700, codec).unwrap();
write_final_file(
&out,
&[],
&[],
&merged,
§ions,
None,
None,
None,
n,
codec,
crate::index::PermSet::ALL,
true,
)
.unwrap();
assert_eq!(
std::fs::read(&out).unwrap(),
reference,
"resumed file differs"
);
drop(tmp);
let _ = std::fs::remove_dir_all(&dir);
}
fn fedlex_shaped_quads(graphs: usize, per_graph: usize) -> Vec<RawQuad> {
let mut quads: Vec<RawQuad> = Vec::new();
for i in 0..40usize {
quads.push((
format!("<http://data.europa.eu/eli/ontology#p{i}>"),
"<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>".to_string(),
"<http://www.w3.org/2002/07/owl#FunctionalProperty>".to_string(),
None,
));
}
for g in 0..graphs {
let name = format!(
"<https://fedlex.data.admin.ch/eli/cc/{:05}/graph>",
g * 7 % 9973
);
let n = if g % 11 == 0 { 1 } else { per_graph };
for i in 0..n {
quads.push((
format!("<https://fedlex.data.admin.ch/eli/cc/{g}/art_{i}>"),
format!("<http://data.europa.eu/eli/ontology#pred{}>", i % 6),
match i % 4 {
0 => format!("<https://fedlex.data.admin.ch/eli/cc/{g}>"),
1 => format!("\"Artikel {i}\"@de"),
2 => format!("\"{i}\"^^<http://www.w3.org/2001/XMLSchema#integer>"),
_ => format!("_:b{}", i % 13),
},
Some(name.clone()),
));
}
quads.push((
"<https://fedlex.data.admin.ch/eli/cc/shared>".to_string(),
"<http://data.europa.eu/eli/ontology#in_force>".to_string(),
"\"true\"^^<http://www.w3.org/2001/XMLSchema#boolean>".to_string(),
Some(name.clone()),
));
if g == 0 {
quads.push((
"<https://fedlex.data.admin.ch/eli/cc/shared>".to_string(),
"<http://data.europa.eu/eli/ontology#in_force>".to_string(),
"\"true\"^^<http://www.w3.org/2001/XMLSchema#boolean>".to_string(),
None,
));
}
}
for i in 0..(quads.len() / 20) {
let j = i * 17 % quads.len();
quads.push(quads[j].clone());
}
quads
}
#[test]
fn named_graph_external_build_is_byte_identical() {
let quads = fedlex_shaped_quads(60, 9);
let reference = build_reference(quads.clone());
let (bytes, stats) = build_ext(quads.clone(), 0);
assert_eq!(
bytes, reference,
"named-graph external build must be byte-identical"
);
let header = crate::Header::from_bytes(&bytes).unwrap();
assert_eq!(stats.statements as u64, header.quad_count);
assert!(
stats.statements < quads.len(),
"fixture must contain duplicates for this to mean anything"
);
assert!(stats.named_graphs > 1);
assert!(stats.default_triples > 0, "default graph must be populated");
let rete = crate::Rete::open(&bytes).unwrap();
assert_eq!(rete.named_graph_count(), stats.named_graphs);
}
#[test]
fn multi_chunk_named_graph_build_is_byte_identical() {
let quads = fedlex_shaped_quads(80, 7);
let reference = build_reference(quads.clone());
let dir = std::env::temp_dir().join(format!("rete-extbuild-mcng-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let out = dir.join("out.rete");
let tmp = TmpDir::create(&dir).unwrap();
let mut chunker = Chunker::new(&tmp, 4 * 1024, GraphLayout::PerGraph); for q in quads.iter().cloned() {
chunker.push(q).unwrap();
}
let chunks = chunker.finish().unwrap();
assert!(
chunks.len() >= 8,
"expected many chunks, got {}",
chunks.len()
);
let bytes = finish_from_chunks(&tmp, &chunks, &out, 256, 4).unwrap();
assert_eq!(
bytes, reference,
"multi-chunk named-graph build must be byte-identical"
);
drop(tmp);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn oversized_named_graph_spills_and_stays_identical() {
let mut quads: Vec<RawQuad> = Vec::new();
for i in 0..4000usize {
quads.push((
format!("<http://ex/s{}>", i % 400),
"<http://ex/cites>".to_string(),
format!("<http://ex/o{i}>"),
Some("<http://ex/big>".to_string()),
));
}
for i in 0..30usize {
quads.push((
format!("<http://ex/s{i}>"),
"<http://ex/p>".to_string(),
format!("\"lit {i}\""),
Some("<http://ex/tiny>".to_string()),
));
}
quads.push((
"<http://ex/d>".into(),
"<http://ex/p>".into(),
"<http://ex/o>".into(),
None,
));
let reference = build_reference(quads.clone());
let dir = std::env::temp_dir().join(format!("rete-extbuild-big-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let out = dir.join("out.rete");
let tmp = TmpDir::create(&dir).unwrap();
let mut chunker = Chunker::new(&tmp, 64 * 1024, GraphLayout::PerGraph);
for q in quads.iter().cloned() {
chunker.push(q).unwrap();
}
let chunks = chunker.finish().unwrap();
let bytes = finish_from_chunks(&tmp, &chunks, &out, 512, 4).unwrap();
assert_eq!(
bytes, reference,
"a spilled named graph must encode identically to an in-RAM one"
);
drop(tmp);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn named_only_build_is_byte_identical() {
let quads: Vec<RawQuad> = (0..50usize)
.map(|i| {
(
format!("<http://ex/s{i}>"),
"<http://ex/p>".to_string(),
format!("\"v{i}\""),
Some(format!("<http://ex/g{}>", i % 5)),
)
})
.collect();
let reference = build_reference(quads.clone());
let (bytes, stats) = build_ext(quads, 0);
assert_eq!(bytes, reference, "named-only build must be byte-identical");
assert_eq!(stats.default_triples, 0);
assert_eq!(stats.named_graphs, 5);
let rete = crate::Rete::open(&bytes).unwrap();
assert_eq!(rete.named_graph_count(), 5);
}
#[test]
fn named_graph_output_is_queryable() {
let quads = fedlex_shaped_quads(12, 5);
let (bytes, _) = build_ext(quads.clone(), 0);
let rete = crate::Rete::open(&bytes).unwrap();
type Quad = (String, String, String, Option<String>);
let expected: std::collections::BTreeSet<Quad> = quads.iter().cloned().collect();
let mut got: std::collections::BTreeSet<Quad> = std::collections::BTreeSet::new();
for (s, p, o) in rete.dump(None) {
got.insert((s, p, o, None));
}
let names: Vec<String> = rete.graph_names().iter().map(|g| g.to_string()).collect();
for name in &names {
for (s, p, o) in rete.dump(Some(name)) {
got.insert((s, p, o, Some(name.clone())));
}
}
assert_eq!(
got, expected,
"quads lost or misfiled by the external build"
);
let out = crate::eval_query(
&rete,
"SELECT (COUNT(DISTINCT ?g) AS ?n) WHERE { GRAPH ?g { ?s ?p ?o } }",
)
.unwrap();
match out {
crate::QueryOutput::Select(_, rows) => assert_eq!(rows.len(), 1),
other => panic!("expected select, got {other:?}"),
}
}
fn finish_from_chunks(
tmp: &TmpDir,
chunks: &[ChunkInfo],
out: &Path,
run_len: usize,
ram_triples: usize,
) -> Result<Vec<u8>, ExtBuildError> {
let (merged, remaps) = merge_dictionaries(tmp, chunks)?;
let has_named = chunks.iter().any(|c| c.quad_count > 0);
let (tri_runs, quad_runs) = remap_chunks(
tmp,
chunks,
&remaps,
false,
crate::index::IndexPermutation::Spo,
run_len,
run_len,
None,
None,
None,
)?;
let codec = crate::file::writer_codec();
let perms = crate::index::PermSet::ALL;
let (sections, default_count) = index_permutations(
tmp,
tri_runs,
perms,
run_len,
codec,
"",
&mut |_, _| Ok(()),
&mut |_, _| {},
)?;
let named = if has_named {
Some(build_named_graphs(
tmp,
quad_runs,
merged.graph_names.as_deref().unwrap(),
merged.graph_count,
run_len,
ram_triples,
perms,
codec,
)?)
} else {
None
};
let quads = default_count + named.as_ref().map(|n| n.triples).unwrap_or(0);
write_final_file(
out,
&[],
&[],
&merged,
§ions,
None,
named.as_ref(),
None,
quads,
codec,
perms,
true,
)?;
Ok(std::fs::read(out)?)
}
fn text_quads() -> Vec<RawQuad> {
let mut q = fedlex_shaped_quads(40, 9);
for i in 0..600usize {
let s = format!("<http://ex/w{}>", i % 150);
q.push((
s.clone(),
"<http://purl.org/dc/terms/title>".to_string(),
format!("\"Glucose Études {} on ünïcode café\"@en", i % 37),
None,
));
q.push((
s.clone(),
"<http://purl.org/dc/terms/description>".to_string(),
format!("\"The entity {} has been modified.\"", i % 150),
None,
));
q.push((
format!("<http://ex/w{}/prov/se/1>", i % 150),
"<http://purl.org/dc/terms/description>".to_string(),
format!("\"The entity has been created ({i}).\""),
Some(format!("<http://ex/w{}/prov/>", i % 150)),
));
}
q
}
fn text_scopes() -> Vec<crate::text_index::TextIndexScope> {
use crate::text_index::{TextIndexGraphs, TextIndexScope};
let none: [&str; 0] = [];
vec![
TextIndexScope::default(),
TextIndexScope::from_options(TextIndexGraphs::All, none, none).unwrap(),
TextIndexScope::from_options(
TextIndexGraphs::Default,
none,
["http://purl.org/dc/terms/description"],
)
.unwrap(),
TextIndexScope::from_options(
TextIndexGraphs::All,
["http://purl.org/dc/terms/description"],
none,
)
.unwrap(),
]
}
fn build_reference_text(
quads: &[RawQuad],
scope: &crate::text_index::TextIndexScope,
) -> Vec<u8> {
let (bytes, _stats) = ingest::assemble_dataset_streaming_algo(
|visit: &mut dyn FnMut(RawQuad)| {
for q in quads.iter().cloned() {
visit(q);
}
Ok(())
},
false,
scope.clone(),
None,
crate::PyramidAlgo::Louvain,
|_, _, _| Vec::new(),
)
.unwrap();
bytes
}
fn build_ext_text(
quads: &[RawQuad],
scope: &crate::text_index::TextIndexScope,
split: Option<(u64, usize)>,
) -> Vec<u8> {
static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
let seq = SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let dir =
std::env::temp_dir().join(format!("rete-extbuild-text-{}-{seq}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let out = dir.join("out.rete");
TEST_SPLIT.with(|c| c.set(split));
let res = build_external(
|visit| {
for q in quads.iter().cloned() {
visit(q)?;
}
Ok(())
},
&out,
ExternalBuildOptions {
memory_budget: 0,
tmp_dir: Some(dir.clone()),
text_index: Some(scope.clone()),
..Default::default()
},
);
TEST_SPLIT.with(|c| c.set(None));
res.unwrap();
let bytes = std::fs::read(&out).unwrap();
let left: Vec<_> = std::fs::read_dir(&dir)
.unwrap()
.map(|e| e.unwrap().file_name())
.collect();
assert_eq!(left, [std::ffi::OsString::from("out.rete")]);
let _ = std::fs::remove_dir_all(&dir);
bytes
}
#[test]
fn external_text_index_is_byte_identical_to_in_ram() {
let quads = text_quads();
for scope in text_scopes() {
let reference = build_reference_text(&quads, &scope);
let rh = crate::Header::from_bytes(&reference).unwrap();
assert!(
rh.text_index_len > 0,
"{scope:?}: the fixture must index something"
);
assert_eq!(rh.text_index_partial, scope.restricts_predicates());
assert_eq!(
rh.text_index_covers_named_graphs(),
scope.covers_named_graphs()
);
for split in [None, Some((4 * 1024, 1024))] {
let bytes = build_ext_text(&quads, &scope, split);
assert_eq!(
bytes, reference,
"{scope:?}, split {split:?}: external build must be byte-identical"
);
}
}
}
#[test]
fn external_text_index_is_searchable_in_its_scope() {
let quads = text_quads();
let scopes = text_scopes();
let search = |scope: &crate::text_index::TextIndexScope, word: &str| -> usize {
let bytes = build_ext_text(&quads, scope, Some((4 * 1024, 1024)));
assert!(crate::verify(&bytes).unwrap());
let rete = crate::Rete::open(&bytes).unwrap();
assert!(rete.has_text_index());
assert_eq!(rete.text_index().unwrap().scope(), scope);
rete.text_search(&[word], None, 0).len()
};
assert_eq!(search(&scopes[0], "glucose"), 150);
assert_eq!(search(&scopes[2], "glucose"), 150);
assert_eq!(
search(&scopes[3], "glucose"),
0,
"an include-only scope drops titles"
);
assert_eq!(search(&scopes[0], "modified"), 150);
assert_eq!(search(&scopes[2], "modified"), 0, "excluded predicate");
assert_eq!(
search(&scopes[0], "created"),
0,
"named graphs not in scope"
);
assert_eq!(search(&scopes[1], "created"), 150);
assert_eq!(search(&scopes[3], "created"), 150);
assert_eq!(search(&scopes[0], "études"), 150);
}
#[test]
fn external_text_index_with_nothing_in_scope_writes_no_section() {
let quads = test_quads(400);
let none: [&str; 0] = [];
let scope = crate::text_index::TextIndexScope::from_options(
crate::text_index::TextIndexGraphs::Default,
["http://ex/never"],
none,
)
.unwrap();
let reference = build_reference_text(&quads, &scope);
let bytes = build_ext_text(&quads, &scope, None);
assert_eq!(bytes, reference);
assert_eq!(crate::Header::from_bytes(&bytes).unwrap().text_index_len, 0);
}
fn build_reference_quad(quads: Vec<RawQuad>, perms: crate::index::PermSet) -> Vec<u8> {
let (bytes, _) = ingest::assemble_dataset_streaming_with_layout(
move |visit: &mut dyn FnMut(RawQuad)| {
for q in quads.iter().cloned() {
visit(q);
}
Ok(())
},
false,
false,
None,
crate::PyramidAlgo::Louvain,
perms,
GraphLayout::Quad,
|_, _, _| Vec::new(),
)
.unwrap();
bytes
}
fn build_ext_quad(quads: Vec<RawQuad>, perms: crate::index::PermSet) -> (Vec<u8>, BuildStats) {
static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
let seq = SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let dir =
std::env::temp_dir().join(format!("rete-extbuild-quad-{}-{seq}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let out = dir.join("out.rete");
let stats = build_external(
|visit| {
for q in quads.iter().cloned() {
visit(q)?;
}
Ok(())
},
&out,
ExternalBuildOptions {
memory_budget: 0,
tmp_dir: Some(dir.clone()),
perms,
graphs: GraphLayout::Quad,
..Default::default()
},
)
.unwrap();
let bytes = std::fs::read(&out).unwrap();
let _ = std::fs::remove_dir_all(&dir);
(bytes, stats)
}
#[test]
fn quad_external_build_is_byte_identical() {
for perms in [crate::index::PermSet::ALL, crate::index::PermSet::CORE] {
let quads = fedlex_shaped_quads(60, 9);
let reference = build_reference_quad(quads.clone(), perms);
let (bytes, stats) = build_ext_quad(quads.clone(), perms);
assert_eq!(bytes, reference, "quad external build, {perms:?}");
let header = crate::Header::from_bytes(&bytes).unwrap();
assert!(header.has_quad_index());
assert_eq!(stats.statements as u64, header.quad_count);
let rete = crate::Rete::open(&bytes).unwrap();
assert_eq!(rete.named_graph_count(), stats.named_graphs);
assert!(crate::file::verify(&bytes).unwrap());
}
}
#[test]
fn multi_chunk_multi_run_quad_build_is_byte_identical() {
let quads = fedlex_shaped_quads(80, 7);
let reference = build_reference_quad(quads.clone(), crate::index::PermSet::ALL);
let dir = std::env::temp_dir().join(format!("rete-extbuild-mcq-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let out = dir.join("out.rete");
let tmp = TmpDir::create(&dir).unwrap();
let mut chunker = Chunker::new(&tmp, 4 * 1024, GraphLayout::Quad);
for q in quads.iter().cloned() {
chunker.push(q).unwrap();
}
let chunks = chunker.finish().unwrap();
assert!(
chunks.len() >= 8,
"expected many chunks, got {}",
chunks.len()
);
let (merged, remaps) = merge_dictionaries(&tmp, &chunks).unwrap();
assert_eq!(merged.graph_count, 0, "no separate graph-name merge");
let (tri_runs, quad_runs) = remap_chunks(
&tmp,
&chunks,
&remaps,
true,
crate::index::IndexPermutation::Spo,
256,
64,
None,
None,
None,
)
.unwrap();
assert!(quad_runs.len() >= 4, "{} quad run(s)", quad_runs.len());
let codec = crate::file::writer_codec();
let perms = crate::index::PermSet::ALL;
let (sections, default_count) = index_permutations(
&tmp,
tri_runs,
perms,
256,
codec,
"",
&mut |_, _| Ok(()),
&mut |_, _| {},
)
.unwrap();
let quad = build_quad_section(&tmp, quad_runs, 64, perms, codec).unwrap();
write_final_file(
&out,
&[],
&[],
&merged,
§ions,
None,
None,
Some(&quad),
default_count + quad.quad_count,
codec,
perms,
true,
)
.unwrap();
let bytes = std::fs::read(&out).unwrap();
assert_eq!(bytes, reference, "multi-chunk, multi-run quad build");
drop(tmp);
let _ = std::fs::remove_dir_all(&dir);
}
fn frequent_object_quads() -> Vec<RawQuad> {
let rdf_type = "<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>";
let mut q = test_quads(1_500);
for i in 0..3_000usize {
let s = format!("<http://ex/item/{i:05}>");
q.push((
s.clone(),
rdf_type.into(),
format!("<http://vocab/C{}>", i % 9),
None,
));
q.push((
s.clone(),
"<http://ex/status>".into(),
format!("\"st{}\"", i % 40),
None,
));
q.push((
s.clone(),
"<http://ex/doi>".into(),
format!("<https://doi.org/10.1/x{i}>"),
None,
));
let g = format!("<http://ex/g{}>", i % 3);
q.push((s, "<http://ex/inGraph>".into(), g.clone(), Some(g)));
}
for i in 0..4_000usize {
q.push((
format!("<http://ex/item/{:05}>", i % 3_000),
"<http://ex/about>".into(),
"<http://ex/hub>".into(),
None,
));
}
q.push((
"<http://ex/hub>".into(),
"<http://ex/label>".into(),
"\"hub\"".into(),
None,
));
q
}
#[test]
fn frequent_objects_external_build_is_byte_identical() {
use crate::write_opts::TEST_FREQUENT_OBJECTS;
let quads = frequent_object_quads();
let plain = build_reference(quads.clone());
for n in [1usize, 9, 64, 100_000] {
TEST_FREQUENT_OBJECTS.with(|c| c.set(Some(n)));
let reference = build_reference(quads.clone());
let (one_chunk, _) = build_ext(quads.clone(), 0);
TEST_SPLIT.with(|c| c.set(Some((16 << 10, 1 << 16))));
let collector = crate::build_trace::Collector::start();
let (chunked, _) = build_ext(quads.clone(), 0);
let chunks = collector.snapshot().counter("chunks").unwrap();
drop(collector);
let (quad_chunked, _) = build_ext_quad(quads.clone(), crate::index::PermSet::ALL);
TEST_SPLIT.with(|c| c.set(None));
let (quad_one, _) = build_ext_quad(quads.clone(), crate::index::PermSet::ALL);
let quad_reference = build_reference_quad(quads.clone(), crate::index::PermSet::ALL);
TEST_FREQUENT_OBJECTS.with(|c| c.set(None));
assert!(chunks >= 8, "{chunks} chunk(s)");
assert_ne!(reference, plain, "N = {n}: nothing was promoted");
assert_eq!(one_chunk, reference, "N = {n}: one chunk");
assert_eq!(chunked, reference, "N = {n}: {chunks} chunks");
assert_eq!(quad_one, quad_reference, "N = {n}: quad index, one chunk");
assert_eq!(quad_chunked, quad_reference, "N = {n}: quad index, chunked");
let rete = crate::Rete::open(&chunked).unwrap();
let dict = rete.dictionary();
let shared = |t: &str| dict.object_id(t).unwrap() <= dict.shared_count();
assert!(shared("<http://ex/hub>"));
if n >= 9 {
assert!(
shared("<http://ex/g0>"),
"N = {n}: the busiest object-only term"
);
}
if n == 64 {
assert!(shared("\"st39\"") && shared("<http://vocab/C8>"));
}
}
}
#[test]
fn quad_external_build_refuses_other_permutation_sets() {
let dir = std::env::temp_dir().join(format!("rete-extbuild-q5-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let r = build_external(
|visit| visit(("<a>".into(), "<p>".into(), "<b>".into(), Some("<g>".into()))),
&dir.join("out.rete"),
ExternalBuildOptions {
tmp_dir: Some(dir.clone()),
perms: crate::index::PermSet::from_bits(0b0001_1111).unwrap(),
graphs: GraphLayout::Quad,
..Default::default()
},
);
assert!(r.is_err());
let _ = std::fs::remove_dir_all(&dir);
}
}