use serde::{Deserialize, Serialize};
use rete_core::{eval_query, CountingReader, QueryOutput, RangeReader, Rete};
use super::card::ExampleQuery;
pub(crate) const BUILD_INFO_SCHEMA: u8 = 1;
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct BuildInfo {
#[serde(default)]
pub schema: u8,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub started_at: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub built_at: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub builder: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub engine: Option<Engine>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub host: Option<Host>,
#[serde(default, skip_serializing_if = "BuildParams::is_empty")]
pub params: BuildParams,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub inputs: Vec<InputRecord>,
#[serde(default, skip_serializing_if = "is_zero")]
pub inputs_omitted: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub counts: Option<Counts>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub run: Option<RunRecord>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub spill: Option<SpillRecord>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub query_costs: Option<QueryCosts>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub dropped_queries: Vec<DroppedQuery>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub recarded: Option<Recarded>,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct Recarded {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub at: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub by: Option<String>,
pub from_content_hash: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub engine: Option<Engine>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub host: Option<Host>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub input: Option<InputRecord>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub params: Option<RecardParams>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub run: Option<RunRecord>,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct RecardParams {
#[serde(default)]
pub keep_curated: bool,
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub in_place: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub card_file: Option<FileDigest>,
#[serde(default)]
pub card_costs: bool,
#[serde(default)]
pub memory_budget_mb: u64,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct Engine {
pub name: String,
pub version: String,
pub commit: String,
pub target: String,
pub profile: String,
#[serde(default)]
pub features: Vec<String>,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct Host {
pub os: String,
pub arch: String,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct InputRecord {
pub name: String,
pub format: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub compression: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub bytes: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub blake3: Option<String>,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct FileDigest {
pub name: String,
pub bytes: u64,
pub blake3: String,
}
impl FileDigest {
pub(crate) fn of(path: &str, bytes: &[u8]) -> Self {
FileDigest {
name: file_name(path),
bytes: bytes.len() as u64,
blake3: blake3::hash(bytes).to_hex().to_string(),
}
}
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct Counts {
pub statements_read: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub statements_inferred: Option<u64>,
pub statements_unique: u64,
pub duplicates_dropped: u64,
pub default_graph_triples: u64,
pub named_graphs: u64,
pub named_graph_quads: u64,
pub terms: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub invalid_iris: Option<InvalidIris>,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct InvalidIris {
pub statements: u64,
pub occurrences: u64,
pub repairable: u64,
pub unrepairable: u64,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct RunRecord {
pub wall_ms: u64,
#[serde(default)]
pub phases: Vec<PhaseTime>,
pub peak_rss_bytes: Option<u64>,
pub memory_limit_bytes: Option<u64>,
pub cpus: Option<u64>,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct PhaseTime {
pub phase: String,
pub ms: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub peak_rss_bytes: Option<u64>,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct SpillRecord {
pub chunks: u64,
pub merge_passes: u64,
pub sort_runs: u64,
pub scratch_peak_bytes: u64,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct BuildParams {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub command: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub format: Option<String>,
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub no_pyramid: bool,
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub text_index: bool,
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub label_index: bool,
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub no_schema: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub text_index_graphs: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub text_index_include: Vec<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub text_index_exclude: Vec<String>,
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub materialize: bool,
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub reason: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub pyramid_algo: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub memory_budget_mb: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub codec: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub card_top_n: Option<u32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub write: Option<WriteRecord>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub permutations: Option<u8>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub named_graphs: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub collapse_graphs: Option<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub strict: Option<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub quoted_triple_syntax: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub type_predicate: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub card_costs: Option<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub card_file: Option<FileDigest>,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct WriteRecord {
pub tile_kib: u32,
pub dict_chunk_kib: u32,
pub zstd_level: i32,
pub zstd_frame: String,
#[serde(default, skip_serializing_if = "is_zero")]
pub frequent_objects: u64,
}
fn is_zero(n: &u64) -> bool {
*n == 0
}
pub(crate) fn write_record() -> Option<WriteRecord> {
let o = rete_core::write_opts::write_options();
Some(WriteRecord {
tile_kib: (o.tile_budget / 1024) as u32,
dict_chunk_kib: (o.dict_chunk_budget / 1024) as u32,
zstd_level: o.zstd_level,
zstd_frame: match o.frame {
rete_core::write_opts::FrameShape::Sized => "sized",
rete_core::write_opts::FrameShape::Stream => "stream",
}
.to_string(),
frequent_objects: o.frequent_objects as u64,
})
}
impl BuildParams {
fn is_empty(&self) -> bool {
*self == BuildParams::default()
}
pub(crate) fn set_text_index(&mut self, scope: Option<&rete_core::TextIndexScope>) {
use rete_core::{TextIndexGraphs, TextIndexPredicates};
self.text_index = scope.is_some();
self.text_index_graphs = None;
self.text_index_include.clear();
self.text_index_exclude.clear();
let Some(scope) = scope else {
return;
};
if scope.graphs == TextIndexGraphs::All {
self.text_index_graphs = Some("all".to_string());
}
match &scope.predicates {
TextIndexPredicates::All => {}
TextIndexPredicates::Only(l) => self.text_index_include = l.clone(),
TextIndexPredicates::Except(l) => self.text_index_exclude = l.clone(),
}
}
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct QueryCosts {
pub context: CostContext,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub queries: Vec<QueryCost>,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct CostContext {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub engine: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub transport: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub note: Option<String>,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct QueryCost {
pub id: String,
pub bytes: u64,
pub requests: u64,
pub rows: u64,
pub debug_ms: u64,
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(crate) struct DroppedQuery {
pub id: String,
pub why: String,
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub contradicts_claim: bool,
}
impl BuildInfo {
pub(crate) fn to_json_bytes(&self) -> Vec<u8> {
serde_json::to_vec(self).expect("BuildInfo serializes")
}
pub(crate) fn from_json_bytes(b: &[u8]) -> anyhow::Result<Self> {
serde_json::from_slice(b).map_err(|e| anyhow::anyhow!("malformed build info: {e}"))
}
}
pub(crate) fn builder_version() -> String {
format!("rete-cli {}", env!("CARGO_PKG_VERSION"))
}
pub(crate) fn codec_name(codec: u8) -> &'static str {
match codec {
rete_core::CODEC_ZSTD => "zstd",
rete_core::CODEC_NONE => "none",
_ => "unknown",
}
}
pub(crate) fn new_build_info(params: BuildParams) -> BuildInfo {
BuildInfo {
schema: BUILD_INFO_SCHEMA,
built_at: timestamp(),
builder: Some(builder_version()),
engine: Some(engine()),
host: Some(host()),
params,
..Default::default()
}
}
pub(crate) const MAX_INPUTS_LISTED: usize = 64;
pub(crate) fn reproducible() -> bool {
std::env::var("RETE_BUILD_RECORD").is_ok_and(|v| v.trim() == "reproducible")
}
pub(crate) fn timestamp() -> Option<String> {
if reproducible() && std::env::var("SOURCE_DATE_EPOCH").is_err() {
None
} else {
Some(now_rfc3339())
}
}
pub(crate) fn engine() -> Engine {
Engine {
name: "rete-cli".into(),
version: env!("CARGO_PKG_VERSION").into(),
commit: env!("RETE_GIT_COMMIT").into(),
target: env!("RETE_TARGET").into(),
profile: env!("RETE_PROFILE").into(),
features: rete_core::build_trace::enabled_features()
.into_iter()
.map(String::from)
.collect(),
}
}
pub(crate) fn host() -> Host {
Host {
os: std::env::consts::OS.into(),
arch: std::env::consts::ARCH.into(),
}
}
pub(crate) fn file_name(path: &str) -> String {
if path == "-" {
return "stdin".into();
}
std::path::Path::new(path)
.file_name()
.map(|n| n.to_string_lossy().into_owned())
.unwrap_or_else(|| path.rsplit(['/', '\\']).next().unwrap_or(path).to_string())
}
#[derive(Clone)]
pub(crate) struct InputTap {
pub name: String,
pub format: &'static str,
pub bytes: std::rc::Rc<std::cell::Cell<u64>>,
pub hasher: std::rc::Rc<std::cell::RefCell<blake3::Hasher>>,
pub gzip: std::rc::Rc<std::cell::Cell<bool>>,
}
impl InputTap {
pub(crate) fn new(path: &str, format: &'static str) -> Self {
InputTap {
name: file_name(path),
format,
bytes: Default::default(),
hasher: Default::default(),
gzip: Default::default(),
}
}
fn record(&self) -> InputRecord {
InputRecord {
name: self.name.clone(),
format: self.format.into(),
compression: self.gzip.get().then(|| "gzip".into()),
bytes: Some(self.bytes.get()),
blake3: Some(self.hasher.borrow().finalize().to_hex().to_string()),
}
}
}
pub(crate) struct Recorder {
collector: rete_core::build_trace::Collector,
started: std::time::Instant,
started_at: Option<String>,
reproducible: bool,
pub taps: Vec<InputTap>,
pub statements_read: std::rc::Rc<std::cell::Cell<u64>>,
phase_peaks: std::cell::RefCell<Vec<(rete_core::build_trace::Phase, u64)>>,
sampler: std::cell::RefCell<Option<RssSampler>>,
}
struct RssSampler {
hwm_before: Option<u64>,
max: std::sync::Arc<std::sync::atomic::AtomicU64>,
stop: std::sync::Arc<std::sync::atomic::AtomicBool>,
thread: Option<std::thread::JoinHandle<()>>,
}
impl RssSampler {
const INTERVAL: std::time::Duration = std::time::Duration::from_millis(10);
fn start() -> Self {
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
let max = Arc::new(AtomicU64::new(current_rss_bytes().unwrap_or(0)));
let stop = Arc::new(AtomicBool::new(false));
let thread = current_rss_bytes().is_some().then(|| {
let (max, stop) = (max.clone(), stop.clone());
std::thread::spawn(move || {
while !stop.load(Ordering::Relaxed) {
if let Some(rss) = current_rss_bytes() {
max.fetch_max(rss, Ordering::Relaxed);
}
std::thread::sleep(Self::INTERVAL);
}
})
});
RssSampler {
hwm_before: peak_rss_bytes(),
max,
stop,
thread,
}
}
fn finish(mut self) -> Option<u64> {
use std::sync::atomic::Ordering;
self.stop.store(true, Ordering::Relaxed);
let sampled = self
.thread
.take()
.and_then(|t| t.join().ok())
.map(|()| self.max.load(Ordering::Relaxed))?;
let sampled = sampled.max(current_rss_bytes().unwrap_or(0));
match (self.hwm_before, peak_rss_bytes()) {
(Some(before), Some(after)) if after > before => Some(after),
_ => Some(sampled),
}
}
}
impl Drop for RssSampler {
fn drop(&mut self) {
self.stop.store(true, std::sync::atomic::Ordering::Relaxed);
}
}
fn current_rss_bytes() -> Option<u64> {
let status = std::fs::read_to_string("/proc/self/status").ok()?;
let line = status.lines().find(|l| l.starts_with("VmRSS:"))?;
let kb: u64 = line.split_whitespace().nth(1)?.parse().ok()?;
Some(kb * 1024)
}
impl Recorder {
pub(crate) fn start(inputs: &[(&str, &'static str)]) -> Self {
Recorder {
collector: rete_core::build_trace::Collector::start(),
started: std::time::Instant::now(),
started_at: timestamp(),
reproducible: reproducible(),
taps: inputs.iter().map(|(p, f)| InputTap::new(p, f)).collect(),
statements_read: Default::default(),
phase_peaks: Default::default(),
sampler: Default::default(),
}
}
pub(crate) fn lap(&self, phase: rete_core::build_trace::Phase) {
rete_core::build_trace::lap(phase);
}
pub(crate) fn begin_measured(&self) {
*self.sampler.borrow_mut() = Some(RssSampler::start());
}
pub(crate) fn lap_measured(&self, phase: rete_core::build_trace::Phase) {
self.lap(phase);
if let Some(peak) = self
.sampler
.borrow_mut()
.take()
.and_then(RssSampler::finish)
{
self.phase_peaks.borrow_mut().push((phase, peak));
}
self.begin_measured();
}
fn phase_peak(&self, phase: rete_core::build_trace::Phase) -> Option<u64> {
self.phase_peaks
.borrow()
.iter()
.filter(|(p, _)| *p == phase)
.map(|&(_, b)| b)
.max()
}
pub(crate) fn begin(&self, info: &mut BuildInfo) {
info.started_at = self.started_at.clone();
info.inputs_omitted = self.taps.len().saturating_sub(MAX_INPUTS_LISTED) as u64;
info.inputs = self
.taps
.iter()
.take(MAX_INPUTS_LISTED)
.map(|t| InputRecord {
name: t.name.clone(),
format: t.format.into(),
..Default::default()
})
.collect();
}
pub(crate) fn inputs(&self) -> Vec<InputRecord> {
self.taps
.iter()
.take(MAX_INPUTS_LISTED)
.map(InputTap::record)
.collect()
}
pub(crate) fn spill(&self) -> SpillRecord {
let t = self.collector.snapshot();
SpillRecord {
chunks: t.counter("chunks").unwrap_or(0),
merge_passes: t.counter("merge_passes").unwrap_or(0),
sort_runs: t.counter("sort_runs").unwrap_or(0),
scratch_peak_bytes: t.counter("scratch_peak_bytes").unwrap_or(0),
}
}
pub(crate) fn run(&self) -> Option<RunRecord> {
if self.reproducible {
return None;
}
Some(RunRecord {
wall_ms: self.started.elapsed().as_millis() as u64,
phases: self
.collector
.snapshot()
.phases
.into_iter()
.map(|(p, d)| PhaseTime {
phase: p.as_str().into(),
ms: d.as_millis() as u64,
peak_rss_bytes: self.phase_peak(p),
})
.collect(),
peak_rss_bytes: peak_rss_bytes(),
memory_limit_bytes: memory_limit_bytes(),
cpus: std::thread::available_parallelism()
.ok()
.map(|n| n.get() as u64),
})
}
pub(crate) fn is_reproducible(&self) -> bool {
self.reproducible
}
pub(crate) fn seal(&self, info: &mut BuildInfo) {
if !self.taps.is_empty() {
info.inputs = self.inputs();
}
info.built_at = timestamp();
info.run = self.run();
if self.reproducible {
zero_debug_ms(info);
}
}
}
pub(crate) fn zero_debug_ms(info: &mut BuildInfo) {
if let Some(c) = info.query_costs.as_mut() {
for q in &mut c.queries {
q.debug_ms = 0;
}
}
}
pub(crate) fn peak_rss_bytes() -> Option<u64> {
let status = std::fs::read_to_string("/proc/self/status").ok()?;
let line = status.lines().find(|l| l.starts_with("VmHWM:"))?;
let kb: u64 = line.split_whitespace().nth(1)?.parse().ok()?;
Some(kb * 1024)
}
pub(crate) fn memory_limit_bytes() -> Option<u64> {
let v2 = std::fs::read_to_string("/proc/self/cgroup")
.ok()
.and_then(|c| {
c.lines()
.find_map(|l| l.strip_prefix("0::").map(str::to_string))
})
.map(|rel| format!("/sys/fs/cgroup{}/memory.max", rel.trim_end_matches('/')));
for path in v2
.into_iter()
.chain(["/sys/fs/cgroup/memory.max".to_string()])
.chain(["/sys/fs/cgroup/memory/memory.limit_in_bytes".to_string()])
{
if let Ok(text) = std::fs::read_to_string(&path) {
let t = text.trim();
if t == "max" {
return None;
}
if let Ok(n) = t.parse::<u64>() {
return (n < (1u64 << 60)).then_some(n);
}
}
}
None
}
fn padded(json: Vec<u8>, len: usize) -> Vec<u8> {
let mut out = json;
if out.len() < len {
out.resize(len, b' ');
}
out
}
impl BuildInfo {
pub(crate) fn widest(&self, pending: Pending, rec: &Recorder) -> BuildInfo {
use rete_core::build_trace::Phase;
const MAX: u64 = u64::MAX;
let phases: Vec<Phase> = match pending {
Pending::Write => {
let mut p: Vec<Phase> = rec
.collector
.snapshot()
.phases
.iter()
.map(|(p, _)| *p)
.collect();
if !p.contains(&Phase::Write) {
p.push(Phase::Write);
}
p
}
};
let widest_run = || RunRecord {
wall_ms: MAX,
phases: phases
.iter()
.map(|p| PhaseTime {
phase: p.as_str().into(),
ms: MAX,
peak_rss_bytes: rec.phase_peak(*p).map(|_| MAX),
})
.collect(),
peak_rss_bytes: Some(MAX),
memory_limit_bytes: Some(MAX),
cpus: Some(MAX),
};
let run = !rec.is_reproducible();
let mut w = self.clone();
if let Some(r) = w.recarded.as_mut() {
if run {
r.run = Some(widest_run());
}
return w;
}
if run {
w.run = Some(widest_run());
}
w
}
pub(crate) fn reserve_bytes(&self, pending: Pending, rec: &Recorder) -> Vec<u8> {
self.widest(pending, rec).to_json_bytes()
}
}
#[derive(Debug, Clone, Copy)]
pub(crate) enum Pending {
Write,
}
pub(crate) fn write_image_with_build_info(
path: &std::path::Path,
image: &[u8],
info: &[u8],
) -> anyhow::Result<u64> {
use std::io::Write;
let plan = rete_core::plan_build_info(image, image.len() as u64, info.len() as u64)?;
let mut out = std::io::BufWriter::with_capacity(8 << 20, std::fs::File::create(path)?);
out.write_all(&plan.header)?;
out.write_all(&image[rete_core::format::HEADER_LEN..plan.insert as usize])?;
out.write_all(info)?;
out.write_all(&image[plan.tail_start as usize..])?;
out.flush()?;
Ok(plan.new_len)
}
pub(crate) fn counts(
stats: &rete_core::ingest::BuildStats,
read: u64,
inferred: Option<u64>,
iri: &rete_core::IriReport,
) -> Counts {
let unique = stats.statements as u64;
let default = stats.default_triples as u64;
Counts {
statements_read: read,
statements_inferred: inferred,
statements_unique: unique,
duplicates_dropped: (read + inferred.unwrap_or(0)).saturating_sub(unique),
default_graph_triples: default,
named_graphs: stats.named_graphs as u64,
named_graph_quads: unique.saturating_sub(default),
terms: stats.terms as u64,
invalid_iris: (!iri.is_empty()).then(|| InvalidIris {
statements: iri.statements(),
occurrences: iri.occurrences(),
repairable: iri.repairable(),
unrepairable: iri.unrepairable(),
}),
}
}
pub(crate) fn patch_build_info(path: &std::path::Path, info: &[u8]) -> anyhow::Result<()> {
use std::io::{Read, Seek, SeekFrom, Write};
let mut f = std::fs::OpenOptions::new()
.read(true)
.write(true)
.open(path)?;
let mut head = vec![0u8; rete_core::format::HEADER_LEN];
f.read_exact(&mut head)?;
let h = rete_core::format::Header::from_bytes(&head)?;
let reserved = h.build_info_len as usize;
if reserved == 0 || info.len() > reserved {
drop(f);
eprintln!(
"warning: the build record ({} bytes) outgrew its reservation ({reserved}); \
rewriting the file to fit it",
info.len()
);
write_build_info_streaming(path, info)?;
return Ok(());
}
f.seek(SeekFrom::Start(h.build_info_offset))?;
f.write_all(&padded(info.to_vec(), reserved))?;
f.sync_all()?;
Ok(())
}
pub(crate) fn now_rfc3339() -> String {
let secs = std::env::var("SOURCE_DATE_EPOCH")
.ok()
.and_then(|v| v.parse::<i64>().ok())
.unwrap_or_else(|| {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or(0)
});
epoch_to_rfc3339(secs)
}
pub(crate) fn epoch_to_rfc3339(secs: i64) -> String {
let (y, mo, d, h, m, s) = civil_from_epoch(secs);
format!("{y:04}-{mo:02}-{d:02}T{h:02}:{m:02}:{s:02}Z")
}
pub(crate) fn civil_from_epoch(secs: i64) -> (i64, i64, i64, i64, i64, i64) {
let days = secs.div_euclid(86_400);
let rem = secs.rem_euclid(86_400);
let (h, m, s) = (rem / 3600, (rem / 60) % 60, rem % 60);
let z = days + 719_468;
let era = z.div_euclid(146_097);
let doe = z.rem_euclid(146_097);
let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
let y = yoe + era * 400;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
let mp = (5 * doy + 2) / 153;
let d = doy - (153 * mp + 2) / 5 + 1;
let mo = if mp < 10 { mp + 3 } else { mp - 9 };
let y = if mo <= 2 { y + 1 } else { y };
(y, mo, d, h, m, s)
}
struct ImageReader(std::sync::Arc<Vec<u8>>);
impl RangeReader for ImageReader {
fn len(&self) -> u64 {
self.0.len() as u64
}
fn read_at(&self, offset: u64, len: u64) -> std::io::Result<Vec<u8>> {
let start = offset as usize;
let end = start
.checked_add(len as usize)
.filter(|&e| e <= self.0.len())
.ok_or_else(|| {
std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "range out of bounds")
})?;
Ok(self.0[start..end].to_vec())
}
}
pub(crate) struct BudgetReader<R> {
inner: R,
state: std::sync::Arc<BudgetState>,
}
struct BudgetState {
budget: u64,
limit: std::sync::atomic::AtomicU64,
spent: std::sync::atomic::AtomicU64,
}
pub(crate) struct Leash(std::sync::Arc<BudgetState>);
impl Leash {
pub(crate) fn arm(&self) {
use std::sync::atomic::Ordering::Relaxed;
let spent = self.0.spent.load(Relaxed);
self.0
.limit
.store(spent.saturating_add(self.0.budget), Relaxed);
}
}
impl<R: RangeReader> BudgetReader<R> {
pub(crate) fn new(inner: R, budget: u64) -> Self {
Self {
inner,
state: std::sync::Arc::new(BudgetState {
budget,
limit: std::sync::atomic::AtomicU64::new(budget),
spent: std::sync::atomic::AtomicU64::new(0),
}),
}
}
pub(crate) fn after_open(inner: R, budget: u64) -> (Self, Leash) {
let state = std::sync::Arc::new(BudgetState {
budget,
limit: std::sync::atomic::AtomicU64::new(u64::MAX),
spent: std::sync::atomic::AtomicU64::new(0),
});
(
Self {
inner,
state: state.clone(),
},
Leash(state),
)
}
fn charge(&self, len: u64) -> std::io::Result<()> {
use std::sync::atomic::Ordering::Relaxed;
let spent = self.state.spent.fetch_add(len, Relaxed) + len;
if spent > self.state.limit.load(Relaxed) {
return Err(std::io::Error::other(format!(
"byte budget exhausted: this query wanted more than {} bytes",
self.state.budget
)));
}
Ok(())
}
}
impl<R: RangeReader> RangeReader for BudgetReader<R> {
fn len(&self) -> u64 {
self.inner.len()
}
fn read_at(&self, offset: u64, len: u64) -> std::io::Result<Vec<u8>> {
self.charge(len)?;
self.inner.read_at(offset, len)
}
fn read_many(&self, ranges: &[(u64, u64)]) -> std::io::Result<Vec<Vec<u8>>> {
self.charge(ranges.iter().map(|&(_, l)| l).sum())?;
self.inner.read_many(ranges)
}
fn concurrency(&self) -> usize {
self.inner.concurrency()
}
}
pub(crate) struct Measured {
pub cost: QueryCost,
pub error: Option<String>,
pub useless: Option<String>,
pub vacuous: bool,
}
struct Graded {
rows: u64,
error: Option<String>,
useless: Option<String>,
vacuous: bool,
}
impl Graded {
fn failed(why: String) -> Self {
Self {
rows: 0,
error: Some(why.clone()),
useless: Some(why),
vacuous: false,
}
}
}
fn grade(out: &QueryOutput) -> Graded {
let answers = |rows: u64| Graded {
rows,
error: None,
useless: None,
vacuous: false,
};
let empty = |why: &str| Graded {
rows: 0,
error: None,
useless: Some(why.to_string()),
vacuous: false,
};
match out {
QueryOutput::Select(_, rows) if rows.is_empty() => empty("measured 0 rows on the file"),
QueryOutput::Select(_, rows) if rows.iter().all(|b| b.is_empty()) => Graded {
rows: rows.len() as u64,
error: None,
useless: Some(format!(
"measured {} row(s) binding no variable at all — an aggregate over an empty \
solution sequence, which is a row without an answer in it",
rows.len()
)),
vacuous: true,
},
QueryOutput::Select(_, rows) => answers(rows.len() as u64),
QueryOutput::Ask(false) => empty("measured ASK false on the file"),
QueryOutput::Ask(true) => answers(1),
QueryOutput::Construct(ts) if ts.is_empty() => empty("constructed no triples on the file"),
QueryOutput::Construct(ts) => answers(ts.len() as u64),
other => Graded::failed(format!("returned an unexpected result form: {other:?}")),
}
}
fn measure_one<R: RangeReader + Send + Sync + 'static>(
reader: std::sync::Arc<CountingReader<R>>,
q: &ExampleQuery,
leash: Option<Leash>,
) -> Measured {
let start = std::time::Instant::now();
let open = Rete::open_ranged_lazy(reader.clone());
if let (Ok(_), Some(l)) = (&open, &leash) {
l.arm();
}
let graded = match open {
Ok(rete) => match eval_query(&rete, &q.sparql) {
Ok(_) if rete.index_incomplete() => Graded::failed(format!(
"did not finish: a range read was refused after {} bytes (byte budget \
exhausted, or the source failed)",
reader.bytes_read()
)),
Ok(result) => grade(&result),
Err(e) => Graded::failed(format!("failed to run: {e}")),
},
Err(e) => Graded::failed(format!("the file did not open: {e}")),
};
Measured {
cost: QueryCost {
id: q.id.clone(),
bytes: reader.bytes_read(),
requests: reader.requests(),
rows: graded.rows,
debug_ms: start.elapsed().as_millis() as u64,
},
error: graded.error,
useless: graded.useless,
vacuous: graded.vacuous,
}
}
pub(crate) fn measure_query<R, F>(open: F, q: &ExampleQuery) -> anyhow::Result<Measured>
where
R: RangeReader + Send + Sync + 'static,
F: FnOnce() -> anyhow::Result<R>,
{
Ok(measure_one(
std::sync::Arc::new(CountingReader::new(open()?)),
q,
None,
))
}
pub(crate) fn measure_starter<R: RangeReader + Send + Sync + 'static>(
src: R,
q: &ExampleQuery,
) -> Measured {
let (reader, leash) = BudgetReader::after_open(src, rete_core::card::STARTER_MAX_BYTES);
measure_one(
std::sync::Arc::new(CountingReader::new(reader)),
q,
Some(leash),
)
}
pub(crate) fn measure_query_costs(
image: std::sync::Arc<Vec<u8>>,
queries: &[ExampleQuery],
) -> Vec<Measured> {
queries
.iter()
.map(|q| {
let m = measure_starter(ImageReader(image.clone()), q);
rete_core::build_progress::advance(1);
m
})
.collect()
}
pub(crate) fn keep_answering(
card: &mut super::card::DatasetCard,
measured: Vec<Measured>,
) -> (Vec<QueryCost>, Vec<DroppedQuery>) {
use rete_core::card::{claim_of_query, Claim};
let mut kept = Vec::with_capacity(measured.len());
let mut dropped = Vec::new();
for m in measured {
match m.useless {
None => kept.push(m.cost),
Some(why) => {
let sparql = card
.queries
.iter()
.find(|q| q.id == m.cost.id)
.map(|q| q.sparql.as_str())
.unwrap_or_default();
let contradicts_claim = !m.vacuous
&& m.error.is_none()
&& claim_of_query(&m.cost.id, sparql) == Claim::CannotBeEmpty;
dropped.push(DroppedQuery {
id: m.cost.id,
why,
contradicts_claim,
});
}
}
}
if !dropped.is_empty() {
eprintln!(
"dropping {} starter quer{} the measurement showed useless on this file:",
dropped.len(),
if dropped.len() == 1 { "y" } else { "ies" }
);
for d in &dropped {
eprintln!(" {:<14} {}", d.id, d.why);
if d.contradicts_claim {
eprintln!(
" {:<14} ^ GENERATOR DEFECT: its template claims this query cannot come back empty. Fix the rule in crates/rete-core/src/card_queries.rs.",
""
);
}
}
let ids: std::collections::BTreeSet<&str> = dropped.iter().map(|d| d.id.as_str()).collect();
card.queries.retain(|q| !ids.contains(q.id.as_str()));
}
(kept, dropped)
}
pub(crate) fn query_costs(queries: Vec<QueryCost>) -> QueryCosts {
QueryCosts {
context: cost_context(IN_MEMORY_TRANSPORT),
queries,
}
}
pub(crate) const IN_MEMORY_TRANSPORT: &str =
"local in-memory image; cold lazy open per query; logical range reads, no block cache";
pub(crate) const COST_NOTE: &str =
"bytes/requests are properties of file layout + query (portable); debug_ms is one machine's \
reference timing, not a guarantee";
pub(crate) fn cost_context(transport: &str) -> CostContext {
CostContext {
engine: Some(builder_version()),
transport: Some(transport.to_string()),
note: Some(COST_NOTE.to_string()),
}
}
const REWRITE_CHUNK: usize = 4 << 20;
pub(crate) fn write_build_info_streaming(
path: &std::path::Path,
info: &[u8],
) -> anyhow::Result<u64> {
use std::io::{Read, Seek, SeekFrom, Write};
let mut src = std::fs::File::open(path)?;
let file_len = src.metadata()?.len();
let mut head = vec![0u8; rete_core::format::HEADER_LEN];
src.read_exact(&mut head)?;
let before = rete_core::format::Header::from_bytes(&head)?;
let plan = rete_core::plan_build_info(&head, file_len, info.len() as u64)?;
let tmp = path.with_extension("rete.costs-tmp");
{
let mut out = std::io::BufWriter::new(std::fs::File::create(&tmp)?);
out.write_all(&plan.header)?;
let mut copy = |from: u64, to: u64, out: &mut dyn Write| -> anyhow::Result<()> {
src.seek(SeekFrom::Start(from))?;
let mut left = to - from;
let mut buf = vec![0u8; REWRITE_CHUNK.min(left.max(1) as usize)];
while left > 0 {
let n = buf.len().min(left as usize);
src.read_exact(&mut buf[..n])?;
out.write_all(&buf[..n])?;
left -= n as u64;
}
Ok(())
};
copy(rete_core::format::HEADER_LEN as u64, plan.insert, &mut out)?;
out.write_all(info)?;
copy(plan.tail_start, file_len, &mut out)?;
out.flush()?;
out.into_inner()
.map_err(|e| anyhow::anyhow!("flushing {}: {e}", tmp.display()))?
.sync_all()?;
}
drop(src);
let mut written_head = vec![0u8; rete_core::format::HEADER_LEN];
std::fs::File::open(&tmp)?.read_exact(&mut written_head)?;
let after = rete_core::format::Header::from_bytes(&written_head)
.map_err(|e| anyhow::anyhow!("the rewritten file has no readable header: {e}"))?;
if after.content_hash != before.content_hash {
let _ = std::fs::remove_file(&tmp);
anyhow::bail!("refusing to commit: the rewrite changed the content hash");
}
if after.build_info_len != info.len() as u64 {
let _ = std::fs::remove_file(&tmp);
anyhow::bail!("refusing to commit: the rewritten build-info section is the wrong length");
}
std::fs::rename(&tmp, path)?;
Ok(plan.new_len)
}
fn thousands(n: u64) -> String {
let s = n.to_string();
let mut out = String::with_capacity(s.len() + s.len() / 3);
for (i, c) in s.chars().enumerate() {
if i > 0 && (s.len() - i).is_multiple_of(3) {
out.push(',');
}
out.push(c);
}
out
}
fn human_bytes(n: u64) -> String {
const UNITS: [&str; 5] = ["B", "KiB", "MiB", "GiB", "TiB"];
let mut v = n as f64;
let mut u = 0;
while v >= 1024.0 && u + 1 < UNITS.len() {
v /= 1024.0;
u += 1;
}
if u == 0 {
format!("{n} B")
} else {
format!("{v:.1} {}", UNITS[u])
}
}
fn human_ms(ms: u64) -> String {
if ms < 1000 {
format!("{ms} ms")
} else if ms < 120_000 {
format!("{:.1} s", ms as f64 / 1000.0)
} else {
let s = ms / 1000;
format!("{} min {:02} s", s / 60, s % 60)
}
}
fn command_line(p: &BuildParams) -> String {
let mut f: Vec<String> = vec![format!(
"rete {}",
p.command
.as_deref()
.map(|c| c.trim_end_matches(" --memory-budget-mb"))
.unwrap_or("build")
)];
if let Some(v) = &p.format {
f.push(format!("--format {v}"));
}
if let Some(mb) = p.memory_budget_mb {
f.push(format!("--memory-budget-mb {mb}"));
}
if let Some(n) = p.permutations {
f.push(format!("--permutations {n}"));
}
if let Some(g) = &p.named_graphs {
f.push(format!("--named-graphs {g}"));
}
if p.collapse_graphs == Some(true) {
f.push("--collapse-graphs".into());
}
if p.no_pyramid {
f.push("--no-pyramid".into());
}
if let Some(a) = &p.pyramid_algo {
f.push(format!("--pyramid-algo {a}"));
}
if let Some(t) = &p.type_predicate {
f.push(format!("--type-predicate {t}"));
}
if p.label_index {
f.push("--label-index".into());
}
if p.no_schema {
f.push("--no-schema".into());
}
if p.text_index {
f.push("--text-index".into());
}
if let Some(g) = &p.text_index_graphs {
f.push(format!("--text-index-graphs {g}"));
}
for iri in &p.text_index_include {
f.push(format!("--text-index-include {iri}"));
}
for iri in &p.text_index_exclude {
f.push(format!("--text-index-exclude {iri}"));
}
if p.materialize {
f.push("--materialize".into());
}
if p.reason {
f.push("--reason".into());
}
if p.strict == Some(true) {
f.push("--strict".into());
}
if let Some(s) = &p.quoted_triple_syntax {
f.push(format!("--quoted-triple-syntax {s}"));
}
if let Some(w) = &p.write {
f.push(format!("--tile-kib {}", w.tile_kib));
f.push(format!("--dict-chunk-kib {}", w.dict_chunk_kib));
f.push(format!("--zstd-level {}", w.zstd_level));
if w.frequent_objects > 0 {
f.push(format!("--frequent-objects {}", w.frequent_objects));
}
}
if let Some(c) = &p.card_file {
f.push(format!("--card-file {}", c.name));
}
if p.card_costs == Some(false) {
f.push("--no-card-costs".into());
}
f.join(" ")
}
fn short_hash(h: &str) -> &str {
&h[..h.len().min(16)]
}
fn format_run(out: &mut String, run: &RunRecord, indent: &str) {
use std::fmt::Write;
let mut line = human_ms(run.wall_ms);
if let Some(c) = run.cpus {
let _ = write!(line, " on {c} CPU(s)");
}
let _ = writeln!(out, "{indent}{:<12}: {line}", "wall time");
let peak = run
.peak_rss_bytes
.map(human_bytes)
.unwrap_or_else(|| "not measured (no /proc on this OS)".into());
let limit = run
.memory_limit_bytes
.map(|l| format!(" of a {} limit", human_bytes(l)))
.unwrap_or_else(|| " (no memory limit)".into());
let _ = writeln!(out, "{indent}{:<12}: {peak}{limit}", "peak RAM");
if !run.phases.is_empty() {
let phases: Vec<String> = run
.phases
.iter()
.map(|p| match p.peak_rss_bytes {
Some(b) => format!("{} {} (peak {})", p.phase, human_ms(p.ms), human_bytes(b)),
None => format!("{} {}", p.phase, human_ms(p.ms)),
})
.collect();
for (i, chunk) in phases.chunks(4).enumerate() {
let label = if i == 0 { "phases" } else { "" };
let sep = if i == 0 { ":" } else { " " };
let _ = writeln!(out, "{indent}{label:<12}{sep} {}", chunk.join(" · "));
}
}
}
pub(crate) fn format_build_info(info: &BuildInfo) -> String {
use std::fmt::Write;
let mut out = String::new();
const I: &str = " ";
let _ = writeln!(out, " build:");
let opt = |out: &mut String, label: &str, v: &Option<String>| {
if let Some(v) = v {
let _ = writeln!(out, "{I}{label:<12}: {v}");
}
};
match (&info.started_at, &info.built_at) {
(Some(s), Some(b)) => {
let _ = writeln!(out, "{I}{:<12}: {b} (started {s})", "built at");
}
_ => opt(&mut out, "built at", &info.built_at),
}
match &info.engine {
Some(e) => {
let mut line = format!("{} {}", e.name, e.version);
let _ = write!(line, ", commit {}", short_hash(&e.commit));
let _ = write!(line, ", {} ({})", e.target, e.profile);
if !e.features.is_empty() {
let _ = write!(line, "; features: {}", e.features.join(", "));
}
let _ = writeln!(out, "{I}{:<12}: {line}", "engine");
}
None => opt(&mut out, "builder", &info.builder),
}
if let Some(h) = &info.host {
let _ = writeln!(out, "{I}{:<12}: {} {}", "host", h.os, h.arch);
}
let p = &info.params;
if !p.is_empty() {
let _ = writeln!(out, "{I}{:<12}: {}", "command", command_line(p));
if let Some(c) = &p.card_file {
let _ = writeln!(
out,
"{I}{:<12}: {} ({} B, blake3 {}…)",
"card file",
c.name,
thousands(c.bytes),
short_hash(&c.blake3)
);
}
if let Some(w) = &p.write {
let _ = writeln!(
out,
"{I}{:<12}: {}, {} KiB tiles, {} KiB dictionary chunks, zstd level {} ({} frames)",
"writer",
p.codec.as_deref().unwrap_or("?"),
w.tile_kib,
w.dict_chunk_kib,
w.zstd_level,
w.zstd_frame
);
} else if let Some(c) = &p.codec {
let _ = writeln!(out, "{I}{:<12}: {c}", "codec");
}
if let Some(n) = p.card_top_n {
let _ = writeln!(out, "{I}{:<12}: {n}", "card top-N");
}
}
for (n, i) in info.inputs.iter().enumerate() {
let mut line = format!("{} ({}", i.name, i.format);
if let Some(c) = &i.compression {
let _ = write!(line, ", {c}");
}
line.push(')');
if let Some(b) = i.bytes {
let _ = write!(line, ", {} B", thousands(b));
}
if let Some(h) = &i.blake3 {
let _ = write!(line, ", blake3 {}…", short_hash(h));
}
let label = if n == 0 { "inputs" } else { "" };
let sep = if n == 0 { ":" } else { " " };
let _ = writeln!(out, "{I}{label:<12}{sep} {line}");
}
if info.inputs_omitted > 0 {
let _ = writeln!(
out,
"{I}{:<12} … and {} more input(s), not listed",
"", info.inputs_omitted
);
}
if let Some(c) = &info.counts {
let mut line = format!("{} read", thousands(c.statements_read));
if let Some(n) = c.statements_inferred {
let _ = write!(line, " + {} inferred", thousands(n));
}
let _ = write!(
line,
", {} unique ({} duplicate(s) dropped)",
thousands(c.statements_unique),
thousands(c.duplicates_dropped)
);
let _ = writeln!(out, "{I}{:<12}: {line}", "statements");
let _ = writeln!(
out,
"{I}{:<12} {} in the default graph, {} in {} named graph(s); {} terms",
"",
thousands(c.default_graph_triples),
thousands(c.named_graph_quads),
thousands(c.named_graphs),
thousands(c.terms)
);
if let Some(b) = &c.invalid_iris {
let _ = writeln!(
out,
"{I}{:<12}: {} occurrence(s) in {} statement(s), stored verbatim ({} not \
repairable by `export --sanitize-iris`)",
"invalid IRIs",
thousands(b.occurrences),
thousands(b.statements),
thousands(b.unrepairable)
);
}
}
if let Some(run) = &info.run {
format_run(&mut out, run, I);
}
if let Some(s) = &info.spill {
let _ = writeln!(
out,
"{I}{:<12}: {} chunk(s); {} k-way merge(s) over {} sorted run(s); scratch peak {}",
"spill",
thousands(s.chunks),
thousands(s.merge_passes),
thousands(s.sort_runs),
human_bytes(s.scratch_peak_bytes)
);
}
if let Some(r) = &info.recarded {
let _ = writeln!(
out,
"{I}{:<12}: {} by {} (from content hash {})",
"re-carded",
r.at.as_deref().unwrap_or("?"),
r.engine
.as_ref()
.map(|e| format!("{} {}, commit {}", e.name, e.version, short_hash(&e.commit)))
.or_else(|| r.by.clone())
.unwrap_or_else(|| "?".into()),
r.from_content_hash
);
if let Some(i) = &r.input {
let _ = writeln!(
out,
"{I} {:<10}: {}{}",
"from",
i.name,
i.bytes
.map(|b| format!(" ({} B)", thousands(b)))
.unwrap_or_default()
);
}
if let Some(run) = &r.run {
format_run(&mut out, run, &format!("{I} "));
}
}
if let Some(costs) = &info.query_costs {
let _ = writeln!(
out,
"{I}query costs (cold; bytes/requests portable, ms = one machine's debug ref):"
);
if let Some(t) = &costs.context.transport {
let _ = writeln!(out, " measured over: {t}");
}
for c in &costs.queries {
let _ = writeln!(
out,
" {:<14} {:>9} B in {:>2} req · {:>5} row(s) · {} ms",
c.id, c.bytes, c.requests, c.rows, c.debug_ms
);
}
}
if !info.dropped_queries.is_empty() {
let _ = writeln!(
out,
"{I}starter queries generated, measured, and NOT shipped:"
);
for d in &info.dropped_queries {
let _ = writeln!(
out,
" {:<14} {}{}",
d.id,
d.why,
if d.contradicts_claim {
" [generator defect: its template claimed this could not happen]"
} else {
""
}
);
}
}
while out.ends_with('\n') {
out.pop();
}
out
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn rfc3339_formats_known_epochs() {
assert_eq!(epoch_to_rfc3339(0), "1970-01-01T00:00:00Z");
assert_eq!(epoch_to_rfc3339(951_782_400), "2000-02-29T00:00:00Z");
assert_eq!(epoch_to_rfc3339(1_785_801_600), "2026-08-04T00:00:00Z");
assert_eq!(epoch_to_rfc3339(86_399), "1970-01-01T23:59:59Z");
}
#[test]
fn build_info_round_trips_and_omits_absent_fields() {
let info = new_build_info(BuildParams {
command: Some("build".into()),
no_pyramid: true,
codec: Some(codec_name(rete_core::CODEC_ZSTD).into()),
card_top_n: Some(100),
..Default::default()
});
let bytes = info.to_json_bytes();
let back = BuildInfo::from_json_bytes(&bytes).unwrap();
assert_eq!(info, back);
let text = String::from_utf8(bytes).unwrap();
assert!(text.contains("built_at"));
assert!(text.contains("rete-cli"));
assert!(text.contains("no_pyramid"));
assert!(!text.contains("memory_budget_mb"));
assert!(!text.contains("query_costs"));
assert!(!text.contains("text_index"));
}
#[test]
fn empty_build_info_parses_from_empty_object() {
let b = BuildInfo::from_json_bytes(b"{}").unwrap();
assert!(b.built_at.is_none() && b.builder.is_none());
assert!(b.params.is_empty());
}
#[test]
fn a_sealed_record_always_fits_its_reservation() {
use rete_core::build_trace::{lap, Phase};
let rec = Recorder::start(&[("/home/someone/data/in.nq.gz", "nq"), ("-", "nq")]);
let mut info = new_build_info(BuildParams {
command: Some("build".into()),
..Default::default()
});
rec.begin(&mut info);
for p in Phase::ALL.iter().rev().filter(|p| **p != Phase::Write) {
if matches!(p, Phase::CardDerive | Phase::StarterQueries) {
rec.begin_measured();
rec.lap_measured(*p);
} else {
lap(*p);
}
}
for t in &rec.taps {
t.bytes.set(u64::MAX);
t.gzip.set(true);
}
info.inputs = rec.inputs();
info.counts = Some(Counts {
statements_read: u64::MAX,
statements_inferred: Some(u64::MAX),
statements_unique: u64::MAX,
duplicates_dropped: u64::MAX,
default_graph_triples: u64::MAX,
named_graphs: u64::MAX,
named_graph_quads: u64::MAX,
terms: u64::MAX,
invalid_iris: Some(InvalidIris {
statements: u64::MAX,
occurrences: u64::MAX,
repairable: u64::MAX,
unrepairable: u64::MAX,
}),
});
info.params.codec = Some("zstd".into());
let reserve = info.reserve_bytes(Pending::Write, &rec);
lap(Phase::Write);
rec.seal(&mut info);
if let Some(r) = info.run.as_mut() {
r.wall_ms = u64::MAX;
r.peak_rss_bytes = Some(u64::MAX);
r.memory_limit_bytes = Some(u64::MAX);
r.cpus = Some(u64::MAX);
for p in &mut r.phases {
p.ms = u64::MAX;
if p.peak_rss_bytes.is_some() {
p.peak_rss_bytes = Some(u64::MAX);
}
}
}
let sealed = info.to_json_bytes();
assert!(
sealed.len() <= reserve.len(),
"sealed {} > reserved {}",
sealed.len(),
reserve.len()
);
assert_eq!(info.inputs[0].name, "in.nq.gz");
assert_eq!(info.inputs[1].name, "stdin");
}
#[test]
fn patching_fills_the_reservation_in_place() {
let (image, _) = rete_core::ingest::assemble_dataset(
vec![(
"<http://x/a>".into(),
"<http://x/p>".into(),
"<http://x/b>".into(),
None,
)],
br#"{"title":"t"}"#,
);
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("p.rete");
let reserve = vec![b' '; 1024];
write_image_with_build_info(&path, &image, &reserve).unwrap();
assert_eq!(
std::fs::read(&path).unwrap(),
rete_core::attach_build_info(&image, &reserve).unwrap(),
"the streamed splice is attach_build_info's bytes"
);
let info = new_build_info(BuildParams {
command: Some("build".into()),
..Default::default()
});
patch_build_info(&path, &info.to_json_bytes()).unwrap();
let bytes = std::fs::read(&path).unwrap();
let section = rete_core::read_build_info(&bytes).unwrap().unwrap();
assert_eq!(section.len(), 1024, "patched in place, not rewritten");
assert!(section.ends_with(b" "));
assert_eq!(BuildInfo::from_json_bytes(§ion).unwrap(), info);
assert!(rete_core::verify(&bytes).unwrap());
assert_eq!(
rete_core::attach_build_info(&bytes, &[]).unwrap(),
rete_core::attach_build_info(&image, &[]).unwrap()
);
}
#[test]
fn every_result_shape_grades_the_same_way_for_both_callers() {
use rete_core::Binding;
let unbound = Binding::default();
let bound = Binding::from([("s".to_string(), "http://x/a".to_string())]);
let vars = vec!["s".to_string()];
let triple = || ("s".to_string(), "p".to_string(), "o".to_string());
let cases = [
(
QueryOutput::Select(vars.clone(), vec![]),
0,
true,
false,
false,
),
(
QueryOutput::Select(vars.clone(), vec![unbound.clone()]),
1,
true,
true,
false,
),
(
QueryOutput::Select(vars.clone(), vec![bound.clone(), unbound.clone()]),
2,
false,
false,
false,
),
(
QueryOutput::Select(vars.clone(), vec![bound]),
1,
false,
false,
false,
),
(QueryOutput::Ask(false), 0, true, false, false),
(QueryOutput::Ask(true), 1, false, false, false),
(QueryOutput::Construct(vec![]), 0, true, false, false),
(
QueryOutput::Construct(vec![triple()]),
1,
false,
false,
false,
),
];
for (out, rows, useless, vacuous, error) in cases {
let g = grade(&out);
let what = format!("{out:?}");
assert_eq!(g.rows, rows, "rows for {what}");
assert_eq!(g.useless.is_some(), useless, "useless for {what}");
assert_eq!(g.vacuous, vacuous, "vacuous for {what}");
assert_eq!(g.error.is_some(), error, "error for {what}");
assert_eq!(
g.useless.is_some() && !g.vacuous,
useless && !vacuous,
"zero-row emptiness for {what}"
);
}
let f = Graded::failed("the file did not open: nope".into());
assert_eq!(f.rows, 0);
assert_eq!(f.error.as_deref(), f.useless.as_deref());
assert!(
!f.vacuous,
"a failure is a zero-row emptiness, not a vacuous one"
);
}
}