qualia_core_db/query/
ingest.rs

1//! Streaming import path for RDF text sources → canonical `.q42` volume.
2//!
3//! Pipeline: a Rio parser on the main thread streams triples into a bounded channel; a pool of worker
4//! shards hashes each triple into an `NQuin` and (in [`IngestMode::Complete`]) interns every
5//! subject/predicate/object string into a per-shard lexicon; a collector gathers the quins; the main
6//! thread merges the lexicon, sorts the quins by object hash, and writes the volume via
7//! [`crate::q42_volume::UnifiedVolumeBuilder`] — the GOVERNING `.q42` layout (160-byte SuperBlock
8//! headers, block directory, BIDX object index, Merkle-DAG, and a real lexicon section).
9//!
10//! History: this path previously wrote headerless LZ4 blocks and an empty lexicon, so
11//! `Q42Volume::read_all_quins` could not read the graph back and all literal text was discarded while
12//! the shrunk file was reported as "compression". Both defects are fixed — see [`IngestMode`].
13
14use crate::{q_hash, NQuin};
15use log;
16
17const OBJECT_HASH_MASK: u64 = 0x0FFF_FFFF_FFFF_FFFF;
18use crossbeam_channel::{bounded, Receiver, Sender};
19use rio_api::parser::TriplesParser;
20use rio_turtle::{NTriplesParser, TurtleParser};
21use rio_xml::RdfXmlParser;
22use std::collections::HashMap;
23use std::fs::File;
24use std::io::{self, BufReader, Cursor};
25use std::path::Path;
26use std::thread;
27use std::time::Instant;
28use sysinfo::System;
29use tempfile::TempDir;
30
31/// Base IRI for relative terms / empty `xml:base` in catalog RDF.
32pub fn catalog_base_iri(path: &Path) -> Option<oxiri::Iri<String>> {
33    let stem = path.file_stem()?.to_str()?.to_ascii_lowercase();
34    let iri = match stem.as_str() {
35        "earl" => "http://www.w3.org/ns/earl#".to_string(),
36        "music" => "http://purl.org/ontology/mo/".to_string(),
37        other => format!("https://www.w3.org/ns/{other}#"),
38    };
39    oxiri::Iri::parse(iri).ok()
40}
41
42/// Expand Turtle `prefix:` empty local names (`dc:`, `foaf:`, `ns1:`) to IRIs.
43///
44/// Rio rejects those tokens even though Turtle 1.1 allows them. Catalog
45/// ontologies (Music Ontology in particular) use them as namespace objects.
46pub fn expand_empty_turtle_prefixed_names(src: &str) -> String {
47    let mut prefixes: Vec<(String, String)> = Vec::new();
48    for line in src.lines() {
49        let trimmed = line.trim();
50        let rest = trimmed
51            .strip_prefix("@prefix ")
52            .or_else(|| trimmed.strip_prefix("PREFIX "));
53        let Some(rest) = rest else {
54            continue;
55        };
56        let rest = rest.trim();
57        let Some(colon) = rest.find(':') else {
58            continue;
59        };
60        let name = rest[..colon].trim();
61        let Some(start) = rest[colon + 1..].find('<') else {
62            continue;
63        };
64        let after = &rest[colon + 1 + start + 1..];
65        let Some(end) = after.find('>') else {
66            continue;
67        };
68        prefixes.push((name.to_string(), after[..end].to_string()));
69    }
70    prefixes.sort_by(|a, b| b.0.len().cmp(&a.0.len()));
71    if prefixes.is_empty() {
72        return src.to_string();
73    }
74
75    let bytes = src.as_bytes();
76    let mut out = String::with_capacity(src.len() + 256);
77    let mut i = 0;
78    let mut in_iri = false;
79    let mut in_string = false;
80    let mut long_string = false;
81    let mut in_prefix_decl = false;
82    while i < bytes.len() {
83        let c = bytes[i] as char;
84        if !in_iri && !in_string && starts_with_keyword(bytes, i, b"@prefix") {
85            in_prefix_decl = true;
86        } else if !in_iri && !in_string && starts_with_keyword(bytes, i, b"PREFIX") {
87            in_prefix_decl = true;
88        }
89        if in_prefix_decl {
90            out.push(c);
91            if c == '>' {
92                in_prefix_decl = false;
93            }
94            i += 1;
95            continue;
96        }
97        if in_iri {
98            out.push(c);
99            if c == '>' {
100                in_iri = false;
101            }
102            i += 1;
103            continue;
104        }
105        if in_string {
106            out.push(c);
107            if c == '\\' && i + 1 < bytes.len() {
108                out.push(bytes[i + 1] as char);
109                i += 2;
110                continue;
111            }
112            if long_string
113                && c == '"'
114                && i + 2 < bytes.len()
115                && bytes[i + 1] == b'"'
116                && bytes[i + 2] == b'"'
117            {
118                out.push('"');
119                out.push('"');
120                i += 3;
121                in_string = false;
122                long_string = false;
123                continue;
124            }
125            if !long_string && c == '"' {
126                in_string = false;
127            }
128            i += 1;
129            continue;
130        }
131        if c == '<' {
132            in_iri = true;
133            out.push(c);
134            i += 1;
135            continue;
136        }
137        if c == '"' {
138            in_string = true;
139            long_string = i + 2 < bytes.len() && bytes[i + 1] == b'"' && bytes[i + 2] == b'"';
140            out.push(c);
141            i += 1;
142            continue;
143        }
144        if let Some((name, iri)) = prefixes.iter().find(|(name, _)| {
145            let start = i;
146            let end = start + name.len();
147            bytes.get(end) == Some(&b':')
148                && bytes.get(start..end) == Some(name.as_bytes())
149                && (start == 0 || !is_prefix_name_char(bytes[start - 1]))
150        }) {
151            let local_start = i + name.len() + 1;
152            if empty_local_name_follows(bytes, local_start) {
153                out.push('<');
154                out.push_str(iri);
155                out.push('>');
156                i = local_start;
157                continue;
158            }
159            if let Some(hash_at) = hash_terminated_local_name(bytes, local_start) {
160                out.push('<');
161                out.push_str(iri);
162                out.push_str(&src[local_start..=hash_at]);
163                out.push('>');
164                i = hash_at + 1;
165                continue;
166            }
167        }
168        out.push(c);
169        i += 1;
170    }
171    out
172}
173
174fn starts_with_keyword(bytes: &[u8], i: usize, keyword: &[u8]) -> bool {
175    bytes.get(i..i + keyword.len()) == Some(keyword)
176        && (i == 0 || !is_prefix_name_char(bytes[i - 1]))
177        && match bytes.get(i + keyword.len()) {
178            Some(b) if is_prefix_name_char(*b) => false,
179            _ => true,
180        }
181}
182
183fn is_prefix_name_char(b: u8) -> bool {
184    b.is_ascii_alphanumeric() || b == b'_' || b == b'-'
185}
186
187fn repair_rdfxml_empty_base(src: &str, base: &str) -> String {
188    src.replace("xml:base=\"\"", &format!("xml:base=\"{base}\""))
189        .replace("xml:base=''", &format!("xml:base='{base}'"))
190}
191
192fn empty_local_name_follows(bytes: &[u8], i: usize) -> bool {
193    match bytes.get(i) {
194        None => true,
195        Some(b) => matches!(*b, b' ' | b'\t' | b'\r' | b'\n' | b',' | b';' | b'.' | b')' | b']'),
196    }
197}
198
199fn hash_terminated_local_name(bytes: &[u8], start: usize) -> Option<usize> {
200    let mut i = start;
201    if i >= bytes.len() || !is_prefix_name_char(bytes[i]) {
202        return None;
203    }
204    i += 1;
205    while i < bytes.len() && is_prefix_name_char(bytes[i]) {
206        i += 1;
207    }
208    if bytes.get(i) == Some(&b'#') {
209        Some(i)
210    } else {
211        None
212    }
213}
214
215fn ingest_scratch_dir() -> io::Result<TempDir> {
216    if let Some(parent) = std::env::var_os("QUALIA_INGEST_SCRATCH") {
217        std::fs::create_dir_all(&parent)?;
218        TempDir::new_in(parent)
219    } else {
220        TempDir::new()
221    }
222}
223
224/// How much of the source graph the `.q42` retains.
225///
226/// The historical ingest hashed every subject/predicate/object into a 48-byte quin and wrote an
227/// **empty** lexicon (`lex_length: 0`). That threw away every URI and every literal — the source text
228/// was irrecoverable — while the shrunk output was presented as "compression". It was not compression;
229/// it was data loss reported as a size win. That is exactly the kind of claim-vs-reality gap this
230/// project's integrity rules (CLAUDE.md §15) forbid. This enum makes the choice explicit and the
231/// reporting honest.
232#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
233pub enum IngestMode {
234    /// **Lossless.** Intern every subject/predicate URI and every literal (full UTF-8 / Unicode) into
235    /// the q42 lexicon, so `Q42LexMmap::lookup_hash(quin.field)` recovers the original term. The `.q42`
236    /// is a faithful, reversible representation of the source graph. This is the default — honesty over
237    /// a smaller file.
238    #[default]
239    Complete,
240    /// **Lossy, structure-only.** Store only the 48-byte hash quins; discard all human-readable text
241    /// (URIs and literals). Smaller on disk, but the original strings CANNOT be recovered. The size
242    /// reduction is data loss, not compression, and is reported as such. Use only when the graph
243    /// structure alone is wanted and the term strings are available elsewhere.
244    StripLiterals,
245}
246
247impl IngestMode {
248    fn label(self) -> &'static str {
249        match self {
250            IngestMode::Complete => "COMPLETE (lossless — all URIs & literals retained)",
251            IngestMode::StripLiterals => {
252                "STRIP-LITERALS (lossy — human-readable text discarded, structure only)"
253            }
254        }
255    }
256}
257
258/// Represents a raw string-based Triple extracted from RDF/XML
259#[derive(Debug)]
260pub struct RawTriple {
261    pub subject: String,
262    pub predicate: String,
263    pub object: String,
264    pub packed_object: Option<u64>,
265}
266
267/// Back-compat entry point: ingests losslessly ([`IngestMode::Complete`]) — the honest default.
268pub fn streaming_import_rdf(in_path: &str, out_path: &str) -> std::io::Result<u64> {
269    streaming_import_rdf_with_mode(in_path, out_path, IngestMode::Complete)
270}
271
272/// Stream-ingest an RDF source into a `.q42` volume under an explicit [`IngestMode`].
273pub fn streaming_import_rdf_with_mode(
274    in_path: &str,
275    out_path: &str,
276    mode: IngestMode,
277) -> std::io::Result<u64> {
278    streaming_import_rdf_with_mode_inner(in_path, out_path, mode, None)
279}
280
281/// Stream-ingest RDF into a front-embedded logical volume root and immutable,
282/// size-capped child Q42 segments.
283pub fn streaming_import_rdf_volume_set_with_mode(
284    in_path: &str,
285    root_path: &str,
286    mode: IngestMode,
287    max_segment_bytes: u64,
288) -> std::io::Result<u64> {
289    streaming_import_rdf_with_mode_inner(in_path, root_path, mode, Some(max_segment_bytes))
290}
291
292fn streaming_import_rdf_with_mode_inner(
293    in_path: &str,
294    out_path: &str,
295    mode: IngestMode,
296    max_segment_bytes: Option<u64>,
297) -> std::io::Result<u64> {
298    let start_time = Instant::now();
299    println!("Initializing Native Ingestion Pipeline...");
300
301    // 1. Hardware Detection & Scaling
302    let mut sys = System::new_all();
303    sys.refresh_all();
304    let logical_cores = sys.cpus().len();
305
306    // Constraint: Use no more than 80% of available CPU resources
307    let target_workers = std::cmp::max(1, (logical_cores as f32 * 0.8).floor() as usize);
308    println!("Hardware Sieve: Detected {} logical cores. Spinning up {} parallel hasher shards (capped at 80%).", logical_cores, target_workers);
309
310    // 2. Channel Setup
311    // Use bounded channels to strictly enforce the 512MB RAM floor (backpressure)
312    let (tx_raw, rx_raw): (Sender<RawTriple>, Receiver<RawTriple>) = bounded(10_000);
313    let (tx_bin, rx_bin): (Sender<NQuin>, Receiver<NQuin>) = bounded(10_000);
314
315    // 3. Spawn Parallel Hasher Shards (Workers)
316    // Each worker returns its local hash→string lexicon shard (empty in StripLiterals mode); the main
317    // thread merges the shards into one lexicon and writes it into the volume so the terms are
318    // recoverable. This is the fix for the historical data loss: previously the strings were hashed and
319    // thrown away here, and no lexicon was ever written.
320    let collect_lexicon = mode == IngestMode::Complete;
321    let mut worker_handles: Vec<thread::JoinHandle<HashMap<u64, String>>> = vec![];
322    for _worker_id in 0..target_workers {
323        let rx = rx_raw.clone();
324        let tx = tx_bin.clone();
325
326        let handle = thread::spawn(move || {
327            let mut _local_count = 0u64;
328            let mut lex: HashMap<u64, String> = HashMap::new();
329            for triple in rx {
330                let s_hash = q_hash(&triple.subject);
331                let p_hash = q_hash(&triple.predicate);
332                // Objects that pack into the quin inline (typed int/decimal/bool) carry their full value
333                // in the field itself — no string needed. Everything else is stored under the same
334                // masked hash that lands in `quin.object`, so `lookup_hash(quin.object)` resolves it.
335                let is_inline = triple.packed_object.is_some();
336                let o_hash = triple
337                    .packed_object
338                    .unwrap_or_else(|| q_hash(&triple.object) & OBJECT_HASH_MASK);
339                let context = 0u64;
340                let metadata = 0u64;
341                let parity = s_hash ^ p_hash ^ o_hash ^ context ^ metadata;
342
343                let quin = NQuin {
344                    subject: s_hash,
345                    predicate: p_hash,
346                    object: o_hash,
347                    context,
348                    metadata,
349                    parity,
350                };
351
352                if collect_lexicon {
353                    // `or_insert_with(|| moved)` still moves the string even when the entry is
354                    // occupied, so branch explicitly to avoid needless clones/moves on hot hits.
355                    if !lex.contains_key(&s_hash) {
356                        lex.insert(s_hash, triple.subject);
357                    }
358                    if !lex.contains_key(&p_hash) {
359                        lex.insert(p_hash, triple.predicate);
360                    }
361                    if !is_inline && !lex.contains_key(&o_hash) {
362                        lex.insert(o_hash, triple.object);
363                    }
364                }
365
366                // Send back to the writer thread
367                if tx.send(quin).is_err() {
368                    break;
369                }
370                _local_count += 1;
371            }
372            if _local_count > 0 {
373                log::debug!(
374                    "Ontology Ingest: worker shard finished {} triples ({} lexemes)",
375                    _local_count,
376                    lex.len()
377                );
378            }
379            lex
380        });
381        worker_handles.push(handle);
382    }
383
384    // Drop the extra transmitters so channels close correctly
385    drop(tx_bin);
386
387    // 4. Drain hashed Quins into bounded external-sort runs instead of one
388    // whole-graph Vec. The TempDir owns every run and cleans it on success,
389    // error, or unwind.
390    let sorter_temp = ingest_scratch_dir()?;
391    let sorter_path = sorter_temp.path().to_owned();
392    let collector_handle = thread::spawn(
393        move || -> std::io::Result<crate::external_sort::ExternalSorter> {
394            let mut sorter = crate::external_sort::ExternalSorter::new(sorter_path);
395            for quin in rx_bin {
396                sorter.push(quin)?;
397            }
398            Ok(sorter)
399        },
400    );
401
402    // 5. The Streaming Sieve (Main Thread)
403    // Uses Rio to read the file sequentially without loading the whole graph into RAM.
404    let in_file = File::open(&in_path)?;
405    let buf_reader = BufReader::new(in_file);
406
407    let mut triples_read = 0;
408
409    // Setup a callback that parses Rio triples and sends them to the worker queue
410    log::info!("Ontology Ingest: streaming triples from {}", in_path);
411    let mut on_triple = |t: rio_api::model::Triple| -> Result<(), std::io::Error> {
412        let subject = t.subject.to_string();
413        let predicate = t.predicate.to_string();
414        let object = t.object.to_string();
415        let mut packed_object = None;
416
417        if let rio_api::model::Term::Literal(rio_api::model::Literal::Typed { value, datatype }) =
418            t.object
419        {
420            let dt = datatype.iri;
421            if dt == "http://www.w3.org/2001/XMLSchema#integer" {
422                if let Ok(num) = value.parse::<i64>() {
423                    let max_val = (1i64 << 59) - 1;
424                    let min_val = -(1i64 << 59);
425                    if num >= min_val && num <= max_val {
426                        let unsigned = (num as u64) & crate::resolver::INLINE_VALUE_MASK;
427                        packed_object = Some(crate::resolver::INLINE_TAG_INTEGER | unsigned);
428                    }
429                }
430            } else if dt == "http://www.w3.org/2001/XMLSchema#decimal" {
431                if let Ok(num) = value.parse::<f64>() {
432                    let scaled = num * 1_000_000.0;
433                    let max_val = ((1i64 << 59) - 1) as f64;
434                    let min_val = (-(1i64 << 59)) as f64;
435                    if scaled >= min_val && scaled <= max_val {
436                        let num_i64 = scaled.round() as i64;
437                        let unsigned = (num_i64 as u64) & crate::resolver::INLINE_VALUE_MASK;
438                        packed_object = Some(crate::resolver::INLINE_TAG_DECIMAL | unsigned);
439                    }
440                }
441            } else if dt == "http://www.w3.org/2001/XMLSchema#boolean" {
442                if value == "true" || value == "1" {
443                    packed_object = Some(crate::resolver::INLINE_TAG_BOOLEAN | 1);
444                } else if value == "false" || value == "0" {
445                    packed_object = Some(crate::resolver::INLINE_TAG_BOOLEAN | 0);
446                }
447            }
448        }
449
450        let raw = RawTriple {
451            subject,
452            predicate,
453            object,
454            packed_object,
455        };
456        if tx_raw.send(raw).is_ok() {
457            triples_read += 1;
458        }
459        Ok(())
460    };
461
462    let path_lower = in_path.to_lowercase();
463    let base_iri = catalog_base_iri(Path::new(in_path));
464    let mut parse_error: Option<String> = None;
465    if path_lower.ends_with(".rdf") || path_lower.ends_with(".xml") || path_lower.ends_with(".owl")
466    {
467        log::info!("Ontology Ingest: parsing RDF/XML source {}", in_path);
468        let raw = std::fs::read_to_string(in_path)?;
469        let repaired = if let Some(base) = base_iri.as_ref() {
470            repair_rdfxml_empty_base(&raw, base.as_str())
471        } else {
472            raw
473        };
474        let mut parser = RdfXmlParser::new(Cursor::new(repaired), base_iri.clone());
475        if let Err(e) = parser.parse_all(&mut on_triple) {
476            parse_error = Some(format!("RDF/XML: {e}"));
477        }
478        log::info!("Ontology Ingest: completed RDF/XML parse for {}", in_path);
479    } else if path_lower.ends_with(".ttl") {
480        log::info!("Ontology Ingest: parsing Turtle source {}", in_path);
481        let raw = std::fs::read_to_string(in_path)?;
482        let expanded = expand_empty_turtle_prefixed_names(&raw);
483        let mut parser = TurtleParser::new(Cursor::new(expanded), base_iri.clone());
484        if let Err(e) = parser.parse_all(&mut on_triple) {
485            parse_error = Some(format!("Turtle: {e}"));
486        }
487        log::info!("Ontology Ingest: completed Turtle parse for {}", in_path);
488    } else if path_lower.ends_with(".nt") {
489        log::info!("Ontology Ingest: parsing N-Triples source {}", in_path);
490        let mut parser = NTriplesParser::new(buf_reader);
491        if let Err(e) = parser.parse_all(&mut on_triple) {
492            parse_error = Some(format!("N-Triples: {e}"));
493        }
494        log::info!("Ontology Ingest: completed N-Triples parse for {}", in_path);
495    } else if path_lower.ends_with(".n3") {
496        log::info!("Ontology Ingest: parsing N3 source {}", in_path);
497        let text = std::fs::read_to_string(in_path).unwrap_or_default();
498        let mut parser = crate::modalities::logic::n3_parser::N3Parser::new(&text);
499        let mut webizen = crate::webizen::SlgArena::new();
500        let mut rules_parsed = 0;
501
502        let on_n3_event = |event: crate::modalities::logic::n3_parser::N3Event| -> Result<(), crate::modalities::logic::n3_parser::N3ParserError> {
503            match event {
504                crate::modalities::logic::n3_parser::N3Event::StaticTriple(triple) => {
505                    let subject = match triple.subject {
506                        crate::modalities::logic::n3_parser::Term::Uri(s)
507                        | crate::modalities::logic::n3_parser::Term::Variable(s)
508                        | crate::modalities::logic::n3_parser::Term::Literal(s)
509                        | crate::modalities::logic::n3_parser::Term::Formula(s) => s.to_string(),
510                    };
511                    let predicate = match triple.predicate {
512                        crate::modalities::logic::n3_parser::Term::Uri(s)
513                        | crate::modalities::logic::n3_parser::Term::Variable(s)
514                        | crate::modalities::logic::n3_parser::Term::Literal(s)
515                        | crate::modalities::logic::n3_parser::Term::Formula(s) => s.to_string(),
516                    };
517                    let object = match triple.object {
518                        crate::modalities::logic::n3_parser::Term::Uri(s)
519                        | crate::modalities::logic::n3_parser::Term::Variable(s)
520                        | crate::modalities::logic::n3_parser::Term::Literal(s)
521                        | crate::modalities::logic::n3_parser::Term::Formula(s) => s.to_string(),
522                    };
523                    let raw = RawTriple {
524                        subject,
525                        predicate,
526                        object,
527                        packed_object: None,
528                    };
529                    if tx_raw.send(raw).is_ok() {
530                        triples_read += 1;
531                    }
532                }
533                crate::modalities::logic::n3_parser::N3Event::LogicRule(rule) => {
534                    webizen.register_rule(&rule);
535                    rules_parsed += 1;
536                }
537                crate::modalities::logic::n3_parser::N3Event::AspBlock(_)
538                | crate::modalities::logic::n3_parser::N3Event::DiffuseBlock(_) => {
539                    // Pass these modalities to the Webizen
540                }
541            }
542            Ok(())
543        };
544
545        if let Err(e) = parser.parse_all(on_n3_event) {
546            parse_error = Some(format!("N3: {e}"));
547        }
548        let fired = webizen.fire_registered_rules(crate::q_hash("q42:ingestSession"));
549        println!(
550            "Registered {} N3 Logic Rules; fired {} through Core-1 Sentinel VM.",
551            rules_parsed, fired
552        );
553        log::info!(
554            "Ontology Ingest: completed N3 parse for {} (rules parsed: {}, fired: {})",
555            in_path,
556            rules_parsed,
557            fired
558        );
559    } else {
560        drop(tx_raw);
561        return Err(io::Error::new(
562            io::ErrorKind::InvalidInput,
563            format!("unsupported RDF extension for {in_path}; expected .rdf, .xml, .owl, .ttl, .nt, or .n3"),
564        ));
565    }
566
567    if let Some(error) = parse_error {
568        drop(tx_raw);
569        return Err(io::Error::new(
570            io::ErrorKind::InvalidData,
571            format!(
572                "RDF parse failed after {triples_read} triples in {in_path}: {error}"
573            ),
574        ));
575    }
576
577    // Drop the main sender so workers know to terminate
578    drop(tx_raw);
579
580    // 6. Join the workers first (this closes the collector's channel), merging each shard's lexicon
581    // into one hash→string map. First-writer-wins on hash collisions across shards — deterministic
582    // given the same input regardless of shard scheduling, since a given term always hashes to the same
583    // key.
584    let mut lexicon: HashMap<u64, String> = HashMap::new();
585    for handle in worker_handles {
586        let shard = handle.join().unwrap();
587        if lexicon.is_empty() {
588            lexicon = shard;
589        } else {
590            for (k, v) in shard {
591                lexicon.entry(k).or_insert(v);
592            }
593        }
594    }
595
596    let mut sorter = collector_handle
597        .join()
598        .map_err(|_| std::io::Error::other("Q42 external-sort collector thread panicked"))??;
599    if mode == IngestMode::Complete {
600        for (hash, term) in &lexicon {
601            sorter.push_lex(*hash, term);
602        }
603    }
604    let total_written = match max_segment_bytes {
605        Some(cap) => {
606            sorter
607                .merge_volume_set(std::path::Path::new(out_path), cap)?
608                .blocks_written
609        }
610        None => sorter.merge(std::path::Path::new(out_path))?,
611    };
612
613    // Lexicon byte size actually written — read back cheaply from the finished header (mmap, no
614    // re-serialize) so the report reflects what is really on disk.
615    let (lex_length, out_bytes) =
616        crate::q42_volume::Q42Volume::open(std::path::Path::new(out_path))
617            .ok()
618            .map(|root| {
619                let mut lex_bytes = root.header().lex_length;
620                let mut logical_bytes = std::fs::metadata(out_path).map(|m| m.len()).unwrap_or(0);
621                if let Ok(Some(manifest)) = root.volume_manifest() {
622                    let parent = std::path::Path::new(out_path)
623                        .parent()
624                        .unwrap_or_else(|| std::path::Path::new("."));
625                    for segment in manifest.segments {
626                        logical_bytes = logical_bytes.saturating_add(segment.byte_length);
627                    }
628                    for segment in manifest.lexicon_segments {
629                        logical_bytes = logical_bytes.saturating_add(segment.byte_length);
630                    if let Ok(shard) =
631                        crate::q42_volume::Q42Volume::open(&parent.join(&segment.locator))
632                        {
633                            lex_bytes = lex_bytes.saturating_add(shard.header().lex_length);
634                        }
635                    }
636                }
637                (lex_bytes, logical_bytes)
638            })
639            .unwrap_or((0, std::fs::metadata(out_path).map(|m| m.len()).unwrap_or(0)));
640
641    let duration = start_time.elapsed();
642
643    // 8. Honest reporting — state the mode and, for the lossy mode, that the size reduction is data
644    // loss, NOT compression (CLAUDE.md §15: no claim-vs-reality gap).
645    let src_bytes = std::fs::metadata(in_path).map(|m| m.len()).unwrap_or(0);
646    println!("✅ Import Complete!");
647    println!("Parsed {} triples.", triples_read);
648    log::info!("Ontology Ingest: parsed {} triples", triples_read);
649    println!("Wrote {} Super-Quins to {}.", total_written, out_path);
650    println!("Ingest mode: {}", mode.label());
651    match mode {
652        IngestMode::Complete => {
653            println!(
654                "Lexicon: {} unique terms retained ({} bytes) — every URI and literal recoverable (full Unicode).",
655                lexicon.len(),
656                lex_length
657            );
658            println!(
659                "Source {} B → .q42 {} B (lossless: 48-byte structure + complete lexicon; reversible to the source terms).",
660                src_bytes, out_bytes
661            );
662        }
663        IngestMode::StripLiterals => {
664            println!(
665                "⚠  STRIP-LITERALS mode: human-readable text (URIs and literals) was DISCARDED and is NOT recoverable."
666            );
667            println!(
668                "Source {} B → .q42 {} B. This size reduction is DATA LOSS (structure-only), not compression.",
669                src_bytes, out_bytes
670            );
671        }
672    }
673    println!("Total Time: {:?}", duration);
674    let total_superblocks =
675        (total_written + (crate::QUINS_PER_BLOCK as u64) - 1) / crate::QUINS_PER_BLOCK as u64;
676    log::info!(
677        "Ontology Ingest: Completed {} SuperBlocks ({} quins, mode {:?}) in {:?}",
678        total_superblocks,
679        total_written,
680        mode,
681        duration
682    );
683
684    Ok(total_written)
685}
686
687pub fn verify_integrity(
688    input_path: std::path::PathBuf,
689    dataset_path: std::path::PathBuf,
690) -> std::io::Result<bool> {
691    use crate::rdf_star::RdfStarParser;
692    use crate::sparql_library::parsers::turtle_star::TurtleStarParser;
693    use std::fs::File;
694    use std::io::BufReader;
695
696    // Retained only as a fast legacy diagnostic. XOR has compensating-error
697    // collisions; `verify-graph` performs the bounded encoded-set proof.
698    let mut source_checksum: u64 = 0;
699    let mut source_records: u64 = 0;
700    let file = File::open(&input_path)?;
701    let mut reader = BufReader::new(file);
702    let mut parser = TurtleStarParser::new(0);
703
704    let mut buffer = Vec::new();
705    while {
706        buffer.clear();
707        std::io::BufRead::read_until(&mut reader, b'\n', &mut buffer)? > 0
708    } {
709        let mut slice = buffer.as_slice();
710        if slice.ends_with(b"\r\n") {
711            slice = &slice[..slice.len() - 2];
712        } else if slice.ends_with(b"\n") {
713            slice = &slice[..slice.len() - 1];
714        }
715
716        if slice.is_empty() || slice[0] == b'#' || slice.iter().all(|b| b.is_ascii_whitespace()) {
717            continue;
718        }
719
720        let (s, p, o) = parser.parse_triple(slice).map_err(|e| {
721            std::io::Error::new(
722                std::io::ErrorKind::InvalidData,
723                format!("source contains an unparseable RDF triple: {e}"),
724            )
725        })?;
726        source_checksum ^= s ^ p ^ o;
727        source_records += 1;
728    }
729
730    println!("Source Checksum: 0x{:016X}", source_checksum);
731
732    // Dataset calculation
733    let mut dataset_checksum: u64 = 0;
734    let mut dataset_records: u64 = 0;
735
736    let volume = match crate::q42_volume::Q42Volume::open(&dataset_path) {
737        Ok(v) => v,
738        Err(e) => {
739            return Err(std::io::Error::new(
740                std::io::ErrorKind::InvalidData,
741                format!("Failed to open Q42 volume: {}", e),
742            ))
743        }
744    };
745
746    let mut sb_buf = vec![0u8; crate::q42_volume::SUPERBLOCK_SIZE];
747    for i in 0..volume.block_count() as usize {
748        let _ = volume.read_superblock_into(i, &mut sb_buf)?;
749        let quin_count = u64::from_le_bytes(sb_buf[16..24].try_into().unwrap()) as usize;
750        let mut off = crate::q42_volume::SUPERBLOCK_HEADER;
751        for _ in 0..quin_count {
752            let quin: crate::NQuin =
753                bytemuck::pod_read_unaligned(&sb_buf[off..off + crate::q42_volume::QUIN_SIZE]);
754            if !quin.verify_ecc_parity() {
755                return Err(std::io::Error::new(
756                    std::io::ErrorKind::InvalidData,
757                    format!("Q42 parity mismatch in block {i}"),
758                ));
759            }
760            dataset_checksum ^= quin.parity;
761            dataset_records += 1;
762            off += crate::q42_volume::QUIN_SIZE;
763        }
764    }
765
766    println!("Dataset Checksum: 0x{:016X}", dataset_checksum);
767
768    println!("Source records: {source_records}");
769    println!("Dataset records: {dataset_records}");
770    Ok(source_checksum == dataset_checksum
771        && source_records == dataset_records
772        && source_records != 0)
773}
774
775#[cfg(test)]
776mod tests {
777    use super::*;
778    use tempfile::TempDir;
779
780    #[test]
781    fn rdf_ingest_uses_external_runs_and_embeds_lossless_lexicon() {
782        let dir = TempDir::new().unwrap();
783        let input = dir.path().join("input.nt");
784        let output = dir.path().join("output.q42");
785        std::fs::write(
786            &input,
787            "<https://example.test/s> <https://example.test/p> \"value\" .\n",
788        )
789        .unwrap();
790        assert_eq!(
791            streaming_import_rdf_with_mode(
792                input.to_str().unwrap(),
793                output.to_str().unwrap(),
794                IngestMode::Complete,
795            )
796            .unwrap(),
797            1
798        );
799        let volume = crate::q42_volume::Q42Volume::open(&output).unwrap();
800        assert_eq!(volume.block_count(), 1);
801        assert!(volume.lex_view().unwrap().entry_count() >= 3);
802    }
803
804    #[test]
805    fn empty_turtle_prefixed_names_expand_to_iris() {
806        let src = "@prefix dc: <http://purl.org/dc/terms/> .\n@prefix ns1: <http://purl.org/ontology/mo/> .\n<http://ex/s> rdfs:seeAlso dc: ;\n  rdfs:isDefinedBy ns1: .\n";
807        let expanded = expand_empty_turtle_prefixed_names(src);
808        assert!(expanded.contains("<http://purl.org/dc/terms/>"));
809        assert!(expanded.contains("<http://purl.org/ontology/mo/>"));
810        assert!(!expanded.contains(" dc: ;"));
811        assert!(!expanded.contains(" ns1: ."));
812    }
813
814    #[test]
815    fn turtle_with_empty_prefix_names_ingests() {
816        let dir = TempDir::new().unwrap();
817        let input = dir.path().join("music-mini.ttl");
818        let output = dir.path().join("music-mini.q42");
819        std::fs::write(
820            &input,
821            "@prefix dc: <http://purl.org/dc/terms/> .\n\
822             @prefix ns1: <http://purl.org/ontology/mo/> .\n\
823             @prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .\n\
824             <http://purl.org/ontology/mo/Track> rdfs:isDefinedBy ns1: .\n\
825             <http://purl.org/ontology/mo/> dc:title \"The Music Ontology\" .\n",
826        )
827        .unwrap();
828        let written =
829            streaming_import_rdf(input.to_str().unwrap(), output.to_str().unwrap()).unwrap();
830        assert!(written >= 1, "expected at least one SuperBlock, got {written}");
831        let report = crate::q42_volume::Q42InspectReport::from_path(&output).unwrap();
832        assert!(!report.lexicon_has_no_terms);
833        assert!(report.flags & crate::q42_volume::FLAG_PERMISSIVE_COMMONS != 0);
834    }
835
836    #[test]
837    fn rdfxml_empty_xml_base_ingests_with_catalog_base() {
838        let dir = TempDir::new().unwrap();
839        let input = dir.path().join("earl.rdf");
840        let output = dir.path().join("earl.q42");
841        std::fs::write(
842            &input,
843            concat!(
844                "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n",
845                "<rdf:RDF xml:base=\"\"\n",
846                "         xmlns:rdf=\"http://www.w3.org/1999/02/22-rdf-syntax-ns#\"\n",
847                "         xmlns:rdfs=\"http://www.w3.org/2000/01/rdf-schema#\">\n",
848                "  <rdf:Description rdf:about=\"#Assertion\">\n",
849                "    <rdfs:label xml:lang=\"en\">Assertion</rdfs:label>\n",
850                "  </rdf:Description>\n",
851                "</rdf:RDF>\n",
852            ),
853        )
854        .unwrap();
855        let written = streaming_import_rdf(input.to_str().unwrap(), output.to_str().unwrap())
856            .expect("empty xml:base must not fail closed after catalog base is applied");
857        assert!(written >= 1);
858        let report = crate::q42_volume::Q42InspectReport::from_path(&output).unwrap();
859        assert!(!report.lexicon_has_no_terms);
860    }
861
862    #[test]
863    fn broken_turtle_fails_closed() {
864        let dir = TempDir::new().unwrap();
865        let input = dir.path().join("bad.ttl");
866        let output = dir.path().join("bad.q42");
867        std::fs::write(&input, "@prefix broken\n").unwrap();
868        let err = streaming_import_rdf(input.to_str().unwrap(), output.to_str().unwrap())
869            .unwrap_err();
870        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
871        assert!(err.to_string().contains("parse failed"));
872    }
873}
qualia_core_db/query/ingest.rs

qualia_core_db/query/
ingest.rs
