qualia_core_db/query/
graph_proof.rs

1//! Bounded-memory proof of the encoded graph represented by an N-Triples source
2//! and a Q42 volume.
3//!
4//! The `qualia-cli ingest semantic` pipeline stores the default graph as hashed
5//! `(subject, predicate, object, context)` records.  This module compares those
6//! records exactly as a *set*, using sorted fixed-width disk runs rather than an
7//! in-memory graph or a lossy aggregate checksum.
8//! Blank-node graphs require lexical Q42 terms for canonical isomorphism.
9
10use std::cmp::Ordering;
11use std::fs::File;
12use std::io::{self, BufRead, BufReader, BufWriter, Read, Write};
13use std::path::{Path, PathBuf};
14
15use tempfile::TempDir;
16
17use crate::mini_parser::hash_token;
18use crate::q42_volume::{Q42Volume, QUIN_SIZE, SUPERBLOCK_HEADER, SUPERBLOCK_SIZE};
19
20const RECORD_BYTES: u64 = 32;
21const READ_BUFFER_BYTES: usize = 32 * 1024;
22const MERGE_FAN_IN: usize = 16;
23
24/// Default RAM reserved for the sort buffer.  The verifier uses bounded I/O
25/// buffers in addition to this allocation.
26pub const DEFAULT_GRAPH_PROOF_MEMORY_BYTES: usize = 32 * 1024 * 1024;
27/// Default maximum temporary on-disk footprint.  The verifier fails closed
28/// rather than exhausting an arbitrary temp volume.
29pub const DEFAULT_GRAPH_PROOF_TEMP_BYTES: u64 = 24 * 1024 * 1024 * 1024;
30
31/// Resource limits for [`prove_cli_ntriples_q42_equivalence`].
32#[derive(Clone, Copy, Debug)]
33pub struct GraphProofOptions {
34    /// Upper bound for the in-memory record sort buffer, in bytes.
35    pub memory_limit_bytes: usize,
36    /// Upper bound for live temporary run files, in bytes.
37    pub temporary_byte_budget: u64,
38}
39
40impl Default for GraphProofOptions {
41    fn default() -> Self {
42        Self {
43            memory_limit_bytes: DEFAULT_GRAPH_PROOF_MEMORY_BYTES,
44            temporary_byte_budget: DEFAULT_GRAPH_PROOF_TEMP_BYTES,
45        }
46    }
47}
48
49/// The level of RDF claim that can be made from an encoded Q42 comparison.
50#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Serialize)]
51pub enum RdfIsomorphismStatus {
52    /// The source contains no blank-node labels, so exact encoded-set equality
53    /// is a ground-graph equivalence proof in the ingestion representation.
54    GroundGraphProven,
55    /// The sets match only when original blank-node labels are treated as
56    /// identities.  Canonical blank-node isomorphism needs lexical Q42 terms.
57    BlankNodeCanonicalizationRequired,
58    /// The encoded graph sets differ, so no isomorphism claim is possible.
59    Different,
60}
61
62/// Result of an exact, external-sort comparison.
63#[derive(Clone, Debug, serde::Serialize)]
64pub struct GraphProofReport {
65    /// Number of N-Triples accepted by the CLI-compatible normaliser.
66    pub source_records: u64,
67    /// Number of Q42 Quins examined.
68    pub q42_records: u64,
69    /// Number of unique encoded quads in the source.
70    pub source_unique_records: u64,
71    /// Number of unique encoded quads in the Q42 volume.
72    pub q42_unique_records: u64,
73    /// Source records that do not occur in the Q42 set.
74    pub missing_from_q42: u64,
75    /// Q42 records that do not occur in the source set.
76    pub unexpected_in_q42: u64,
77    /// A representative missing encoded quad, if any.
78    pub first_missing: Option<[u64; 4]>,
79    /// A representative unexpected encoded quad, if any.
80    pub first_unexpected: Option<[u64; 4]>,
81    /// Lines skipped because they are blank, comments, or not accepted by the
82    /// same three-token N-Triples compatibility parser used by CLI ingest.
83    pub source_skipped_lines: u64,
84    pub source_contains_blank_nodes: bool,
85    pub rdf_isomorphism: RdfIsomorphismStatus,
86}
87
88impl GraphProofReport {
89    /// Exact equality of the two encoded RDF default-graph sets.
90    pub fn encoded_sets_match(&self) -> bool {
91        self.missing_from_q42 == 0 && self.unexpected_in_q42 == 0
92    }
93}
94
95/// Compare an N-Triples input to a Q42 volume without retaining either graph
96/// in memory.
97///
98/// Source normalisation intentionally mirrors `qualia-cli ingest semantic`:
99/// it takes the first three ASCII-whitespace-delimited tokens and applies
100/// [`hash_token`] to each.  This proves the bytes that that ingest mode can
101/// encode, rather than pretending the current hash-only volume can recover
102/// lexical RDF values that it never stored.
103pub fn prove_cli_ntriples_q42_equivalence(
104    source_path: &Path,
105    q42_path: &Path,
106    options: GraphProofOptions,
107) -> io::Result<GraphProofReport> {
108    let records_per_chunk = records_per_chunk(options.memory_limit_bytes)?;
109    let workspace = TempDir::new()?;
110    let mut budget = TempBudget::new(options.temporary_byte_budget);
111
112    let mut source_spool = DiskSpool::new(workspace.path(), "source", records_per_chunk);
113    let (source_skipped_lines, source_contains_blank_nodes) =
114        stream_source_records(source_path, &mut source_spool, &mut budget)?;
115    let source_records = source_spool.record_count;
116    let source_runs = source_spool.finish(&mut budget)?;
117    let source_run = merge_to_one(source_runs, workspace.path(), "source", &mut budget)?;
118
119    let mut q42_spool = DiskSpool::new(workspace.path(), "q42", records_per_chunk);
120    stream_q42_records(q42_path, &mut q42_spool, &mut budget)?;
121    let q42_records = q42_spool.record_count;
122    let q42_runs = q42_spool.finish(&mut budget)?;
123    let q42_run = merge_to_one(q42_runs, workspace.path(), "q42", &mut budget)?;
124
125    let comparison = compare_unique_sets(&source_run.path, &q42_run.path)?;
126    let encoded_sets_match = comparison.missing == 0 && comparison.unexpected == 0;
127    let rdf_isomorphism = if !encoded_sets_match {
128        RdfIsomorphismStatus::Different
129    } else if source_contains_blank_nodes {
130        RdfIsomorphismStatus::BlankNodeCanonicalizationRequired
131    } else {
132        RdfIsomorphismStatus::GroundGraphProven
133    };
134
135    Ok(GraphProofReport {
136        source_records,
137        q42_records,
138        source_unique_records: comparison.left_unique,
139        q42_unique_records: comparison.right_unique,
140        missing_from_q42: comparison.missing,
141        unexpected_in_q42: comparison.unexpected,
142        first_missing: comparison.first_missing.map(QuadRecord::as_array),
143        first_unexpected: comparison.first_unexpected.map(QuadRecord::as_array),
144        source_skipped_lines,
145        source_contains_blank_nodes,
146        rdf_isomorphism,
147    })
148}
149
150#[repr(C)]
151#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, bytemuck::Pod, bytemuck::Zeroable)]
152struct QuadRecord {
153    subject: u64,
154    predicate: u64,
155    object: u64,
156    context: u64,
157}
158
159impl QuadRecord {
160    fn as_array(self) -> [u64; 4] {
161        [self.subject, self.predicate, self.object, self.context]
162    }
163}
164
165impl Ord for QuadRecord {
166    fn cmp(&self, other: &Self) -> Ordering {
167        (self.object, self.subject, self.predicate, self.context).cmp(&(
168            other.object,
169            other.subject,
170            other.predicate,
171            other.context,
172        ))
173    }
174}
175
176impl PartialOrd for QuadRecord {
177    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
178        Some(self.cmp(other))
179    }
180}
181
182struct DiskRun {
183    path: PathBuf,
184    bytes: u64,
185}
186
187struct TempBudget {
188    maximum: u64,
189    current: u64,
190}
191
192impl TempBudget {
193    fn new(maximum: u64) -> Self {
194        Self {
195            maximum,
196            current: 0,
197        }
198    }
199
200    fn reserve(&mut self, bytes: u64) -> io::Result<()> {
201        let next = self.current.checked_add(bytes).ok_or_else(|| {
202            io::Error::new(
203                io::ErrorKind::Other,
204                "graph-proof temporary budget overflow",
205            )
206        })?;
207        if next > self.maximum {
208            return Err(io::Error::new(
209                io::ErrorKind::Other,
210                format!(
211                    "graph-proof temporary budget exceeded: need {next} bytes, limit is {} bytes",
212                    self.maximum
213                ),
214            ));
215        }
216        self.current = next;
217        Ok(())
218    }
219
220    fn release(&mut self, bytes: u64) {
221        self.current = self.current.saturating_sub(bytes);
222    }
223}
224
225struct DiskSpool {
226    temp_dir: PathBuf,
227    prefix: &'static str,
228    records: Vec<QuadRecord>,
229    runs: Vec<DiskRun>,
230    run_index: usize,
231    record_count: u64,
232}
233
234impl DiskSpool {
235    fn new(temp_dir: &Path, prefix: &'static str, records_per_chunk: usize) -> Self {
236        Self {
237            temp_dir: temp_dir.to_path_buf(),
238            prefix,
239            records: Vec::with_capacity(records_per_chunk),
240            runs: Vec::new(),
241            run_index: 0,
242            record_count: 0,
243        }
244    }
245
246    fn push(&mut self, record: QuadRecord, budget: &mut TempBudget) -> io::Result<()> {
247        self.records.push(record);
248        self.record_count += 1;
249        if self.records.len() == self.records.capacity() {
250            self.flush(budget)?;
251        }
252        Ok(())
253    }
254
255    fn flush(&mut self, budget: &mut TempBudget) -> io::Result<()> {
256        if self.records.is_empty() {
257            return Ok(());
258        }
259        self.records.sort_unstable();
260        let bytes = self.records.len() as u64 * RECORD_BYTES;
261        budget.reserve(bytes)?;
262        let path = self
263            .temp_dir
264            .join(format!("{}-run-{:08}.bin", self.prefix, self.run_index));
265        self.run_index += 1;
266        let write_result = write_records(&path, &self.records);
267        if let Err(error) = write_result {
268            budget.release(bytes);
269            return Err(error);
270        }
271        self.records.clear();
272        self.runs.push(DiskRun { path, bytes });
273        Ok(())
274    }
275
276    fn finish(&mut self, budget: &mut TempBudget) -> io::Result<Vec<DiskRun>> {
277        self.flush(budget)?;
278        Ok(std::mem::take(&mut self.runs))
279    }
280}
281
282fn records_per_chunk(memory_limit_bytes: usize) -> io::Result<usize> {
283    let records = memory_limit_bytes / RECORD_BYTES as usize;
284    if records == 0 {
285        return Err(io::Error::new(
286            io::ErrorKind::InvalidInput,
287            "graph-proof memory limit must hold at least one 32-byte record",
288        ));
289    }
290    Ok(records)
291}
292
293fn stream_source_records(
294    source_path: &Path,
295    spool: &mut DiskSpool,
296    budget: &mut TempBudget,
297) -> io::Result<(u64, bool)> {
298    let source = File::open(source_path)?;
299    let mut reader = BufReader::with_capacity(READ_BUFFER_BYTES, source);
300    let mut buffer = Vec::with_capacity(READ_BUFFER_BYTES);
301    let mut skipped = 0u64;
302    let mut has_blank_nodes = false;
303
304    loop {
305        buffer.clear();
306        if reader.read_until(b'\n', &mut buffer)? == 0 {
307            break;
308        }
309        let line = std::str::from_utf8(&buffer).map_err(|_| {
310            io::Error::new(io::ErrorKind::InvalidData, "N-Triples source is not UTF-8")
311        })?;
312        let line = line.trim();
313        if line.is_empty() || line.starts_with('#') {
314            skipped += 1;
315            continue;
316        }
317        let mut tokens = line.split_ascii_whitespace();
318        let (Some(subject), Some(predicate), Some(object)) =
319            (tokens.next(), tokens.next(), tokens.next())
320        else {
321            return Err(io::Error::new(
322                io::ErrorKind::InvalidData,
323                "source contains a non-comment line without an RDF triple",
324            ));
325        };
326        has_blank_nodes |=
327            subject.starts_with("_:") || predicate.starts_with("_:") || object.starts_with("_:");
328        spool.push(
329            QuadRecord {
330                subject: hash_token(subject),
331                predicate: hash_token(predicate),
332                object: hash_token(object),
333                context: 0,
334            },
335            budget,
336        )?;
337    }
338    Ok((skipped, has_blank_nodes))
339}
340
341fn stream_q42_records(
342    q42_path: &Path,
343    spool: &mut DiskSpool,
344    budget: &mut TempBudget,
345) -> io::Result<()> {
346    let volume = Q42Volume::open(q42_path)?;
347    if volume.volume_manifest()?.is_some() {
348        let set = crate::q42_volume::Q42VolumeSet::open_root(q42_path)?;
349        for segment in set.segments() {
350            stream_q42_volume_records(segment, spool, budget)?;
351        }
352        return Ok(());
353    }
354    stream_q42_volume_records(&volume, spool, budget)
355}
356
357fn stream_q42_volume_records(
358    volume: &Q42Volume,
359    spool: &mut DiskSpool,
360    budget: &mut TempBudget,
361) -> io::Result<()> {
362    let mut buffer = [0u8; SUPERBLOCK_SIZE];
363    for block_index in 0..volume.block_count() as usize {
364        volume.read_superblock_into(block_index, &mut buffer)?;
365        let quin_count = u64::from_le_bytes(buffer[16..24].try_into().unwrap()) as usize;
366        if quin_count > crate::QUINS_PER_BLOCK {
367            return Err(io::Error::new(
368                io::ErrorKind::InvalidData,
369                "Q42 superblock declares too many Quins",
370            ));
371        }
372        let mut offset = SUPERBLOCK_HEADER;
373        for _ in 0..quin_count {
374            let quin: crate::NQuin =
375                bytemuck::pod_read_unaligned(&buffer[offset..offset + QUIN_SIZE]);
376            if !quin.verify_ecc_parity() {
377                return Err(io::Error::new(
378                    io::ErrorKind::InvalidData,
379                    format!("Q42 parity mismatch in block {block_index}"),
380                ));
381            }
382            spool.push(
383                QuadRecord {
384                    subject: quin.subject,
385                    predicate: quin.predicate,
386                    object: quin.object,
387                    context: quin.context,
388                },
389                budget,
390            )?;
391            offset += QUIN_SIZE;
392        }
393    }
394    Ok(())
395}
396
397fn write_records(path: &Path, records: &[QuadRecord]) -> io::Result<()> {
398    let mut writer = BufWriter::with_capacity(READ_BUFFER_BYTES, File::create(path)?);
399    for record in records {
400        writer.write_all(bytemuck::bytes_of(record))?;
401    }
402    writer.flush()
403}
404
405fn merge_to_one(
406    mut runs: Vec<DiskRun>,
407    temp_dir: &Path,
408    prefix: &str,
409    budget: &mut TempBudget,
410) -> io::Result<DiskRun> {
411    if runs.is_empty() {
412        let path = temp_dir.join(format!("{prefix}-empty.bin"));
413        File::create(&path)?;
414        return Ok(DiskRun { path, bytes: 0 });
415    }
416    let mut round = 0usize;
417    while runs.len() > 1 {
418        let mut next = Vec::with_capacity(runs.len().div_ceil(MERGE_FAN_IN));
419        let mut group_index = 0usize;
420        while !runs.is_empty() {
421            let group_len = runs.len().min(MERGE_FAN_IN);
422            let group: Vec<DiskRun> = runs.drain(..group_len).collect();
423            let bytes: u64 = group.iter().map(|run| run.bytes).sum();
424            budget.reserve(bytes)?;
425            let output = temp_dir.join(format!("{prefix}-merge-{round:04}-{group_index:08}.bin"));
426            group_index += 1;
427            if let Err(error) = merge_group(&group, &output) {
428                budget.release(bytes);
429                return Err(error);
430            }
431            for run in group {
432                std::fs::remove_file(&run.path)?;
433                budget.release(run.bytes);
434            }
435            next.push(DiskRun {
436                path: output,
437                bytes,
438            });
439        }
440        runs = next;
441        round += 1;
442    }
443    Ok(runs.pop().expect("non-empty run list"))
444}
445
446fn merge_group(group: &[DiskRun], output: &Path) -> io::Result<()> {
447    let mut readers = Vec::with_capacity(group.len());
448    for run in group {
449        readers.push(RecordReader::open(&run.path)?);
450    }
451    let mut writer = BufWriter::with_capacity(READ_BUFFER_BYTES, File::create(output)?);
452    loop {
453        let mut minimum: Option<(usize, QuadRecord)> = None;
454        for (index, reader) in readers.iter().enumerate() {
455            if let Some(record) = reader.peek {
456                if minimum.is_none_or(|(_, current)| record < current) {
457                    minimum = Some((index, record));
458                }
459            }
460        }
461        let Some((index, record)) = minimum else {
462            break;
463        };
464        writer.write_all(bytemuck::bytes_of(&record))?;
465        readers[index].advance()?;
466    }
467    writer.flush()
468}
469
470struct RecordReader {
471    reader: BufReader<File>,
472    peek: Option<QuadRecord>,
473}
474
475impl RecordReader {
476    fn open(path: &Path) -> io::Result<Self> {
477        let mut reader = Self {
478            reader: BufReader::with_capacity(READ_BUFFER_BYTES, File::open(path)?),
479            peek: None,
480        };
481        reader.advance()?;
482        Ok(reader)
483    }
484
485    fn advance(&mut self) -> io::Result<()> {
486        let mut bytes = [0u8; RECORD_BYTES as usize];
487        match self.reader.read_exact(&mut bytes) {
488            Ok(()) => {
489                self.peek = Some(bytemuck::pod_read_unaligned(&bytes));
490                Ok(())
491            }
492            Err(error) if error.kind() == io::ErrorKind::UnexpectedEof => {
493                self.peek = None;
494                Ok(())
495            }
496            Err(error) => Err(error),
497        }
498    }
499
500    fn next_distinct(&mut self) -> io::Result<Option<QuadRecord>> {
501        let Some(record) = self.peek else {
502            return Ok(None);
503        };
504        while self.peek == Some(record) {
505            self.advance()?;
506        }
507        Ok(Some(record))
508    }
509}
510
511#[derive(Default)]
512struct SetComparison {
513    left_unique: u64,
514    right_unique: u64,
515    missing: u64,
516    unexpected: u64,
517    first_missing: Option<QuadRecord>,
518    first_unexpected: Option<QuadRecord>,
519}
520
521fn compare_unique_sets(left: &Path, right: &Path) -> io::Result<SetComparison> {
522    let mut left_reader = RecordReader::open(left)?;
523    let mut right_reader = RecordReader::open(right)?;
524    let mut left_record = left_reader.next_distinct()?;
525    let mut right_record = right_reader.next_distinct()?;
526    let mut result = SetComparison::default();
527
528    loop {
529        match (left_record, right_record) {
530            (Some(left), Some(right)) => match left.cmp(&right) {
531                Ordering::Equal => {
532                    result.left_unique += 1;
533                    result.right_unique += 1;
534                    left_record = left_reader.next_distinct()?;
535                    right_record = right_reader.next_distinct()?;
536                }
537                Ordering::Less => {
538                    result.left_unique += 1;
539                    result.missing += 1;
540                    result.first_missing.get_or_insert(left);
541                    left_record = left_reader.next_distinct()?;
542                }
543                Ordering::Greater => {
544                    result.right_unique += 1;
545                    result.unexpected += 1;
546                    result.first_unexpected.get_or_insert(right);
547                    right_record = right_reader.next_distinct()?;
548                }
549            },
550            (Some(left), None) => {
551                result.left_unique += 1;
552                result.missing += 1;
553                result.first_missing.get_or_insert(left);
554                left_record = left_reader.next_distinct()?;
555            }
556            (None, Some(right)) => {
557                result.right_unique += 1;
558                result.unexpected += 1;
559                result.first_unexpected.get_or_insert(right);
560                right_record = right_reader.next_distinct()?;
561            }
562            (None, None) => return Ok(result),
563        }
564    }
565}
qualia_core_db/query/graph_proof.rs

qualia_core_db/query/
graph_proof.rs
