qualia_cli/ingest/
mod.rs

1//! N-Triples → unified `.q42` v2 volume (embedded lex + bidx + LZ4 SuperBlocks).
2//!
3//! Reads N-Triples line-by-line, buffers all [`NQuin`] records, sorts by
4//! object hash, then writes a single v2 volume via [`qualia_core_db::q42_volume`].
5//!
6//! Legacy v1 sidecars (`.q42.lex`, `.q42.bidx`) remain readable via
7//! [`qualia_core_db::q42_lex::Q42Lexicon::load_for_q42`] but are no longer emitted.
8
9use std::collections::HashMap;
10use std::fs::File;
11use std::io::{BufRead, BufReader};
12use std::path::Path;
13
14use qualia_core_db::external_sort::ExternalSorter;
15use qualia_core_db::mini_parser::hash_token;
16use qualia_core_db::{NQuin, QUINS_PER_BLOCK};
17use rio_api::parser::TriplesParser;
18use rio_xml::RdfXmlParser;
19
20pub mod agent_intent;
21pub mod csv_mapper;
22pub mod detect;
23pub mod json_mapper;
24pub mod mapper;
25pub mod pipeline;
26pub mod writer;
27
28#[derive(Debug)]
29pub enum IngestError {
30    Io(std::io::Error),
31    Other(String),
32}
33
34impl From<std::io::Error> for IngestError {
35    fn from(err: std::io::Error) -> Self {
36        IngestError::Io(err)
37    }
38}
39
40impl std::fmt::Display for IngestError {
41    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
42        match self {
43            IngestError::Io(e) => write!(f, "IO Error: {}", e),
44            IngestError::Other(e) => write!(f, "{}", e),
45        }
46    }
47}
48impl std::error::Error for IngestError {}
49
50// ---------------------------------------------------------------------------
51// Public API
52// ---------------------------------------------------------------------------
53
54/// Statistics returned after a successful ingest.
55#[derive(Debug)]
56pub struct IngestStats {
57    pub triples_ingested: u64,
58    pub blocks_written: u64,
59    pub lex_entries: u64,
60    pub lines_skipped: u64,
61    pub bidx_written: bool,
62}
63
64/// Ingest an N-Triples file at `input` and write a unified v2 `.q42` to `output`.
65///
66/// Uses ExternalSorter (out-of-core K-way merge, ~48 MB peak) to stay within
67/// the 512 MB RAM floor on arbitrarily large inputs.
68pub fn ingest_ntriples(
69    input: &Path,
70    output: &Path,
71) -> Result<IngestStats, Box<dyn std::error::Error>> {
72    let reader = BufReader::new(File::open(input)?);
73
74    let temp_dir = std::env::temp_dir().join("qualia_sort_nt");
75    let mut sorter = ExternalSorter::new(temp_dir);
76    let mut triples: u64 = 0;
77    let mut skipped: u64 = 0;
78
79    for raw_line in reader.lines() {
80        let line = raw_line?;
81        let line = line.trim();
82        if line.is_empty() || line.starts_with('#') {
83            skipped += 1;
84            continue;
85        }
86        let Some((s, p, o)) = parse_nt_line(line) else {
87            skipped += 1;
88            continue;
89        };
90        let sh = hash_token(s);
91        let ph = hash_token(p);
92        let oh = hash_token(o);
93
94        sorter.push(NQuin {
95            subject: sh,
96            predicate: ph,
97            object: oh,
98            context: 0,
99            metadata: 0,
100            parity: NQuin::calculate_parity(sh, ph, oh, 0, 0),
101        })?;
102        triples += 1;
103    }
104
105    let block_seq = sorter.merge(output)?;
106
107    Ok(IngestStats {
108        triples_ingested: triples,
109        blocks_written: block_seq,
110        lex_entries: 0,
111        lines_skipped: skipped,
112        bidx_written: true,
113    })
114}
115
116/// Ingest an RDF/XML file at `input` — same output format as [`ingest_ntriples`].
117///
118/// Streams triples into ExternalSorter so peak RAM stays bounded at ~48 MB
119/// regardless of input size.
120pub fn ingest_rdf_xml(
121    input: &Path,
122    output: &Path,
123) -> Result<IngestStats, Box<dyn std::error::Error>> {
124    let reader = BufReader::new(File::open(input)?);
125
126    let temp_dir = std::env::temp_dir().join("qualia_sort_xml");
127    let mut sorter = ExternalSorter::new(temp_dir);
128    let mut triples: u64 = 0;
129    let mut parse_err: Option<std::io::Error> = None;
130
131    let mut parser = RdfXmlParser::new(reader, None);
132    let _ = parser.parse_all(
133        &mut |t: rio_api::model::Triple| -> Result<(), std::io::Error> {
134            let s = t.subject.to_string();
135            let p = t.predicate.to_string();
136            let o = t.object.to_string();
137
138            let sh = hash_token(&s);
139            let ph = hash_token(&p);
140
141            let mut oh = None;
142
143            if let rio_api::model::Term::Literal(rio_api::model::Literal::Typed {
144                value,
145                datatype,
146            }) = t.object
147            {
148                let dt = datatype.iri;
149                if dt == "http://www.w3.org/2001/XMLSchema#integer" {
150                    if let Ok(num) = value.parse::<i64>() {
151                        let max_val = (1i64 << 59) - 1;
152                        let min_val = -(1i64 << 59);
153                        if num >= min_val && num <= max_val {
154                            let unsigned =
155                                (num as u64) & qualia_core_db::resolver::INLINE_VALUE_MASK;
156                            oh = Some(qualia_core_db::resolver::INLINE_TAG_INTEGER | unsigned);
157                        }
158                    }
159                } else if dt == "http://www.w3.org/2001/XMLSchema#decimal" {
160                    if let Ok(num) = value.parse::<f64>() {
161                        let scaled = num * 1_000_000.0;
162                        let max_val = ((1i64 << 59) - 1) as f64;
163                        let min_val = (-(1i64 << 59)) as f64;
164                        if scaled >= min_val && scaled <= max_val {
165                            let num_i64 = scaled.round() as i64;
166                            let unsigned =
167                                (num_i64 as u64) & qualia_core_db::resolver::INLINE_VALUE_MASK;
168                            oh = Some(qualia_core_db::resolver::INLINE_TAG_DECIMAL | unsigned);
169                        }
170                    }
171                } else if dt == "http://www.w3.org/2001/XMLSchema#boolean" {
172                    if value == "true" || value == "1" {
173                        oh = Some(qualia_core_db::resolver::INLINE_TAG_BOOLEAN | 1);
174                    } else if value == "false" || value == "0" {
175                        oh = Some(qualia_core_db::resolver::INLINE_TAG_BOOLEAN | 0);
176                    }
177                }
178            }
179
180            let oh = oh.unwrap_or_else(|| hash_token(&o) & 0x0FFF_FFFF_FFFF_FFFF);
181
182            sorter
183                .push(NQuin {
184                    subject: sh,
185                    predicate: ph,
186                    object: oh,
187                    context: 0,
188                    metadata: 0,
189                    parity: NQuin::calculate_parity(sh, ph, oh, 0, 0),
190                })
191                .map_err(|e| {
192                    let io_err = std::io::Error::new(std::io::ErrorKind::Other, e.to_string());
193                    parse_err = Some(std::io::Error::new(
194                        std::io::ErrorKind::Other,
195                        e.to_string(),
196                    ));
197                    io_err
198                })?;
199            triples += 1;
200            Ok(())
201        },
202    );
203
204    if let Some(e) = parse_err {
205        return Err(Box::new(e));
206    }
207
208    let block_seq = sorter.merge(output)?;
209
210    Ok(IngestStats {
211        triples_ingested: triples,
212        blocks_written: block_seq,
213        lex_entries: 0,
214        lines_skipped: 0,
215        bidx_written: true,
216    })
217}
218
219// ---------------------------------------------------------------------------
220// N-Triples line parser
221// ---------------------------------------------------------------------------
222
223fn parse_nt_line(line: &str) -> Option<(&str, &str, &str)> {
224    let mut tokens = line.split_ascii_whitespace();
225    let s = tokens.next()?;
226    let p = tokens.next()?;
227    let o = tokens.next()?;
228    Some((s, p, o))
229}
230
231pub fn ingest_chk(input: &Path, output: &Path) -> Result<IngestStats, Box<dyn std::error::Error>> {
232    let reader = File::open(input)?;
233    let temp_dir = std::env::temp_dir().join("qualia_sort_chk");
234    let mut sorter = ExternalSorter::new(temp_dir);
235
236    // .chk format does not use a lexicon currently
237    let triples = qualia_core_db::parsers::chk_parser::parse_chk_stream(reader, 0, &mut sorter)?;
238
239    let block_seq = sorter.merge(output)?;
240
241    Ok(IngestStats {
242        triples_ingested: triples,
243        blocks_written: block_seq,
244        lex_entries: 0,
245        lines_skipped: 0,
246        bidx_written: true,
247    })
248}
249
250pub fn ingest_cbor(input: &Path, output: &Path) -> Result<IngestStats, Box<dyn std::error::Error>> {
251    let file = File::open(input)?;
252    let file_size = file.metadata()?.len();
253    if file_size > 256 * 1024 * 1024 {
254        return Err(format!(
255            "CBOR input is {} MB — exceeds 256 MB guard. Split into smaller files.",
256            file_size / (1024 * 1024)
257        )
258        .into());
259    }
260    // Safety: file size checked above; mmap avoids heap copy of raw bytes.
261    let mmap = unsafe { memmap2::Mmap::map(&file)? };
262    let buffer: &[u8] = &mmap;
263
264    let temp_dir = std::env::temp_dir().join("qualia_sort_cbor");
265    let mut sorter = ExternalSorter::new(temp_dir);
266
267    let triples =
268        qualia_core_db::parsers::cbor_parser::parse_cbor_ld_stream(&buffer, 0, &mut sorter)?;
269
270    let block_seq = sorter.merge(output)?;
271
272    Ok(IngestStats {
273        triples_ingested: triples,
274        blocks_written: block_seq,
275        lex_entries: 0,
276        lines_skipped: 0,
277        bidx_written: true,
278    })
279}
280
281/// Ingest a Turtle-Star file with SPARQL-Star embedded triples.
282///
283/// This function uses the new LexiconEntry type to support embedded triples
284/// in addition to regular string lexicon entries.
285pub fn ingest_turtle_star(
286    input: &Path,
287    output: &Path,
288) -> Result<IngestStats, Box<dyn std::error::Error>> {
289    let parent_dir = output.parent().unwrap_or(Path::new("."));
290    let ingestor = pipeline::IncrementalIngestor::new(parent_dir, 256 * 1024 * 1024);
291
292    // We map the custom IngestError to Box<dyn std::error::Error>
293    ingestor
294        .execute_stream_compilation(input, output)
295        .map_err(|e| Box::new(e) as Box<dyn std::error::Error>)?;
296
297    // For now we return empty stats since we didn't track them in the pipeline struct.
298    Ok(IngestStats {
299        triples_ingested: 0,
300        blocks_written: 0,
301        lex_entries: 0,
302        lines_skipped: 0,
303        bidx_written: true,
304    })
305}
306
307/// Ingest a KML file into a `.q42` volume via `kml_bridge::import_kml`.
308///
309/// Each `<Placemark>` becomes a set of GeoSPARQL + PROV-O quins.  The string
310/// lexicon returned by the bridge is merged into the volume's embedded lexicon.
311pub fn ingest_kml(input: &Path, output: &Path) -> Result<IngestStats, Box<dyn std::error::Error>> {
312    let file = File::open(input)?;
313    let file_size = file.metadata()?.len();
314    if file_size > 256 * 1024 * 1024 {
315        return Err(format!(
316            "KML input is {} MB — exceeds 256 MB guard. Split into smaller files.",
317            file_size / (1024 * 1024)
318        )
319        .into());
320    }
321    // Safety: file size checked above; OS maps pages on-demand, no heap copy.
322    let mmap = unsafe { memmap2::Mmap::map(&file)? };
323    let bytes: &[u8] = &mmap;
324
325    let (quins, str_lex) = qualia_core_db::kml_bridge::import_kml(bytes)
326        .map_err(|e| format!("KML parse error: {e}"))?;
327
328    // Convert the string lexicon into `LexiconEntry::String` entries.
329    let lex: HashMap<u64, qualia_core_db::q42_lex::LexiconEntry> = str_lex
330        .into_iter()
331        .map(|(k, v)| (k, qualia_core_db::q42_lex::LexiconEntry::String(v)))
332        .collect();
333
334    let mut all_quins = quins;
335    all_quins.sort_unstable_by_key(|q| q.object);
336
337    let mut blocks: Vec<Vec<NQuin>> = Vec::new();
338    let mut block_ranges: Vec<(u64, u64)> = Vec::new();
339    for chunk in all_quins.chunks(QUINS_PER_BLOCK) {
340        let min_hash = chunk.iter().map(|q| q.object).min().unwrap_or(0);
341        let max_hash = chunk.iter().map(|q| q.object).max().unwrap_or(0);
342        block_ranges.push((min_hash, max_hash));
343        blocks.push(chunk.to_vec());
344    }
345
346    let triples_ingested = all_quins.len() as u64;
347    let block_seq = blocks.len() as u64;
348    let lex_entries = lex.len() as u64;
349
350    qualia_core_db::q42_volume::write_unified_volume_with_entries(
351        output,
352        &lex,
353        &block_ranges,
354        &blocks,
355    )?;
356
357    Ok(IngestStats {
358        triples_ingested,
359        blocks_written: block_seq,
360        lex_entries,
361        lines_skipped: 0,
362        bidx_written: true,
363    })
364}
365
366/// Ingest a 3D asset (OBJ / STL / GLB) into a `.q42` volume via `asset_bridge`.
367///
368/// The geometry is parsed to a bounding-boxed `Mesh`; `mesh_to_nquins` emits the *semantic* quins
369/// (type, vertex/triangle counts, bounding box, centroid, source format) which are written to the
370/// volume — the artefact is *semantically known*, not just drawn. (Raw vertex/index buffers feed
371/// the GPU renderer separately — Phase 1.2, RENDERER_IMPLEMENTATION_PLAN.)
372pub fn ingest_asset(
373    input: &Path,
374    output: &Path,
375) -> Result<IngestStats, Box<dyn std::error::Error>> {
376    let file = File::open(input)?;
377    let file_size = file.metadata()?.len();
378    if file_size > 256 * 1024 * 1024 {
379        return Err(format!(
380            "asset input is {} MB — exceeds 256 MB guard. Split into smaller files.",
381            file_size / (1024 * 1024)
382        )
383        .into());
384    }
385    // Safety: file size checked above; OS maps pages on-demand, no heap copy.
386    let mmap = unsafe { memmap2::Mmap::map(&file)? };
387    let bytes: &[u8] = &mmap;
388
389    let hint = input.extension().and_then(|e| e.to_str());
390    let mesh = qualia_core_db::render::assets::import_asset(bytes, hint)
391        .map_err(|e| format!("asset parse error: {e}"))?;
392
393    let asset_uri = format!(
394        "urn:qualia:asset:{}",
395        input.file_name().and_then(|s| s.to_str()).unwrap_or("mesh")
396    );
397    let (quins, str_lex) =
398        qualia_core_db::render::assets::mesh_to_nquins(&mesh, &asset_uri, hint.unwrap_or("mesh"));
399
400    // Convert the string lexicon into `LexiconEntry::String` entries.
401    let lex: HashMap<u64, qualia_core_db::q42_lex::LexiconEntry> = str_lex
402        .into_iter()
403        .map(|(k, v)| (k, qualia_core_db::q42_lex::LexiconEntry::String(v)))
404        .collect();
405
406    let mut all_quins = quins;
407    all_quins.sort_unstable_by_key(|q| q.object);
408
409    let mut blocks: Vec<Vec<NQuin>> = Vec::new();
410    let mut block_ranges: Vec<(u64, u64)> = Vec::new();
411    for chunk in all_quins.chunks(QUINS_PER_BLOCK) {
412        let min_hash = chunk.iter().map(|q| q.object).min().unwrap_or(0);
413        let max_hash = chunk.iter().map(|q| q.object).max().unwrap_or(0);
414        block_ranges.push((min_hash, max_hash));
415        blocks.push(chunk.to_vec());
416    }
417
418    let triples_ingested = all_quins.len() as u64;
419    let block_seq = blocks.len() as u64;
420    let lex_entries = lex.len() as u64;
421
422    qualia_core_db::q42_volume::write_unified_volume_with_entries(
423        output,
424        &lex,
425        &block_ranges,
426        &blocks,
427    )?;
428
429    Ok(IngestStats {
430        triples_ingested,
431        blocks_written: block_seq,
432        lex_entries,
433        lines_skipped: 0,
434        bidx_written: true,
435    })
436}
437
438// ──────────────────────────────────────────────────────────────────────────────
439// Phase 2: wrappers for core-db RDF-Star parsers (all streaming via ExternalSorter)
440// ──────────────────────────────────────────────────────────────────────────────
441
442macro_rules! stream_ingest {
443    ($name:ident, $parse_fn:path, $temp_suffix:literal) => {
444        pub fn $name(
445            input: &Path,
446            output: &Path,
447        ) -> Result<IngestStats, Box<dyn std::error::Error>> {
448            let reader = File::open(input)?;
449            let temp_dir = std::env::temp_dir().join($temp_suffix);
450            let mut sorter = ExternalSorter::new(temp_dir);
451            let triples = $parse_fn(reader, 0, &mut sorter)?;
452            let block_seq = sorter.merge(output)?;
453            Ok(IngestStats {
454                triples_ingested: triples,
455                blocks_written: block_seq,
456                lex_entries: 0,
457                lines_skipped: 0,
458                bidx_written: true,
459            })
460        }
461    };
462}
463
464stream_ingest!(
465    ingest_ntriples_star,
466    qualia_core_db::parsers::ntriples_star::parse_ntriples_star_stream,
467    "qualia_sort_nts"
468);
469
470stream_ingest!(
471    ingest_nquads,
472    qualia_core_db::parsers::nquads_star::parse_nquads_star_stream,
473    "qualia_sort_nq"
474);
475
476stream_ingest!(
477    ingest_nquads_star,
478    qualia_core_db::parsers::nquads_star::parse_nquads_star_stream,
479    "qualia_sort_nqs"
480);
481
482stream_ingest!(
483    ingest_turtle,
484    qualia_core_db::parsers::turtle_doc::parse_turtle_doc_stream,
485    "qualia_sort_ttl"
486);
487
488stream_ingest!(
489    ingest_trig,
490    qualia_core_db::parsers::trig_star::parse_trig_star_stream,
491    "qualia_sort_trig"
492);
493
494stream_ingest!(
495    ingest_trig_star,
496    qualia_core_db::parsers::trig_star::parse_trig_star_stream,
497    "qualia_sort_trigs"
498);
499
500stream_ingest!(
501    ingest_n3,
502    qualia_core_db::parsers::turtle_doc::parse_turtle_doc_stream,
503    "qualia_sort_n3"
504);
505
506stream_ingest!(
507    ingest_json_ld,
508    qualia_core_db::parsers::json_ld_stream::parse_json_ld_stream,
509    "qualia_sort_jsonld"
510);
511
512pub fn ingest_json_ld_star(
513    input: &Path,
514    output: &Path,
515) -> Result<IngestStats, Box<dyn std::error::Error>> {
516    let reader = File::open(input)?;
517    let temp_dir = std::env::temp_dir().join("qualia_sort_jsonlds");
518    let mut sorter = ExternalSorter::new(temp_dir);
519    let triples = qualia_core_db::parsers::json_ld_stream::parse_json_ld_star_stream(
520        reader,
521        0,
522        &mut sorter,
523        true,
524    )?;
525    let block_seq = sorter.merge(output)?;
526    Ok(IngestStats {
527        triples_ingested: triples,
528        blocks_written: block_seq,
529        lex_entries: 0,
530        lines_skipped: 0,
531        bidx_written: true,
532    })
533}
534
535/// Dispatch to the correct ingest function based on auto-detected format.
536pub fn ingest_auto(
537    input: &Path,
538    output: &Path,
539) -> Result<(IngestStats, detect::SemanticFormat), Box<dyn std::error::Error>> {
540    let fmt = detect::detect_format(input).ok_or_else(|| {
541        format!(
542            "Cannot auto-detect format for '{}'. Use --format to specify.",
543            input.display()
544        )
545    })?;
546
547    let stats = match fmt {
548        detect::SemanticFormat::NTriples => ingest_ntriples(input, output)?,
549        detect::SemanticFormat::NTriplesStar => ingest_ntriples_star(input, output)?,
550        detect::SemanticFormat::NQuads => ingest_nquads(input, output)?,
551        detect::SemanticFormat::NQuadsStar => ingest_nquads_star(input, output)?,
552        detect::SemanticFormat::Turtle => ingest_turtle(input, output)?,
553        detect::SemanticFormat::TurtleStar => ingest_turtle_star(input, output)?,
554        detect::SemanticFormat::TriG => ingest_trig(input, output)?,
555        detect::SemanticFormat::TriGStar => ingest_trig_star(input, output)?,
556        detect::SemanticFormat::N3 => ingest_n3(input, output)?,
557        detect::SemanticFormat::RdfXml => ingest_rdf_xml(input, output)?,
558        detect::SemanticFormat::JsonLd => ingest_json_ld(input, output)?,
559        detect::SemanticFormat::JsonLdStar => ingest_json_ld_star(input, output)?,
560        detect::SemanticFormat::AgentIntentJsonl => {
561            agent_intent::ingest_agent_intent(input, output)?
562        }
563        detect::SemanticFormat::CborLd => ingest_cbor(input, output)?,
564        detect::SemanticFormat::Kml => ingest_kml(input, output)?,
565        detect::SemanticFormat::Mesh => ingest_asset(input, output)?,
566        detect::SemanticFormat::Chk => ingest_chk(input, output)?,
567        detect::SemanticFormat::Q42 => {
568            return Err("Q42 vaults are already in native format — no ingestion needed.".into())
569        }
570    };
571
572    Ok((stats, fmt))
573}
qualia_cli/ingest/mod.rs

qualia_cli/ingest/
mod.rs
