Skip to main content

rete_core/
ingest.rs

1//! Ingestion: parse RDF text (N-Triples / N-Quads / Turtle) and assemble a
2//! complete `.rete` file image. Shared by the CLI's `build`/`validate` commands
3//! and the wasm bindings (the playground's in-browser builder).
4//!
5//! The N-Triples/N-Quads reader is line-based and keeps terms as their exact
6//! canonical token strings (`<iri>`, `_:bnode`, `"lit"`, `"lit"^^<dt>`,
7//! `"lit"@lang`) so they double as dictionary keys and so a `query` can match
8//! by the same string. This is not a full RDF 1.1 parser — it covers the
9//! canonical N-Triples surface (and is deliberately tolerant of IRIs a strict
10//! parser would reject), enough for v0 ingestion. Turtle goes through `oxttl`.
11//!
12//! That tolerance is *measured*, not silent: pass an [`IriAudit`] to any of the
13//! `*_audited` entry points and the parse counts every IRI outside the
14//! N-Triples `IRIREF` grammar / RFC 3987 (see [`crate::iri`]) — or, with
15//! `strict`, refuses the input on the first one. `rete build` always audits.
16//!
17//! Assembly degrades with the build features: without the `compression`
18//! feature (e.g. on wasm) sections are written with the `NONE` codec — larger
19//! files, byte-compatible readers.
20
21use crate::{Dictionary, DictionaryBuilder, GraphIndexBuilder, PyramidAlgo, DEFAULT_TILE_BUDGET};
22
23mod jelly;
24
25/// A parsed triple as three canonical term tokens.
26pub type RawTriple = (String, String, String);
27
28/// A parsed quad: the triple plus an optional graph term (`None` = default graph).
29pub type RawQuad = (String, String, String, Option<String>);
30
31/// Which **surface** a `<< … >>` in Turtle/TriG input is read as — the input
32/// half of `rete export --quoted-triple-syntax`, same vocabulary, same two
33/// values.
34///
35/// The flag exists because the two standards give one piece of syntax two
36/// meanings, and no amount of looking at the bytes can tell them apart:
37///
38/// | written            | under [`RdfStar`](Self::RdfStar) | under [`Rdf12`](Self::Rdf12)          |
39/// |--------------------|----------------------------------|---------------------------------------|
40/// | `<< s p o >>`      | a **quoted triple**, one term    | a **reifier**: `_:r rdf:reifies <<( s p o )>>`, plus a blank node standing where it was |
41/// | `<<( s p o )>>`    | a syntax error                   | a **triple term**, object position only |
42/// | `{\| … \|}`         | a syntax error                   | an annotation on the statement before it |
43///
44/// One file, two readings, two different graphs — see the round-trip matrix in
45/// `crates/rete-cli/tests/quoted_triple_surfaces.rs`, which asserts both.
46///
47/// `<<( s p o )>>` is unambiguous: it is RDF 1.2's and nothing else's. That
48/// asymmetry is why [`RdfStar`](Self::RdfStar) is the **default**. Reading an
49/// RDF-star file as RDF 1.2 silently yields a different graph; reading an
50/// RDF 1.2 file as RDF-star is a hard parse error that names the flag. Only one
51/// of the two mistakes is survivable, so the default is the one that makes the
52/// other mistake loud.
53///
54/// **N-Triples and N-Quads ignore this entirely.** Their reader is rete's own
55/// (`take_term`), it has accepted both `<< s p o >>` and `<<( s p o )>>` since
56/// #262, and RDF 1.2 N-Triples has no reifier syntax for `<< … >>` to be — so
57/// there is no ambiguity to resolve and nothing to choose. The same goes for
58/// RDF/XML.
59///
60/// Whichever surface reads the file, what gets **stored** is the same canonical
61/// token `<<s p o>>`. rete's term model is a superset of both: a triple term is
62/// a term, and RDF 1.2 reification is ordinary RDF — a blank node, a predicate
63/// and a term — which rete already stored before this flag existed.
64#[derive(Clone, Copy, PartialEq, Eq, Debug, Default, Hash)]
65pub enum QuotedTripleSurface {
66    /// `<< s p o >>` is a **quoted triple** (RDF-star), in subject or object
67    /// position — rete's own storage token. The default, and byte-for-byte the
68    /// behaviour every rete release before this flag had.
69    #[default]
70    RdfStar,
71    /// `<<( s p o )>>` is a **triple term** and `<< s p o >>` is a **reifier**
72    /// (RDF 1.2), with `{| … |}` annotations read too.
73    Rdf12,
74}
75
76impl QuotedTripleSurface {
77    /// Parse the `--quoted-triple-syntax` value, `None` for anything else.
78    /// Shared by `rete build`, `rete validate` and `rete export` so the flag
79    /// cannot come to mean two things in two directions.
80    pub fn parse(s: &str) -> Option<Self> {
81        match s {
82            crate::card_derive::QUOTED_TRIPLE_SURFACE_RDF12 => Some(Self::Rdf12),
83            crate::card_derive::QUOTED_TRIPLE_SURFACE_RDF_STAR => Some(Self::RdfStar),
84            _ => None,
85        }
86    }
87
88    /// The flag spelling of this surface — the inverse of [`Self::parse`].
89    pub const fn as_str(self) -> &'static str {
90        match self {
91            Self::Rdf12 => crate::card_derive::QUOTED_TRIPLE_SURFACE_RDF12,
92            Self::RdfStar => crate::card_derive::QUOTED_TRIPLE_SURFACE_RDF_STAR,
93        }
94    }
95}
96
97/// The one-line hint appended to a Turtle/TriG parse error when the input looks
98/// like RDF 1.2 and was read as RDF-star — i.e. the exact mistake
99/// [`QuotedTripleSurface`]'s default makes, made loud.
100const RDF12_SURFACE_HINT: &str = "\nhint: this input contains `<<(`, the RDF 1.2 triple-term \
101     syntax — which is what `rete export` writes by default. Reading it takes \
102     `--quoted-triple-syntax rdf12`; the default, `rdf-star`, reads `<< s p o >>` as a quoted \
103     triple instead.";
104
105/// Does `text` look like it carries RDF 1.2 triple terms? A deliberately dumb
106/// substring test: `<<(` cannot begin anything else in Turtle, and this is only
107/// ever consulted to *explain a parse that already failed*, never to decide how
108/// to read a file. A `<<(` inside a string literal would make it a false
109/// positive on some other error's hint, which costs one confusing sentence and
110/// nothing else.
111fn looks_like_rdf12(text: &str) -> bool {
112    text.contains("<<(")
113}
114
115/// Why an ingest failed. Parser errors from `oxttl`/`oxrdfxml` and I/O errors
116/// are flattened to their `Display` string rather than wrapped: the variants
117/// then carry no foreign types, which keeps the error usable from the wasm
118/// bindings and stable across upstream parser upgrades.
119#[derive(Debug, thiserror::Error)]
120#[non_exhaustive]
121pub enum IngestError {
122    /// A malformed N-Triples/N-Quads line: its 1-based line number and a short
123    /// reason (`"bad subject"`, `"missing trailing '.'"`, …).
124    #[error("line {0}: {1}")]
125    Line(usize, &'static str),
126    /// The Turtle parser rejected the input.
127    #[error("turtle: {0}")]
128    Turtle(String),
129    /// The TriG parser rejected the input.
130    #[error("trig: {0}")]
131    TriG(String),
132    /// The RDF/XML parser rejected the input.
133    #[error("rdf/xml: {0}")]
134    RdfXml(String),
135    /// The requested format is not one this crate can parse.
136    #[error("unknown input format: {0} (expected nt, nq, ttl, trig, rdf/xml, or jelly)")]
137    UnknownFormat(String),
138    /// A Jelly file that is malformed, uses a feature this reader refuses, or
139    /// breaks a rule of the specification.
140    #[error("jelly: {0}")]
141    Jelly(String),
142    /// [`QuotedTripleSurface::Rdf12`] was asked for on a Turtle/TriG input, but
143    /// this build of `rete-core` has the `rdf12-turtle` feature off, so the
144    /// RDF 1.2 reader was not compiled in. Refused by name rather than read as
145    /// RDF-star, which would be the wrong graph told in silence.
146    #[error(
147        "this build cannot read the RDF 1.2 Turtle/TriG surface: rete-core was compiled without \
148         the `rdf12-turtle` feature (N-Triples/N-Quads still accept `<<( s p o )>>`)"
149    )]
150    Rdf12TurtleUnavailable,
151    /// The input could not be read (streaming builds surface the path here).
152    #[error("io: {0}")]
153    Io(String),
154    /// An IRI outside the N-Triples/N-Quads `IRIREF` grammar / RFC 3987, refused
155    /// because the caller asked for a strict [`IriAudit`] (`rete build
156    /// --strict`). Without `strict` the same finding is only counted.
157    /// `location` is `line N` for the line-based syntaxes, the format name for
158    /// the others (oxttl reports its own positions).
159    #[error("{location}: invalid IRI {token} — {reason}")]
160    InvalidIri {
161        /// Where it was seen: `line 42`, or the format name.
162        location: String,
163        /// The offending term token, verbatim.
164        token: String,
165        /// Which rule it breaks ([`crate::iri::IriDefect::reason`]).
166        reason: &'static str,
167    },
168}
169
170/// Policy **and** tally for the invalid-IRI check that runs during ingest.
171///
172/// rete's reader is deliberately tolerant of IRIs a strict parser rejects (see
173/// [`crate::iri`]), which is what lets it hold every published dataset — and
174/// also what let three of the `scholar/` exports produce N-Quads Oxigraph
175/// refuses to load. Passing an audit makes the tolerance *visible*: the build
176/// still succeeds and stores exactly what it was given, but it can now say how
177/// much of what it stored is not valid RDF. `strict` turns the same finding into
178/// a refusal, on the first offending statement.
179#[derive(Debug, Default)]
180pub struct IriAudit {
181    /// Fail the parse on the first invalid IRI instead of counting it.
182    pub strict: bool,
183    /// What the parse saw. Constant memory — see [`crate::iri::IriReport`].
184    pub report: crate::iri::IriReport,
185}
186
187impl IriAudit {
188    /// Count invalid IRIs; never fail (`rete build`'s default).
189    pub fn counting() -> Self {
190        Self::default()
191    }
192
193    /// Refuse the input at the first invalid IRI (`rete build --strict`).
194    pub fn refusing() -> Self {
195        Self {
196            strict: true,
197            report: crate::iri::IriReport::default(),
198        }
199    }
200}
201
202/// Run one statement past the audit: record any invalid IRI, and in `strict`
203/// mode turn it into an error naming the offending token. `location` is only
204/// called on the error path, so the common case allocates nothing.
205fn audit_quad<F: FnOnce() -> String>(
206    audit: Option<&mut IriAudit>,
207    s: &str,
208    p: &str,
209    o: &str,
210    g: Option<&str>,
211    location: F,
212) -> Result<(), IngestError> {
213    let Some(a) = audit else { return Ok(()) };
214    let Some(defect) = a.report.observe_quad(s, p, o, g) else {
215        return Ok(());
216    };
217    if !a.strict {
218        return Ok(());
219    }
220    let token = [Some(s), Some(p), Some(o), g]
221        .into_iter()
222        .flatten()
223        .find(|t| crate::iri::term_defect(t).is_some())
224        .unwrap_or("<?>")
225        .to_string();
226    Err(IngestError::InvalidIri {
227        location: location(),
228        token,
229        reason: defect.reason(),
230    })
231}
232
233/// Estimate the statement count of an N-Triples/N-Quads text from its newline
234/// count, so the output `Vec` can be pre-sized — a big build otherwise pays
235/// repeated `Vec` doublings, each briefly holding ~2× the (large) spine. An
236/// over-estimate by a few blank/comment lines is harmless (it's a capacity hint).
237fn estimate_statements(input: &str) -> usize {
238    bytecount_newlines(input).max(1)
239}
240
241/// Count `\n` bytes — one linear pass, far cheaper than the parse it sizes.
242fn bytecount_newlines(input: &str) -> usize {
243    input.as_bytes().iter().filter(|&&b| b == b'\n').count()
244}
245
246/// Parse one N-Quads line into a quad, or `None` for a blank/comment line.
247/// Shared by the whole-text [`parse_quads`] and the streaming [`parse_reader`].
248fn parse_nq_line(
249    raw: &str,
250    lineno: usize,
251    audit: Option<&mut IriAudit>,
252) -> Result<Option<RawQuad>, IngestError> {
253    let line = raw.trim();
254    if line.is_empty() || line.starts_with('#') {
255        return Ok(None);
256    }
257    let stripped = line
258        .strip_suffix('.')
259        .ok_or(IngestError::Line(lineno, "missing trailing '.'"))?
260        .trim_end();
261    let (s, rest) = take_term(stripped).ok_or(IngestError::Line(lineno, "bad subject"))?;
262    let (p, rest) =
263        take_term(rest.trim_start()).ok_or(IngestError::Line(lineno, "bad predicate"))?;
264    let (o, rest) = take_term(rest.trim_start()).ok_or(IngestError::Line(lineno, "bad object"))?;
265    let rest = rest.trim();
266    let graph = if rest.is_empty() {
267        None
268    } else {
269        let (g, tail) = take_term(rest).ok_or(IngestError::Line(lineno, "bad graph"))?;
270        if !tail.trim().is_empty() {
271            return Err(IngestError::Line(lineno, "trailing content after graph"));
272        }
273        Some(g)
274    };
275    audit_quad(audit, &s, &p, &o, graph.as_deref(), || {
276        format!("line {lineno}")
277    })?;
278    Ok(Some((s, p, o, graph)))
279}
280
281/// Parse one N-Triples line into a triple, or `None` for a blank/comment line.
282fn parse_nt_line(
283    raw: &str,
284    lineno: usize,
285    audit: Option<&mut IriAudit>,
286) -> Result<Option<RawTriple>, IngestError> {
287    let line = raw.trim();
288    if line.is_empty() || line.starts_with('#') {
289        return Ok(None);
290    }
291    let stripped = line
292        .strip_suffix('.')
293        .ok_or(IngestError::Line(lineno, "missing trailing '.'"))?
294        .trim_end();
295    let (s, rest) = take_term(stripped).ok_or(IngestError::Line(lineno, "bad subject"))?;
296    let (p, rest) =
297        take_term(rest.trim_start()).ok_or(IngestError::Line(lineno, "bad predicate"))?;
298    let (o, rest) = take_term(rest.trim_start()).ok_or(IngestError::Line(lineno, "bad object"))?;
299    if !rest.trim().is_empty() {
300        return Err(IngestError::Line(lineno, "trailing content after object"));
301    }
302    audit_quad(audit, &s, &p, &o, None, || format!("line {lineno}"))?;
303    Ok(Some((s, p, o)))
304}
305
306/// Parse N-Quads text: `subject predicate object [graph] .` per line.
307pub fn parse_quads(input: &str) -> Result<Vec<RawQuad>, IngestError> {
308    parse_quads_audited(input, None)
309}
310
311/// [`parse_quads`] with an optional [`IriAudit`].
312pub fn parse_quads_audited(
313    input: &str,
314    mut audit: Option<&mut IriAudit>,
315) -> Result<Vec<RawQuad>, IngestError> {
316    let mut out = Vec::with_capacity(estimate_statements(input));
317    for (i, raw) in input.lines().enumerate() {
318        if let Some(q) = parse_nq_line(raw, i + 1, audit.as_deref_mut())? {
319            out.push(q);
320        }
321    }
322    Ok(out)
323}
324
325/// Parse N-Triples text into raw term-token triples, skipping blank/comment lines.
326pub fn parse(input: &str) -> Result<Vec<RawTriple>, IngestError> {
327    parse_audited(input, None)
328}
329
330/// [`parse`] with an optional [`IriAudit`].
331pub fn parse_audited(
332    input: &str,
333    mut audit: Option<&mut IriAudit>,
334) -> Result<Vec<RawTriple>, IngestError> {
335    let mut out = Vec::with_capacity(estimate_statements(input));
336    for (i, raw) in input.lines().enumerate() {
337        if let Some(t) = parse_nt_line(raw, i + 1, audit.as_deref_mut())? {
338            out.push(t);
339        }
340    }
341    Ok(out)
342}
343
344/// **Stream-parse** N-Triples (`"nt"`) or N-Quads (`"nq"`) from a reader, one
345/// line at a time, so the whole input text is **never resident** — the big-build
346/// memory win over reading the file into a `String` first (each line String is
347/// transient, freed every iteration). `cap` pre-sizes the output `Vec` (e.g.
348/// `file_len / 64`) to avoid reallocation doublings of the large spine. Turtle is
349/// not streamable here (oxttl needs the whole input); callers use the text path
350/// for `"ttl"`.
351pub fn parse_reader<R: std::io::BufRead>(
352    reader: R,
353    format: &str,
354    cap: usize,
355) -> Result<Vec<RawQuad>, IngestError> {
356    parse_reader_audited(reader, format, cap, None)
357}
358
359/// [`parse_reader`] with an optional [`IriAudit`].
360pub fn parse_reader_audited<R: std::io::BufRead>(
361    reader: R,
362    format: &str,
363    cap: usize,
364    audit: Option<&mut IriAudit>,
365) -> Result<Vec<RawQuad>, IngestError> {
366    parse_reader_audited_surface(reader, format, cap, audit, QuotedTripleSurface::default())
367}
368
369/// [`parse_reader_audited`] reading `<< … >>` in the given
370/// [`QuotedTripleSurface`] (Turtle/TriG only — see the enum).
371pub fn parse_reader_audited_surface<R: std::io::BufRead>(
372    reader: R,
373    format: &str,
374    cap: usize,
375    audit: Option<&mut IriAudit>,
376    surface: QuotedTripleSurface,
377) -> Result<Vec<RawQuad>, IngestError> {
378    let mut out = Vec::with_capacity(cap);
379    stream_reader_audited_surface(reader, format, audit, surface, &mut |q| out.push(q))?;
380    Ok(out)
381}
382
383/// **Stream** N-Triples (`"nt"`) / N-Quads (`"nq"`) from a reader, invoking `f`
384/// once per parsed quad — like [`parse_reader`] but **without collecting** into a
385/// `Vec`. Each quad's term Strings are owned by `f` (and dropped when it returns
386/// if it doesn't retain them), so a caller that only needs to *observe* every
387/// term — e.g. the two-pass [`assemble_dataset_streaming`] building its
388/// dictionary — never materializes the whole quad multiset. The big-graph,
389/// low-RAM ingest primitive. Blank/comment lines are skipped; a parse error stops
390/// the stream and is returned.
391///
392/// Every format streams: the line-based ones a line at a time, Turtle / TriG /
393/// RDF-XML through their `oxttl`/`oxrdfxml` reader parsers, which pull from the
394/// reader incrementally. Nothing here holds the input text — so a 60 GB Turtle
395/// file is as ingestible as a 60 GB `.nt` one, which is what lets the external
396/// (memory-bounded) build accept them.
397pub fn stream_reader<R: std::io::BufRead>(
398    reader: R,
399    format: &str,
400    f: &mut dyn FnMut(RawQuad),
401) -> Result<(), IngestError> {
402    stream_reader_audited(reader, format, None, f)
403}
404
405/// [`stream_reader`] with an optional [`IriAudit`].
406///
407/// Every syntax is audited, not only the line-based ones. `oxttl`/`oxrdfxml`
408/// validate IRIs themselves and reject an invalid one before it reaches us, so
409/// in practice the Turtle/TriG/RDF-XML tallies stay at zero — but auditing them
410/// too means the count is a property of the *file*, not of which reader happened
411/// to produce it, and a future parser change cannot silently reopen the hole.
412pub fn stream_reader_audited<R: std::io::BufRead>(
413    reader: R,
414    format: &str,
415    audit: Option<&mut IriAudit>,
416    f: &mut dyn FnMut(RawQuad),
417) -> Result<(), IngestError> {
418    stream_reader_audited_surface(reader, format, audit, QuotedTripleSurface::default(), f)
419}
420
421/// [`stream_reader_audited`] reading `<< … >>` in the given
422/// [`QuotedTripleSurface`].
423///
424/// The surface reaches only the `"ttl"` and `"trig"` arms: the line-based
425/// readers are rete's own and take both spellings unconditionally, and RDF/XML
426/// has no syntax for either. Under [`QuotedTripleSurface::Rdf12`] Turtle/TriG go
427/// through `oxttl` 0.2 instead of 0.1, and the RDF 1.2 triple terms it hands
428/// back (`<<( s p o )>>`) are folded to rete's stored token on the way out, so
429/// the dictionary a file lands in does not depend on which reader read it.
430///
431/// **This entry point does not carry the RDF 1.2 hint** that the text path adds
432/// to a failed RDF-star parse: it is handed a reader, not a string, and sniffing
433/// the input would mean buffering the 60 GB file this function exists to avoid
434/// buffering. `rete build --memory-budget-mb file.ttl` therefore reports the raw
435/// parser error; every other route reports the hint.
436pub fn stream_reader_audited_surface<R: std::io::BufRead>(
437    reader: R,
438    format: &str,
439    mut audit: Option<&mut IriAudit>,
440    surface: QuotedTripleSurface,
441    f: &mut dyn FnMut(RawQuad),
442) -> Result<(), IngestError> {
443    if surface == QuotedTripleSurface::Rdf12 && matches!(format, "ttl" | "trig") {
444        return rdf12::stream(reader, format, audit, f);
445    }
446    match format {
447        "nt" | "nq" => {
448            for (i, line) in reader.lines().enumerate() {
449                let line = line.map_err(|e| IngestError::Io(e.to_string()))?;
450                if format == "nq" {
451                    if let Some(q) = parse_nq_line(&line, i + 1, audit.as_deref_mut())? {
452                        f(q);
453                    }
454                } else if let Some((s, p, o)) = parse_nt_line(&line, i + 1, audit.as_deref_mut())? {
455                    f((s, p, o, None));
456                }
457            }
458            Ok(())
459        }
460        "ttl" => turtle_star(reader, audit, f),
461        "trig" => trig_star(reader, audit, f),
462        "jelly" => jelly::stream(reader, audit, f),
463        "rdfxml" => {
464            for r in oxrdfxml::RdfXmlParser::new().for_reader(reader) {
465                let t = r.map_err(|e| IngestError::RdfXml(e.to_string()))?;
466                let q = (
467                    t.subject.to_string(),
468                    t.predicate.to_string(),
469                    t.object.to_string(),
470                    None,
471                );
472                audit_quad(audit.as_deref_mut(), &q.0, &q.1, &q.2, None, || {
473                    "rdf/xml".into()
474                })?;
475                f(q);
476            }
477            Ok(())
478        }
479        other => Err(IngestError::UnknownFormat(other.to_string())),
480    }
481}
482
483/// Stream **RDF-star** Turtle: `with_quoted_triples` accepts `<< s p o >>` in
484/// subject/object position, and oxrdf 0.2's `Term::Triple` Displays as the
485/// canonical `<<…>>` token rete's own N-Triples-star tokenizer emits — so no
486/// translation is needed on the way out.
487///
488/// Its own function, and called directly from both the reader path and the text
489/// path, so that the wasm engine — which reaches only the text path and has no
490/// RDF 1.2 reader compiled in — links this and nothing else.
491fn turtle_star<R: std::io::BufRead>(
492    reader: R,
493    mut audit: Option<&mut IriAudit>,
494    f: &mut dyn FnMut(RawQuad),
495) -> Result<(), IngestError> {
496    for r in oxttl::TurtleParser::new()
497        .with_quoted_triples()
498        .for_reader(reader)
499    {
500        let t = r.map_err(|e| IngestError::Turtle(e.to_string()))?;
501        let q = (
502            t.subject.to_string(),
503            t.predicate.to_string(),
504            t.object.to_string(),
505            None,
506        );
507        audit_quad(audit.as_deref_mut(), &q.0, &q.1, &q.2, None, || {
508            "turtle".into()
509        })?;
510        f(q);
511    }
512    Ok(())
513}
514
515/// [`turtle_star`] for TriG — Turtle plus named-graph blocks (`<g> { … }`), the
516/// shape most large RDF dumps ship in.
517fn trig_star<R: std::io::BufRead>(
518    reader: R,
519    mut audit: Option<&mut IriAudit>,
520    f: &mut dyn FnMut(RawQuad),
521) -> Result<(), IngestError> {
522    for r in oxttl::TriGParser::new()
523        .with_quoted_triples()
524        .for_reader(reader)
525    {
526        let q = r.map_err(|e| IngestError::TriG(e.to_string()))?;
527        let q = (
528            q.subject.to_string(),
529            q.predicate.to_string(),
530            q.object.to_string(),
531            graph_token(&q.graph_name),
532        );
533        audit_quad(
534            audit.as_deref_mut(),
535            &q.0,
536            &q.1,
537            &q.2,
538            q.3.as_deref(),
539            || "trig".into(),
540        )?;
541        f(q);
542    }
543    Ok(())
544}
545
546/// The canonical graph token for a parsed quad — `None` for the default graph,
547/// otherwise the same `<iri>` / `_:b` spelling the N-Quads reader produces, so
548/// TriG and N-Quads inputs land identical dictionary keys. `GraphName`'s own
549/// `Display` renders the default graph as `DEFAULT`, which must never become a
550/// term, hence the explicit match.
551fn graph_token(g: &oxrdf::GraphName) -> Option<String> {
552    match g {
553        oxrdf::GraphName::DefaultGraph => None,
554        other => Some(other.to_string()),
555    }
556}
557
558/// The **RDF 1.2** Turtle/TriG reader — `oxttl` 0.2 with its `rdf-12` feature,
559/// living beside the `oxttl` 0.1 reader above rather than replacing it.
560///
561/// Everything RDF 1.2 adds over RDF 1.1 arrives through this one module:
562/// `<<( s p o )>>` triple terms, `<< s p o >>` reifiers (which expand to
563/// `_:r rdf:reifies <<( s p o )>>` plus the statement that mentioned them),
564/// `{| … |}` annotations, and `"…"@lang--dir` directional literals. None of it
565/// needs a storage change, which is the point: a reifier is a blank node, an
566/// IRI and a term, and rete has stored those since v0.
567///
568/// The one translation is at the term boundary. `oxrdf` 0.3 renders a triple
569/// term as `<<( s p o )>>`; rete's dictionary key is `<<s p o>>`. [`take_term`]
570/// already reads both — that is what made RDF 1.2 N-Quads ingestible in #262 —
571/// so the fold is a re-scan of a token oxttl has already validated.
572#[cfg(feature = "rdf12-turtle")]
573mod rdf12 {
574    use super::{audit_quad, take_term, IngestError, IriAudit, RawQuad};
575
576    /// Fold `oxrdf` 0.3's `<<( s p o )>>` rendering of a triple term into rete's
577    /// stored `<<s p o>>`, recursively (a nested triple term folds with it).
578    /// Every other token — the overwhelming majority, and all of them in a file
579    /// with no triple terms — is returned untouched, not even copied.
580    pub(super) fn fold_triple_term(t: String) -> String {
581        if !t.starts_with("<<(") {
582            return t;
583        }
584        match take_term(&t) {
585            Some((token, rest)) if rest.trim().is_empty() => token,
586            // oxttl only ever hands back a well-formed term, so this is
587            // unreachable; returning the original beats mangling it silently.
588            _ => t,
589        }
590    }
591
592    /// `super::graph_token` for `oxrdf` 0.3's `GraphName` — the same rule, a
593    /// different crate version, which is exactly why it cannot be shared.
594    fn graph_token(g: &oxrdf12::GraphName) -> Option<String> {
595        match g {
596            oxrdf12::GraphName::DefaultGraph => None,
597            other => Some(other.to_string()),
598        }
599    }
600
601    /// Stream one RDF 1.2 Turtle (`"ttl"`) or TriG (`"trig"`) input, auditing
602    /// every quad the way the RDF-star reader does.
603    pub(super) fn stream<R: std::io::BufRead>(
604        reader: R,
605        format: &str,
606        mut audit: Option<&mut IriAudit>,
607        f: &mut dyn FnMut(RawQuad),
608    ) -> Result<(), IngestError> {
609        if format == "ttl" {
610            for r in oxttl12::TurtleParser::new().for_reader(reader) {
611                let t = r.map_err(|e| IngestError::Turtle(e.to_string()))?;
612                let q = (
613                    t.subject.to_string(),
614                    t.predicate.to_string(),
615                    fold_triple_term(t.object.to_string()),
616                    None,
617                );
618                audit_quad(audit.as_deref_mut(), &q.0, &q.1, &q.2, None, || {
619                    "turtle".into()
620                })?;
621                f(q);
622            }
623        } else {
624            for r in oxttl12::TriGParser::new().for_reader(reader) {
625                let q = r.map_err(|e| IngestError::TriG(e.to_string()))?;
626                let q = (
627                    q.subject.to_string(),
628                    q.predicate.to_string(),
629                    fold_triple_term(q.object.to_string()),
630                    graph_token(&q.graph_name),
631                );
632                audit_quad(
633                    audit.as_deref_mut(),
634                    &q.0,
635                    &q.1,
636                    &q.2,
637                    q.3.as_deref(),
638                    || "trig".into(),
639                )?;
640                f(q);
641            }
642        }
643        Ok(())
644    }
645}
646
647/// The stand-in for [`rdf12`] in a build with `rdf12-turtle` off: it refuses by
648/// name. Reading the file as RDF-star instead would answer a question nobody
649/// asked, with a different graph.
650#[cfg(not(feature = "rdf12-turtle"))]
651mod rdf12 {
652    use super::{IngestError, IriAudit, RawQuad};
653
654    pub(super) fn stream<R: std::io::BufRead>(
655        _reader: R,
656        _format: &str,
657        _audit: Option<&mut IriAudit>,
658        _f: &mut dyn FnMut(RawQuad),
659    ) -> Result<(), IngestError> {
660        Err(IngestError::Rdf12TurtleUnavailable)
661    }
662}
663
664/// Parse Turtle into canonical N-Triples-token triples via oxttl, reading
665/// `<< s p o >>` as an RDF-star quoted triple. [`parse_turtle_surface`] chooses.
666pub fn parse_turtle(text: &str) -> Result<Vec<RawTriple>, IngestError> {
667    parse_turtle_surface(text, QuotedTripleSurface::default())
668}
669
670/// [`parse_turtle`] in the given [`QuotedTripleSurface`].
671pub fn parse_turtle_surface(
672    text: &str,
673    surface: QuotedTripleSurface,
674) -> Result<Vec<RawTriple>, IngestError> {
675    Ok(parse_text_surface(text, "ttl", surface)?
676        .into_iter()
677        .map(|(s, p, o, _)| (s, p, o))
678        .collect())
679}
680
681/// Parse TriG into canonical quads via oxttl — Turtle plus named-graph blocks
682/// (`<g> { … }`), the shape most large RDF dumps ship in. Reads `<< s p o >>` as
683/// an RDF-star quoted triple; [`parse_trig_surface`] chooses.
684pub fn parse_trig(text: &str) -> Result<Vec<RawQuad>, IngestError> {
685    parse_trig_surface(text, QuotedTripleSurface::default())
686}
687
688/// [`parse_trig`] in the given [`QuotedTripleSurface`].
689pub fn parse_trig_surface(
690    text: &str,
691    surface: QuotedTripleSurface,
692) -> Result<Vec<RawQuad>, IngestError> {
693    parse_text_surface(text, "trig", surface)
694}
695
696/// The shared Turtle/TriG text path: stream the string through whichever reader
697/// `surface` selects, and — this is the part only the text path can do, because
698/// only it holds the input — say so when an RDF-star parse fails on what is
699/// plainly RDF 1.2.
700///
701/// It does not audit: the Turtle/TriG text callers audit over the finished quads
702/// (see [`parse_statements_audited_surface`]), and auditing here too would count
703/// every IRI twice.
704///
705/// The RDF-star arm calls `oxttl` 0.1 **directly** rather than routing through
706/// [`stream_reader_audited_surface`]. That is not style: this is the only ingest
707/// path the wasm engine reaches, and going through the generic reader would
708/// monomorphize it over `&[u8]` and pull the whole `match format` — every
709/// syntax's reader — into a module that previously linked one Turtle parser.
710/// Measured at +17 KB of wasm for code no browser can run, since the RDF 1.2
711/// reader is not compiled there at all. Keeping the arm direct keeps the
712/// browser artifacts what they were.
713fn parse_text_surface(
714    text: &str,
715    format: &str,
716    surface: QuotedTripleSurface,
717) -> Result<Vec<RawQuad>, IngestError> {
718    let mut out = Vec::new();
719    let r = if surface == QuotedTripleSurface::Rdf12 {
720        rdf12::stream(text.as_bytes(), format, None, &mut |q| out.push(q))
721    } else if format == "ttl" {
722        turtle_star(text.as_bytes(), None, &mut |q| out.push(q))
723    } else {
724        trig_star(text.as_bytes(), None, &mut |q| out.push(q))
725    };
726    match r {
727        Ok(()) => Ok(out),
728        // The default surface met `<<(`: that is an RDF 1.2 file being read as
729        // RDF-star, the one mistake this flag exists to prevent, and the parse
730        // error alone ("unexpected character") would send the reader hunting for
731        // a typo that is not there.
732        Err(IngestError::Turtle(m))
733            if surface == QuotedTripleSurface::RdfStar && looks_like_rdf12(text) =>
734        {
735            Err(IngestError::Turtle(format!("{m}{RDF12_SURFACE_HINT}")))
736        }
737        Err(IngestError::TriG(m))
738            if surface == QuotedTripleSurface::RdfStar && looks_like_rdf12(text) =>
739        {
740            Err(IngestError::TriG(format!("{m}{RDF12_SURFACE_HINT}")))
741        }
742        Err(e) => Err(e),
743    }
744}
745
746/// Parse RDF/XML into canonical N-Triples-token triples via oxrdfxml. This is how
747/// most OWL ontologies ship (`.rdf`/`.owl`/`.xml` with an `rdf:RDF` root) — so rete
748/// ingests them directly, no external conversion. (OWL/XML — the non-RDF functional
749/// XML serialization — is a different language; convert it with owlready2 first.)
750pub fn parse_rdfxml(text: &str) -> Result<Vec<RawTriple>, IngestError> {
751    let mut out = Vec::new();
752    for r in oxrdfxml::RdfXmlParser::new().for_reader(text.as_bytes()) {
753        let t = r.map_err(|e| IngestError::RdfXml(e.to_string()))?;
754        out.push((
755            t.subject.to_string(),
756            t.predicate.to_string(),
757            t.object.to_string(),
758        ));
759    }
760    Ok(out)
761}
762
763/// Parse one text input by format name (`"nt"`, `"nq"`, `"ttl"`, `"trig"`, or
764/// `"rdfxml"`) into quads (triples land in the default graph).
765pub fn parse_statements(text: &str, format: &str) -> Result<Vec<RawQuad>, IngestError> {
766    parse_statements_audited(text, format, None)
767}
768
769/// [`parse_statements`] with an optional [`IriAudit`]. The line-based syntaxes
770/// audit as they parse (so `strict` fails on the offending line); the others are
771/// audited over the parsed quads, which is where `oxttl` has already had its say.
772pub fn parse_statements_audited(
773    text: &str,
774    format: &str,
775    audit: Option<&mut IriAudit>,
776) -> Result<Vec<RawQuad>, IngestError> {
777    parse_statements_audited_surface(text, format, audit, QuotedTripleSurface::default())
778}
779
780/// [`parse_statements_audited`] reading `<< … >>` in the given
781/// [`QuotedTripleSurface`]. This is the entry point `rete build` and
782/// `rete validate` use for everything but a streamed line-based input, and the
783/// one that turns "RDF 1.2 file, RDF-star reader" into an error that names the
784/// flag instead of a bare syntax complaint.
785pub fn parse_statements_audited_surface(
786    text: &str,
787    format: &str,
788    mut audit: Option<&mut IriAudit>,
789    surface: QuotedTripleSurface,
790) -> Result<Vec<RawQuad>, IngestError> {
791    let quads = match format {
792        "nq" => return parse_quads_audited(text, audit),
793        "nt" => {
794            return Ok(parse_audited(text, audit)?
795                .into_iter()
796                .map(|(s, p, o)| (s, p, o, None))
797                .collect())
798        }
799        // Straight to the quad path for both: going via `parse_turtle_surface`
800        // would build a `Vec<RawQuad>`, rebuild it as a `Vec<RawTriple>` to
801        // satisfy that function's signature, and rebuild it as quads again —
802        // three copies of every term String on the biggest input rete takes.
803        "ttl" | "trig" => parse_text_surface(text, format, surface)?,
804        "rdfxml" => parse_rdfxml(text)?
805            .into_iter()
806            .map(|(s, p, o)| (s, p, o, None))
807            .collect(),
808        "jelly" => {
809            return Err(IngestError::Jelly(
810                "Jelly is a binary format, so it cannot come in as text; \
811                 use parse_jelly_bytes or stream_reader"
812                    .into(),
813            ))
814        }
815        other => return Err(IngestError::UnknownFormat(other.to_string())),
816    };
817    for (s, p, o, g) in &quads {
818        audit_quad(audit.as_deref_mut(), s, p, o, g.as_deref(), || {
819            format.to_string()
820        })?;
821    }
822    Ok(quads)
823}
824
825/// Parse a whole Jelly file held in memory (`"jelly"` is binary, so it has no
826/// text entry point). The streaming path is [`stream_reader`] with `"jelly"`;
827/// this is for callers that already hold the bytes, such as the browser engine.
828pub fn parse_jelly_bytes(bytes: &[u8]) -> Result<Vec<RawQuad>, IngestError> {
829    parse_jelly_bytes_audited(bytes, None)
830}
831
832/// [`parse_jelly_bytes`] with an optional [`IriAudit`].
833pub fn parse_jelly_bytes_audited(
834    bytes: &[u8],
835    audit: Option<&mut IriAudit>,
836) -> Result<Vec<RawQuad>, IngestError> {
837    let mut out = Vec::new();
838    jelly::stream(bytes, audit, &mut |q| out.push(q))?;
839    Ok(out)
840}
841
842/// Take one term from the front of `s`, returning `(term, remainder)`.
843pub(crate) fn take_term(s: &str) -> Option<(String, &str)> {
844    let bytes = s.as_bytes();
845    let first = *bytes.first()?;
846    match first {
847        // A quoted triple / triple term, in EITHER surface — the inner terms are
848        // themselves terms (so this recurses; nesting works). `<<` starts with `<`
849        // and an IRI scan would stop at the first inner `>`, so it must be handled
850        // before the plain-IRI case.
851        //   * RDF-star:  `<< subject predicate object >>`
852        //   * RDF 1.2:   `<<( subject predicate object )>>`  (triple term)
853        // Both re-emit the SAME canonical token `<<s p o>>`, so a file written in
854        // either surface — and a query written either way — dedupe and match. This
855        // makes RDF 1.2 N-Triples interoperable with the RDF-star we already store,
856        // with no format change and no dependency swap.
857        b'<' if bytes.get(1) == Some(&b'<') => {
858            // Distinguish `<<(` (RDF 1.2) from `<<` (RDF-star) by the char after `<<`.
859            let (inner, rdf12) = match s[2..].strip_prefix('(') {
860                Some(after) => (after.trim_start(), true),
861                None => (s[2..].trim_start(), false),
862            };
863            let (subj, r) = take_term(inner)?;
864            let (pred, r) = take_term(r.trim_start())?;
865            let (obj, r) = take_term(r.trim_start())?;
866            let r = r.trim_start();
867            let rest = if rdf12 {
868                r.strip_prefix(")>>")?
869            } else {
870                r.strip_prefix(">>")?
871            };
872            // Canonical surface = oxrdf's `Triple` Display: `<<s p o>>` (tight
873            // brackets, single spaces between components) — the same token for both
874            // input surfaces, so N-Triples-star, Turtle-star, and RDF 1.2 triple
875            // terms all resolve to one dictionary entry.
876            Some((format!("<<{subj} {pred} {obj}>>"), rest))
877        }
878        b'<' => {
879            // IRI ref: up to the closing '>'.
880            let end = s.find('>')?;
881            Some((s[..=end].to_string(), &s[end + 1..]))
882        }
883        b'_' => {
884            // Blank node: up to whitespace.
885            let end = s.find(char::is_whitespace).unwrap_or(s.len());
886            Some((s[..end].to_string(), &s[end..]))
887        }
888        b'"' => {
889            // Literal: closing unescaped quote, then optional ^^<dt> or @lang.
890            let mut i = 1;
891            let b = s.as_bytes();
892            while i < b.len() {
893                match b[i] {
894                    b'\\' => i += 2, // skip escaped char
895                    b'"' => break,
896                    _ => i += 1,
897                }
898            }
899            if i >= b.len() {
900                return None; // unterminated
901            }
902            let mut end = i + 1; // past closing quote
903            if s[end..].starts_with("^^<") {
904                let close = s[end..].find('>')? + end;
905                // RDF 1.1: `"x"^^xsd:string` IS `"x"`; keep the one canonical
906                // spelling, so the two never become two dictionary terms and
907                // a plain query constant matches either input.
908                if &s[end + 3..close] == "http://www.w3.org/2001/XMLSchema#string" {
909                    return Some((s[..end].to_string(), &s[close + 1..]));
910                }
911                end = close + 1;
912            } else if s[end..].starts_with('@') {
913                // Language tag: '@' then BCP-47 subtags `[a-zA-Z0-9-]+`. Stop at
914                // the first char that can't be part of a tag — normally the
915                // whitespace before the predicate, but ALSO the `>>` that closes
916                // a quoted triple when this literal is its object (`"x"@en>>`),
917                // where there is no separating whitespace.
918                let mut j = end + 1; // past '@'
919                while j < b.len() && (b[j].is_ascii_alphanumeric() || b[j] == b'-') {
920                    j += 1;
921                }
922                end = j;
923            }
924            Some((s[..end].to_string(), &s[end..]))
925        }
926        _ => None,
927    }
928}
929
930/// Split a canonical quoted-triple token `<<s p o>>` (RDF-star) into its three
931/// component term tokens, or `None` if `t` is not a quoted triple. Reuses
932/// [`take_term`] for term-boundary scanning, so nested quoting parses correctly.
933/// The inverse of the `<<…>>` construction; used by the SUBJECT/PREDICATE/OBJECT
934/// SPARQL-star builtins.
935pub(crate) fn quoted_triple_parts(t: &str) -> Option<(String, String, String)> {
936    let inner = t.strip_prefix("<<")?.strip_suffix(">>")?.trim();
937    let (s, r) = take_term(inner)?;
938    let (p, r) = take_term(r.trim_start())?;
939    let (o, r) = take_term(r.trim_start())?;
940    if !r.trim().is_empty() {
941        return None;
942    }
943    Some((s, p, o))
944}
945
946/// Counts describing an assembled file, for status lines and UIs.
947///
948/// In the [`BuildStats`] a metadata callback receives, `statements` and
949/// `default_triples` are the counts **ingested** — the input multiset, before
950/// the index deduplicates it. In the `BuildStats` an assemble function
951/// *returns*, they are the counts actually **written** (see
952/// [`FinalCounts`]), so a status line reports what the file holds. Use
953/// [`FinalCounts`] rather than these when a payload must agree with the header.
954#[derive(Debug, Clone, Copy)]
955pub struct BuildStats {
956    /// Total statements, across the default graph and every named one.
957    pub statements: usize,
958    /// Statements that landed in the default graph.
959    pub default_triples: usize,
960    /// How many named graphs the input mentioned; one index is written per graph.
961    pub named_graphs: usize,
962    /// Distinct terms in the shared dictionary.
963    pub terms: usize,
964    /// Levels in the community pyramid, or `0` when the build skipped it.
965    pub pyramid_levels: u16,
966    /// Whether the file got a label index. The record says so even when the
967    /// caller (a binding) set no option for it.
968    pub label_index: bool,
969}
970
971/// How a build stores the **named graphs** (the default graph is unaffected).
972#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
973pub enum GraphLayout {
974    /// One permutation index per graph, in the NAMED_GRAPHS section (kind 5):
975    /// the graph IRI as text, then a full container per graph. What every rete
976    /// before 0.4 writes and reads.
977    #[default]
978    PerGraph,
979    /// The **quad index** (kind 10, SPEC.md §6.5): every named-graph quad
980    /// stored once, the graph as a fourth column of graph-leading permutations
981    /// in shared tiles, behind a sorted graph directory; graph names become
982    /// dictionary terms. Far smaller and randomly addressable when graphs are
983    /// many and small. Readers before 0.4.0 refuse these files (loudly — see
984    /// `docs/compatibility.md`). Needs the 3- or 6-permutation set.
985    Quad,
986}
987
988/// One pass's named-graph quads, in the layout being built.
989enum NamedQuads {
990    PerGraph(std::collections::BTreeMap<String, Vec<(u32, u32, u32)>>),
991    Quad {
992        /// `(graph subject id, s, p, o)`.
993        quads: Vec<[u32; 4]>,
994        /// Distinct graphs seen (a bitset over subject ids).
995        seen: Vec<u64>,
996        graphs: usize,
997        /// The last graph name and its id: N-Quads dumps keep a graph's
998        /// statements together, so most quads skip the dictionary lookup.
999        last: Option<(String, u32)>,
1000    },
1001}
1002
1003impl NamedQuads {
1004    fn new(layout: GraphLayout, dict: &Dictionary) -> Self {
1005        match layout {
1006            GraphLayout::PerGraph => NamedQuads::PerGraph(Default::default()),
1007            GraphLayout::Quad => NamedQuads::Quad {
1008                quads: Vec::new(),
1009                seen: vec![
1010                    0u64;
1011                    (dict.shared_count() as usize + dict.subject_only_count() as usize) / 64
1012                        + 2
1013                ],
1014                graphs: 0,
1015                last: None,
1016            },
1017        }
1018    }
1019
1020    fn push(&mut self, dict: &Dictionary, graph: String, t: (u32, u32, u32)) {
1021        match self {
1022            NamedQuads::PerGraph(m) => m.entry(graph).or_default().push(t),
1023            NamedQuads::Quad {
1024                quads,
1025                seen,
1026                graphs,
1027                last,
1028            } => {
1029                let g = match last {
1030                    Some((name, id)) if *name == graph => *id,
1031                    _ => {
1032                        let id = dict.subject_id(&graph).expect("observed graph name");
1033                        *last = Some((graph, id));
1034                        id
1035                    }
1036                };
1037                let (w, b) = (g as usize / 64, g % 64);
1038                if let Some(word) = seen.get_mut(w) {
1039                    if *word & (1 << b) == 0 {
1040                        *word |= 1 << b;
1041                        *graphs += 1;
1042                    }
1043                }
1044                quads.push([g, t.0, t.1, t.2]);
1045            }
1046        }
1047    }
1048
1049    fn graph_count(&self) -> usize {
1050        match self {
1051            NamedQuads::PerGraph(m) => m.len(),
1052            NamedQuads::Quad { graphs, .. } => *graphs,
1053        }
1054    }
1055
1056    fn quad_count(&self) -> usize {
1057        match self {
1058            NamedQuads::PerGraph(m) => m.values().map(Vec::len).sum(),
1059            NamedQuads::Quad { quads, .. } => quads.len(),
1060        }
1061    }
1062}
1063
1064/// The **deduplicated** counts of the file being written — what the header
1065/// records, and therefore what any embedded metadata has to report.
1066///
1067/// The input multiset is not the file: every permutation index sorts and
1068/// deduplicates, so a harvest that pages with overlapping windows writes fewer
1069/// quads than it ingested. These counts are known only once the indexes exist,
1070/// which is why a payload that carries them is derived in two stages (see
1071/// [`IntoMetadata`]).
1072#[derive(Debug, Clone, Copy)]
1073pub struct FinalCounts {
1074    /// Unique triples in the default graph.
1075    pub default_triples: u64,
1076    /// Unique quads across the default graph and every named graph — exactly
1077    /// the header's `quad_count`.
1078    pub quads: u64,
1079}
1080
1081/// What a build's metadata callback may return.
1082///
1083/// A `Vec<u8>` is the payload verbatim — the common case (no metadata, or a
1084/// payload that carries no counts). [`DeferredMetadata`] is a payload that
1085/// *does* carry counts: the callback derives everything it needs while the
1086/// source quads are still resident, and returns a finalizer that the writer
1087/// calls with the [`FinalCounts`], which only exist once the indexes have
1088/// deduplicated the input. Without that second stage a card can only report the
1089/// ingested count, which over-reports any input containing duplicates.
1090pub trait IntoMetadata {
1091    /// Produce the metadata section payload for a file with these counts.
1092    fn into_metadata(self, counts: FinalCounts) -> Vec<u8>;
1093}
1094
1095impl IntoMetadata for Vec<u8> {
1096    fn into_metadata(self, _counts: FinalCounts) -> Vec<u8> {
1097        self
1098    }
1099}
1100
1101/// A metadata payload whose final form needs the written (deduplicated) counts.
1102/// See [`IntoMetadata`].
1103pub struct DeferredMetadata(Box<dyn FnOnce(FinalCounts) -> Vec<u8>>);
1104
1105impl DeferredMetadata {
1106    /// Defer `f` until the indexes are built and the true counts are known.
1107    pub fn new(f: impl FnOnce(FinalCounts) -> Vec<u8> + 'static) -> Self {
1108        DeferredMetadata(Box::new(f))
1109    }
1110
1111    /// No metadata section — byte-identical to a metadata-free build. Lets one
1112    /// `match` arm opt out while the other defers, without a type mismatch.
1113    pub fn none() -> Self {
1114        DeferredMetadata::new(|_| Vec::new())
1115    }
1116}
1117
1118impl IntoMetadata for DeferredMetadata {
1119    fn into_metadata(self, counts: FinalCounts) -> Vec<u8> {
1120        (self.0)(counts)
1121    }
1122}
1123
1124/// [`assemble_dataset`] for the writers that finish as format `0x06` (the
1125/// bindings and the browser): named graphs go in the quad index, the only
1126/// layout `0x06` has. Everything else is `assemble_dataset`'s.
1127pub fn assemble_dataset_quad(quads: Vec<RawQuad>, metadata: &[u8]) -> (Vec<u8>, BuildStats) {
1128    let blob = metadata.to_vec();
1129    assemble_dataset_with_quad(quads, move |_, _| blob)
1130}
1131
1132/// [`assemble_dataset_with`] with the named graphs in the quad index.
1133pub fn assemble_dataset_with_quad<M: IntoMetadata>(
1134    quads: Vec<RawQuad>,
1135    metadata: impl FnOnce(&BuildStats, &[RawQuad]) -> M,
1136) -> (Vec<u8>, BuildStats) {
1137    assemble_dataset_with_opts_algo_quad(quads, true, false, None, PyramidAlgo::Louvain, metadata)
1138}
1139
1140/// [`assemble_dataset_with_opts_algo`] for the `0x06` writers, with **the layout
1141/// `rete build --label-index` writes**: the named graphs in the quad index, the
1142/// pyramid block carrying only the community material, the SCHEMA section and
1143/// the label index. The browser engine and the Python, R and Java clients all
1144/// assemble through this, so one input and one set of options is one content
1145/// hash whichever front end built it (the browser and Java, having no zstd
1146/// encoder, write their sections uncompressed and so differ in bytes). The label
1147/// index is on because label prefix search is what a graph built in memory is
1148/// opened to do (the playground, the MCP server), and the legacy label block it
1149/// replaces is gone from the community-only pyramid block.
1150pub fn assemble_dataset_with_opts_algo_quad<M: IntoMetadata>(
1151    quads: Vec<RawQuad>,
1152    with_pyramid: bool,
1153    with_text_index: impl crate::text_index::IntoTextIndex,
1154    type_override: Option<&str>,
1155    algo: PyramidAlgo,
1156    metadata: impl FnOnce(&BuildStats, &[RawQuad]) -> M,
1157) -> (Vec<u8>, BuildStats) {
1158    let text_index = crate::text_index::IntoTextIndex::into_text_index(with_text_index);
1159    let opts = AssembleOptions {
1160        with_pyramid,
1161        with_text_index: text_index.is_some(),
1162        text_index_scope: text_index.as_ref(),
1163        type_override,
1164        algo,
1165        graph_layout: GraphLayout::Quad,
1166        label_index: true,
1167        ..AssembleOptions::sectioned()
1168    };
1169    assemble_dataset_opts(quads, &opts, metadata)
1170}
1171
1172/// Assemble a complete `.rete` file image from parsed quads: one shared
1173/// dictionary, the default-graph index, one index per named graph, and the
1174/// community pyramid. `metadata` is the opaque metadata-section payload (the
1175/// CLI puts a JSON Dataset Card there); pass `&[]` for none — that is
1176/// byte-identical to a metadata-free build.
1177pub fn assemble_dataset(quads: Vec<RawQuad>, metadata: &[u8]) -> (Vec<u8>, BuildStats) {
1178    let blob = metadata.to_vec();
1179    assemble_dataset_with(quads, move |_, _| blob)
1180}
1181
1182/// Like [`assemble_dataset`], but the metadata payload is derived from the
1183/// [`BuildStats`] and the source quads while they are still resident — for
1184/// metadata that describes the graph (the Dataset Card). Returning an empty
1185/// `Vec` is byte-identical to a metadata-free build; return a
1186/// [`DeferredMetadata`] for a payload that must also carry the file's
1187/// deduplicated counts, which exist only once the indexes are built.
1188pub fn assemble_dataset_with<M: IntoMetadata>(
1189    quads: Vec<RawQuad>,
1190    metadata: impl FnOnce(&BuildStats, &[RawQuad]) -> M,
1191) -> (Vec<u8>, BuildStats) {
1192    assemble_dataset_with_opts(quads, true, false, None, metadata)
1193}
1194
1195/// Like [`assemble_dataset_with`], but `with_pyramid = false` skips the Louvain
1196/// community pyramid entirely — no pyramid section is written (header length 0).
1197/// SPARQL / SHACL / triple / reachability queries don't use the pyramid, so a
1198/// pyramid-less file is fully queryable and markedly smaller (the pyramid is the
1199/// largest section on highly-connected graphs). Only the community / summary /
1200/// progressive paths need it.
1201pub fn assemble_dataset_with_opts<M: IntoMetadata>(
1202    quads: Vec<RawQuad>,
1203    with_pyramid: bool,
1204    with_text_index: impl crate::text_index::IntoTextIndex,
1205    type_override: Option<&str>,
1206    metadata: impl FnOnce(&BuildStats, &[RawQuad]) -> M,
1207) -> (Vec<u8>, BuildStats) {
1208    assemble_dataset_with_opts_algo(
1209        quads,
1210        with_pyramid,
1211        with_text_index,
1212        type_override,
1213        PyramidAlgo::Louvain,
1214        metadata,
1215    )
1216}
1217
1218/// Like [`assemble_dataset_with_opts`], but selects the community [`PyramidAlgo`]
1219/// (the in-memory build path for `rete build --pyramid-algo …`).
1220#[allow(clippy::too_many_arguments)]
1221pub fn assemble_dataset_with_opts_algo<M: IntoMetadata>(
1222    quads: Vec<RawQuad>,
1223    with_pyramid: bool,
1224    with_text_index: impl crate::text_index::IntoTextIndex,
1225    type_override: Option<&str>,
1226    algo: PyramidAlgo,
1227    metadata: impl FnOnce(&BuildStats, &[RawQuad]) -> M,
1228) -> (Vec<u8>, BuildStats) {
1229    assemble_dataset_with_perms(
1230        quads,
1231        with_pyramid,
1232        with_text_index,
1233        type_override,
1234        algo,
1235        crate::index::PermSet::ALL,
1236        metadata,
1237    )
1238}
1239
1240/// Like [`assemble_dataset_with_opts_algo`], but writes only the permutations in
1241/// `perms` (`rete build --permutations 3`). Every other byte of the file is
1242/// unchanged; [`crate::index::PermSet::ALL`] reproduces the default build
1243/// exactly, down to the header byte.
1244#[allow(clippy::too_many_arguments)]
1245pub fn assemble_dataset_with_perms<M: IntoMetadata>(
1246    quads: Vec<RawQuad>,
1247    with_pyramid: bool,
1248    with_text_index: impl crate::text_index::IntoTextIndex,
1249    type_override: Option<&str>,
1250    algo: PyramidAlgo,
1251    perms: crate::index::PermSet,
1252    metadata: impl FnOnce(&BuildStats, &[RawQuad]) -> M,
1253) -> (Vec<u8>, BuildStats) {
1254    assemble_dataset_with_layout(
1255        quads,
1256        with_pyramid,
1257        with_text_index,
1258        type_override,
1259        algo,
1260        perms,
1261        GraphLayout::PerGraph,
1262        metadata,
1263    )
1264}
1265
1266/// Like [`assemble_dataset_with_perms`], storing the named graphs in `graphs`
1267/// layout. [`GraphLayout::PerGraph`] reproduces it exactly.
1268///
1269/// # Panics
1270///
1271/// With [`GraphLayout::Quad`] and a permutation set other than
1272/// [`PermSet::CORE`](crate::index::PermSet::CORE) or
1273/// [`PermSet::ALL`](crate::index::PermSet::ALL), when the input has named graphs.
1274#[allow(clippy::too_many_arguments)]
1275pub fn assemble_dataset_with_layout<M: IntoMetadata>(
1276    quads: Vec<RawQuad>,
1277    with_pyramid: bool,
1278    with_text_index: impl crate::text_index::IntoTextIndex,
1279    type_override: Option<&str>,
1280    algo: PyramidAlgo,
1281    perms: crate::index::PermSet,
1282    graphs: GraphLayout,
1283    metadata: impl FnOnce(&BuildStats, &[RawQuad]) -> M,
1284) -> (Vec<u8>, BuildStats) {
1285    let text_index = crate::text_index::IntoTextIndex::into_text_index(with_text_index);
1286    let opts = AssembleOptions {
1287        with_pyramid,
1288        with_text_index: text_index.is_some(),
1289        text_index_scope: text_index.as_ref(),
1290        type_override,
1291        algo,
1292        perms,
1293        graph_layout: graphs,
1294        ..AssembleOptions::legacy()
1295    };
1296    assemble_dataset_opts(quads, &opts, metadata)
1297}
1298
1299/// Everything an in-memory build decides besides its input. The fields the
1300/// older `assemble_*` entry points take positionally, plus the sections added
1301/// after the generation-1 freeze.
1302#[derive(Debug, Clone, Copy)]
1303pub struct AssembleOptions<'a> {
1304    /// Build the community pyramid (kind 4).
1305    pub with_pyramid: bool,
1306    /// Build the full-text index (kind 6, or kind 11 when its scope leaves
1307    /// out some predicates).
1308    pub with_text_index: bool,
1309    /// The scope of that index (`--text-index-graphs`, `--text-index-include`,
1310    /// `--text-index-exclude`); `None` is the default scope, the only one
1311    /// before 0.4.0: the default graph, every predicate.
1312    pub text_index_scope: Option<&'a crate::text_index::TextIndexScope>,
1313    /// The type predicate for the schema (e.g. `wdt:P31`); `rdf:type` if `None`.
1314    pub type_override: Option<&'a str>,
1315    /// The community algorithm of the pyramid.
1316    pub algo: PyramidAlgo,
1317    /// Which index permutations to write.
1318    pub perms: crate::index::PermSet,
1319    /// How the named graphs are stored: one index per graph (kind 5) or the
1320    /// quad index (kind 10, `rete build --named-graphs quad`).
1321    pub graph_layout: GraphLayout,
1322    /// What the pyramid-meta carries besides the summary.
1323    pub pyramid_layout: crate::file::PyramidLayout,
1324    /// Write the LABEL_INDEX section (kind 8, `rete build --label-index`).
1325    pub label_index: bool,
1326    /// Write the SCHEMA section (kind 9; `rete build` writes it unless
1327    /// `--no-schema`).
1328    pub schema: bool,
1329}
1330
1331impl AssembleOptions<'_> {
1332    /// What every `assemble_*` entry point wrote before kinds 8 and 9 existed:
1333    /// a Louvain pyramid with the combined pyramid-meta, no text index, all six
1334    /// permutations, one index per named graph, no LABEL_INDEX or SCHEMA
1335    /// section.
1336    pub fn legacy() -> Self {
1337        AssembleOptions {
1338            with_pyramid: true,
1339            with_text_index: false,
1340            text_index_scope: None,
1341            type_override: None,
1342            algo: PyramidAlgo::Louvain,
1343            perms: crate::index::PermSet::ALL,
1344            graph_layout: GraphLayout::PerGraph,
1345            pyramid_layout: crate::file::PyramidLayout::Combined,
1346            label_index: false,
1347            schema: false,
1348        }
1349    }
1350
1351    /// What `rete build` writes: the pyramid-meta carries only the community
1352    /// material, the schema has its own section, and the label index is
1353    /// opt-in.
1354    pub fn sectioned() -> Self {
1355        AssembleOptions {
1356            pyramid_layout: crate::file::PyramidLayout::CommunityOnly,
1357            schema: true,
1358            ..Self::legacy()
1359        }
1360    }
1361}
1362
1363/// Assemble a file image from parsed quads under explicit [`AssembleOptions`].
1364///
1365/// # Panics
1366///
1367/// With [`GraphLayout::Quad`] and a permutation set other than
1368/// [`PermSet::CORE`](crate::index::PermSet::CORE) or
1369/// [`PermSet::ALL`](crate::index::PermSet::ALL), when the input has named graphs.
1370pub fn assemble_dataset_opts<M: IntoMetadata>(
1371    quads: Vec<RawQuad>,
1372    opts: &AssembleOptions<'_>,
1373    metadata: impl FnOnce(&BuildStats, &[RawQuad]) -> M,
1374) -> (Vec<u8>, BuildStats) {
1375    use crate::build_progress::{self as progress, Stat, Ticker, Unit};
1376    use crate::build_trace::Phase;
1377    let graphs = opts.graph_layout;
1378    let n = Some(quads.len() as u64);
1379    progress::begin(Phase::Dictionary, n, Unit::Statements);
1380    let mut ticker = Ticker::new();
1381    let mut db = DictionaryBuilder::new();
1382    for (s, p, o, g) in &quads {
1383        db.observe(s, p, o);
1384        if let (GraphLayout::Quad, Some(g)) = (graphs, g) {
1385            db.observe_graph(g);
1386        }
1387        ticker.tick(1);
1388    }
1389    ticker.flush();
1390    let dict = db.build();
1391    crate::build_trace::lap(Phase::Dictionary);
1392    progress::set_stat(Stat::Terms, dict.term_count().into());
1393
1394    progress::begin(Phase::Encode, n, Unit::Statements);
1395    let mut default_triples = Vec::new();
1396    let mut named = NamedQuads::new(graphs, &dict);
1397    for (s, p, o, g) in &quads {
1398        let t = dict.encode(s, p, o).expect("observed term");
1399        match g {
1400            None => default_triples.push(t),
1401            Some(graph) => named.push(&dict, graph.clone(), t),
1402        }
1403        ticker.tick(1);
1404    }
1405    drop(ticker);
1406    crate::build_trace::lap(Phase::Encode);
1407
1408    // Derive the metadata blob (the Dataset Card) from the raw quads NOW, while
1409    // they are resident — then DROP them before the memory-heavy pyramid + index
1410    // phases. On a big build the string quads are the largest working set (every
1411    // term an owned String, heavily duplicated) and are fully redundant with the
1412    // dictionary + id-triples once encoded, so freeing them here cuts peak RAM by
1413    // their whole size. `pyramid_levels` is not known yet (0 in the callback);
1414    // it is filled into the returned `stats` by `finish_assembly`, and no metadata
1415    // callback depends on it (the card derives from the quads + term/graph counts).
1416    let stats = BuildStats {
1417        statements: quads.len(),
1418        default_triples: default_triples.len(),
1419        named_graphs: named.graph_count(),
1420        terms: dict.term_count() as usize,
1421        pyramid_levels: 0,
1422        label_index: opts.label_index,
1423    };
1424    let pending = metadata(&stats, &quads);
1425    crate::build_trace::lap(crate::build_trace::Phase::CardDerive);
1426    drop(quads);
1427
1428    finish_assembly(dict, default_triples, named, opts, pending, stats)
1429}
1430
1431/// **Two-pass, low-RAM** assembly: build a `.rete` by **streaming** the input(s)
1432/// twice instead of holding every parsed quad in memory. `stream` is invoked
1433/// **twice** and MUST replay the exact same quads in the same order each time —
1434/// pass 1 observes every term into the dictionary; pass 2 encodes them to
1435/// id-triples. The raw string quads (every term an owned String, heavily
1436/// duplicated — by far the largest working set on a big graph) are **never
1437/// collected**, so peak RAM is bounded by the dictionary + id-triples + index
1438/// rather than the string-quad multiset. Output is **byte-identical** to
1439/// [`assemble_dataset_with_opts`] on the same quads (same dictionary, same
1440/// id-triples in file order, same downstream pipeline).
1441///
1442/// The metadata callback derives the Dataset Card from the built dictionary +
1443/// default-graph id-triples (resolving terms through the dictionary), since the
1444/// raw quads were never retained. `stream` propagates parse/IO errors.
1445pub fn assemble_dataset_streaming<S, M: IntoMetadata>(
1446    stream: S,
1447    with_pyramid: bool,
1448    with_text_index: impl crate::text_index::IntoTextIndex,
1449    type_override: Option<&str>,
1450    metadata: impl FnOnce(&BuildStats, &Dictionary, &[(u32, u32, u32)]) -> M,
1451) -> Result<(Vec<u8>, BuildStats), IngestError>
1452where
1453    S: FnMut(&mut dyn FnMut(RawQuad)) -> Result<(), IngestError>,
1454{
1455    assemble_dataset_streaming_algo(
1456        stream,
1457        with_pyramid,
1458        with_text_index,
1459        type_override,
1460        PyramidAlgo::Louvain,
1461        metadata,
1462    )
1463}
1464
1465/// Like [`assemble_dataset_streaming`], but selects the community [`PyramidAlgo`]
1466/// (the streaming, low-RAM build path for `rete build --pyramid-algo …`).
1467#[allow(clippy::too_many_arguments)]
1468pub fn assemble_dataset_streaming_algo<S, M: IntoMetadata>(
1469    stream: S,
1470    with_pyramid: bool,
1471    with_text_index: impl crate::text_index::IntoTextIndex,
1472    type_override: Option<&str>,
1473    algo: PyramidAlgo,
1474    metadata: impl FnOnce(&BuildStats, &Dictionary, &[(u32, u32, u32)]) -> M,
1475) -> Result<(Vec<u8>, BuildStats), IngestError>
1476where
1477    S: FnMut(&mut dyn FnMut(RawQuad)) -> Result<(), IngestError>,
1478{
1479    assemble_dataset_streaming_with_perms(
1480        stream,
1481        with_pyramid,
1482        with_text_index,
1483        type_override,
1484        algo,
1485        crate::index::PermSet::ALL,
1486        metadata,
1487    )
1488}
1489
1490/// Like [`assemble_dataset_streaming_algo`], but writes only the permutations in
1491/// `perms` — the two-pass low-RAM twin of [`assemble_dataset_with_perms`].
1492#[allow(clippy::too_many_arguments)]
1493pub fn assemble_dataset_streaming_with_perms<S, M: IntoMetadata>(
1494    stream: S,
1495    with_pyramid: bool,
1496    with_text_index: impl crate::text_index::IntoTextIndex,
1497    type_override: Option<&str>,
1498    algo: PyramidAlgo,
1499    perms: crate::index::PermSet,
1500    metadata: impl FnOnce(&BuildStats, &Dictionary, &[(u32, u32, u32)]) -> M,
1501) -> Result<(Vec<u8>, BuildStats), IngestError>
1502where
1503    S: FnMut(&mut dyn FnMut(RawQuad)) -> Result<(), IngestError>,
1504{
1505    assemble_dataset_streaming_with_layout(
1506        stream,
1507        with_pyramid,
1508        with_text_index,
1509        type_override,
1510        algo,
1511        perms,
1512        GraphLayout::PerGraph,
1513        metadata,
1514    )
1515}
1516
1517/// Like [`assemble_dataset_streaming_with_perms`], storing the named graphs in
1518/// `graphs` layout — the two-pass low-RAM twin of
1519/// [`assemble_dataset_with_layout`]. [`GraphLayout::Quad`] with a permutation
1520/// set other than the 3- or 6-permutation one is an error.
1521#[allow(clippy::too_many_arguments)]
1522pub fn assemble_dataset_streaming_with_layout<S, M: IntoMetadata>(
1523    stream: S,
1524    with_pyramid: bool,
1525    with_text_index: impl crate::text_index::IntoTextIndex,
1526    type_override: Option<&str>,
1527    algo: PyramidAlgo,
1528    perms: crate::index::PermSet,
1529    graphs: GraphLayout,
1530    metadata: impl FnOnce(&BuildStats, &Dictionary, &[(u32, u32, u32)]) -> M,
1531) -> Result<(Vec<u8>, BuildStats), IngestError>
1532where
1533    S: FnMut(&mut dyn FnMut(RawQuad)) -> Result<(), IngestError>,
1534{
1535    let text_index = crate::text_index::IntoTextIndex::into_text_index(with_text_index);
1536    let opts = AssembleOptions {
1537        with_pyramid,
1538        with_text_index: text_index.is_some(),
1539        text_index_scope: text_index.as_ref(),
1540        type_override,
1541        algo,
1542        perms,
1543        graph_layout: graphs,
1544        ..AssembleOptions::legacy()
1545    };
1546    assemble_dataset_streaming_opts(stream, &opts, metadata)
1547}
1548
1549/// The two-pass low-RAM twin of [`assemble_dataset_opts`]. A quad-index layout
1550/// with a permutation set other than the 3- or 6-permutation one is an error.
1551pub fn assemble_dataset_streaming_opts<S, M: IntoMetadata>(
1552    mut stream: S,
1553    opts: &AssembleOptions<'_>,
1554    metadata: impl FnOnce(&BuildStats, &Dictionary, &[(u32, u32, u32)]) -> M,
1555) -> Result<(Vec<u8>, BuildStats), IngestError>
1556where
1557    S: FnMut(&mut dyn FnMut(RawQuad)) -> Result<(), IngestError>,
1558{
1559    let graphs = opts.graph_layout;
1560    if graphs == GraphLayout::Quad && crate::quad::check_perms(opts.perms).is_err() {
1561        return Err(IngestError::Io(
1562            "the quad index needs the 3- or 6-permutation set".into(),
1563        ));
1564    }
1565    // Pass 1: observe every term (the dictionary dedups; the string quads are
1566    // freed line by line and never collected). With the quad index, graph
1567    // names are terms too. The stream sets each pass's progress total (the
1568    // input bytes it will read).
1569    use crate::build_progress::{self as progress, Stat, Unit};
1570    use crate::build_trace::Phase;
1571    progress::begin(Phase::ParseDictionary, None, Unit::None);
1572    let mut db = DictionaryBuilder::new();
1573    stream(&mut |(s, p, o, g)| {
1574        db.observe(&s, &p, &o);
1575        if let (GraphLayout::Quad, Some(g)) = (graphs, g) {
1576            db.observe_graph(&g);
1577        }
1578    })?;
1579    let dict = db.build();
1580    crate::build_trace::lap(Phase::ParseDictionary);
1581    progress::set_stat(Stat::Terms, dict.term_count().into());
1582
1583    // Pass 2: encode each quad to an id-triple, bucketing named graphs.
1584    progress::begin(Phase::ParseEncode, None, Unit::None);
1585    let mut default_triples: Vec<(u32, u32, u32)> = Vec::new();
1586    let mut named = NamedQuads::new(graphs, &dict);
1587    stream(&mut |(s, p, o, g)| {
1588        let t = dict.encode(&s, &p, &o).expect("observed term");
1589        match g {
1590            None => default_triples.push(t),
1591            Some(graph) => named.push(&dict, graph, t),
1592        }
1593    })?;
1594    crate::build_trace::lap(crate::build_trace::Phase::ParseEncode);
1595
1596    let statements = default_triples.len() + named.quad_count();
1597    let stats = BuildStats {
1598        statements,
1599        default_triples: default_triples.len(),
1600        named_graphs: named.graph_count(),
1601        terms: dict.term_count() as usize,
1602        pyramid_levels: 0,
1603        label_index: opts.label_index,
1604    };
1605    let pending = metadata(&stats, &dict, &default_triples);
1606    crate::build_trace::lap(crate::build_trace::Phase::CardDerive);
1607    Ok(finish_assembly(
1608        dict,
1609        default_triples,
1610        named,
1611        opts,
1612        pending,
1613        stats,
1614    ))
1615}
1616
1617/// The shared tail of every build path: from a finished dictionary + encoded
1618/// id-triples (default graph + named), build the community pyramid, the optional
1619/// full-text index, the permutation indexes, and serialize the file image. Takes
1620/// the id-triples **by value** so they are freed as the permutations consume them.
1621fn finish_assembly<M: IntoMetadata>(
1622    dict: Dictionary,
1623    mut default_triples: Vec<(u32, u32, u32)>,
1624    named: NamedQuads,
1625    opts: &AssembleOptions<'_>,
1626    pending: M,
1627    mut stats: BuildStats,
1628) -> (Vec<u8>, BuildStats) {
1629    use crate::build_progress::{self as progress, Unit};
1630    use crate::build_trace::Phase;
1631    let has_named = named.graph_count() > 0;
1632    let perms = opts.perms;
1633
1634    // The pyramid and the full-text index are built FIRST, while the dictionary and
1635    // the default id-triples are both still resident — both need the two together.
1636    // The pyramid sees the triples in input order (Louvain is order-sensitive).
1637    let (meta, levels) = if opts.with_pyramid {
1638        progress::begin(Phase::Pyramid, None, Unit::None);
1639        crate::file::build_pyramid_meta_layout(
1640            &dict,
1641            &default_triples,
1642            DEFAULT_TILE_BUDGET,
1643            opts.type_override,
1644            opts.algo,
1645            opts.pyramid_layout,
1646        )
1647    } else {
1648        (Vec::new(), 0)
1649    };
1650    stats.pyramid_levels = levels;
1651    if opts.with_pyramid {
1652        crate::build_trace::lap(crate::build_trace::Phase::Pyramid);
1653    }
1654
1655    // The full-text index over the literals of the requested scope: the
1656    // default graph's (the only scope before 0.4.0), optionally every named
1657    // graph's too, optionally only some predicates'.
1658    let with_text_index = opts
1659        .with_text_index
1660        .then(|| opts.text_index_scope.cloned().unwrap_or_default());
1661    // The named graphs' triples, for an index whose scope covers them: per
1662    // graph, or (in the quad layout) the quads' `(s, p, o)`, copied only when
1663    // the scope needs them; the index does not care which graph a word came
1664    // from.
1665    let quad_spo: Vec<(u32, u32, u32)>;
1666    let named_triples: Vec<&[(u32, u32, u32)]> = match &named {
1667        NamedQuads::PerGraph(m) => m.values().map(Vec::as_slice).collect(),
1668        NamedQuads::Quad { quads, .. } => {
1669            quad_spo = if with_text_index
1670                .as_ref()
1671                .is_some_and(crate::text_index::TextIndexScope::covers_named_graphs)
1672            {
1673                quads.iter().map(|q| (q[1], q[2], q[3])).collect()
1674            } else {
1675                Vec::new()
1676            };
1677            vec![quad_spo.as_slice()]
1678        }
1679    };
1680    if with_text_index.is_some() {
1681        progress::begin(Phase::TextIndex, None, Unit::None);
1682    }
1683    let text_index = match &with_text_index {
1684        Some(scope) => crate::text_index::build_section(
1685            &dict,
1686            scope,
1687            &default_triples,
1688            named_triples.into_iter(),
1689            crate::file::writer_codec(),
1690        ),
1691        None => Vec::new(),
1692    };
1693    if with_text_index.is_some() {
1694        crate::build_trace::lap(crate::build_trace::Phase::TextIndex);
1695    }
1696    let text_index_partial = with_text_index
1697        .as_ref()
1698        .is_some_and(crate::text_index::TextIndexScope::restricts_predicates);
1699    let text_index_named = with_text_index
1700        .as_ref()
1701        .is_some_and(crate::text_index::TextIndexScope::covers_named_graphs);
1702
1703    // The LABEL_INDEX and SCHEMA sections are computed over the DEDUPLICATED
1704    // default graph in SPO order (and the schema's second pass in POS order),
1705    // the order every index build re-sorts into anyway — which is what lets
1706    // the memory-bounded build, reading its sorted permutation streams, write
1707    // the very same bytes. Both need the dictionary, so they run before it is
1708    // dropped. Without either section the triples stay in input order.
1709    let mut label_section = Vec::new();
1710    let mut schema_section = Vec::new();
1711    if opts.label_index || opts.schema {
1712        progress::begin(Phase::LabelsSchema, None, Unit::None);
1713        crate::schema_section::sort_dedup_spo(&mut default_triples);
1714        if opts.label_index {
1715            let entries = crate::label_index::entries_from_sorted_triples(&dict, &default_triples);
1716            label_section =
1717                crate::label_index::encode_section(&entries, crate::file::writer_codec());
1718        }
1719        if opts.schema {
1720            schema_section = crate::schema_section::compute_in_memory(
1721                &dict,
1722                &mut default_triples,
1723                opts.type_override,
1724            )
1725            .encode();
1726        }
1727        crate::build_trace::lap(crate::build_trace::Phase::LabelsSchema);
1728    }
1729
1730    // From here the dictionary is only needed for its own serialized bytes. Encode
1731    // it, capture the header term count, then DROP it before building the
1732    // permutation indexes — which work purely on id-triples and never touch the
1733    // dictionary. On a large graph this frees the single biggest resident structure
1734    // (the dictionary) right when the index sort needs the headroom. The output is
1735    // byte-for-byte identical to serializing the dictionary inline.
1736    let codec = crate::file::writer_codec();
1737    progress::begin(Phase::DictionarySection, None, Unit::None);
1738    let dict_container = crate::file::encode_dict_container(&dict, codec);
1739    let term_count = dict.term_count() as u64;
1740    let has_quoted_triples = dict.has_quoted_triples();
1741    drop(dict);
1742    crate::build_trace::lap(crate::build_trace::Phase::DictionarySection);
1743
1744    // Build each graph's six permutations one-at-a-time on a large graph (a single
1745    // permuted copy resident at a time, each sort still parallel) instead of all
1746    // six concurrently; below the threshold the faster all-permutations-parallel
1747    // build is used (its 6x transient copy is negligible there). `build_seq` is
1748    // byte-identical to `build`.
1749    // Every graph's triples, once per permutation (the quad index's quads
1750    // once, as it finishes).
1751    let (per_graph, quad_rows): (u64, u64) = match &named {
1752        NamedQuads::PerGraph(m) => (m.values().map(|t| t.len() as u64).sum(), 0),
1753        NamedQuads::Quad { quads, .. } => (0, quads.len() as u64),
1754    };
1755    let rows = (default_triples.len() as u64 + per_graph) * perms.len() as u64 + quad_rows;
1756    progress::begin(Phase::IndexPermutations, Some(rows), Unit::Records);
1757    let build_index = |triples: Vec<(u32, u32, u32)>| -> crate::GraphIndex {
1758        let n = triples.len();
1759        let b = GraphIndexBuilder::from_triples(triples)
1760            .with_perms(perms)
1761            .with_progress();
1762        if n > LOWMEM_TRIPLE_THRESHOLD {
1763            b.build_seq()
1764        } else {
1765            b.build()
1766        }
1767    };
1768    let def = build_index(default_triples);
1769    // The named graphs: one index per graph, or one quad index for all of them
1770    // (no per-graph index is ever built on that path).
1771    let (named_indexes, quad_index): (Vec<(String, crate::GraphIndex)>, _) = match named {
1772        NamedQuads::PerGraph(named) => (
1773            named
1774                .into_iter()
1775                .map(|(g, ts)| (g, build_index(ts)))
1776                .collect(),
1777            None,
1778        ),
1779        NamedQuads::Quad { quads, .. } => {
1780            let n = quads.len() as u64;
1781            let q = crate::quad::encode_quad_index(quads, perms, codec)
1782                .expect("the quad index needs the 3- or 6-permutation set");
1783            progress::advance(n);
1784            (Vec::new(), Some(q))
1785        }
1786    };
1787    let named_quads = match &quad_index {
1788        Some(q) => q.quad_count,
1789        None => named_indexes
1790            .iter()
1791            .map(|(_, ix)| ix.triple_count() as u64)
1792            .sum::<u64>(),
1793    };
1794    crate::build_trace::lap(crate::build_trace::Phase::IndexPermutations);
1795
1796    // Only now is the DEDUPLICATED size of the file known: every index sorts and
1797    // dedups, so an input containing duplicate statements (a harvest that pages
1798    // with overlapping windows, a merge of overlapping dumps) writes fewer quads
1799    // than it ingested. `write_dataset_from_parts` stamps exactly this sum into
1800    // the header, so a metadata payload finalized with it cannot disagree with
1801    // the header — which is what an embedded Dataset Card used to do.
1802    let counts = FinalCounts {
1803        default_triples: def.triple_count() as u64,
1804        quads: def.triple_count() as u64 + named_quads,
1805    };
1806    let blob = pending.into_metadata(counts);
1807    progress::begin(Phase::Serialize, None, Unit::None);
1808    // Report what was WRITTEN, not what was read: the returned stats drive the
1809    // CLI's "wrote …: N quads" line, which described the file, not the input.
1810    stats.statements = counts.quads as usize;
1811    stats.default_triples = counts.default_triples as usize;
1812
1813    let bytes = crate::file::write_dataset_from_parts(
1814        &dict_container,
1815        term_count,
1816        &def,
1817        match &quad_index {
1818            Some(q) => crate::file::NamedPart::Quad(q),
1819            None => crate::file::NamedPart::PerGraph(&named_indexes),
1820        },
1821        has_named,
1822        has_quoted_triples,
1823        &meta,
1824        levels,
1825        &blob,
1826        &text_index,
1827        text_index_partial,
1828        text_index_named,
1829        codec,
1830    );
1831    // Kinds 8 and 9 go after everything else: they are outside the content
1832    // hash, so appending them moves nothing and changes no hash.
1833    let bytes = if label_section.is_empty() && schema_section.is_empty() {
1834        bytes
1835    } else {
1836        crate::file::append_sections(
1837            &bytes,
1838            &[
1839                (crate::header::SectionKind::LabelIndex, &label_section),
1840                (crate::header::SectionKind::Schema, &schema_section),
1841            ],
1842        )
1843        .expect("appending to an image this function just wrote")
1844    };
1845    crate::build_trace::lap(crate::build_trace::Phase::Serialize);
1846    (bytes, stats)
1847}
1848
1849/// Above this default-graph triple count, the index is built one permutation at a
1850/// time (low peak RAM) instead of all six in parallel. Chosen so typical builds
1851/// keep the faster parallel path while multi-hundred-million-triple graphs stay
1852/// within a bounded memory budget.
1853const LOWMEM_TRIPLE_THRESHOLD: usize = 30_000_000;
1854
1855#[cfg(test)]
1856mod tests {
1857    use super::*;
1858    use crate::Rete;
1859
1860    #[test]
1861    fn parses_iris_bnodes_literals() {
1862        let input = r#"
1863            # a comment
1864            <http://ex/Alice> <http://ex/knows> <http://ex/Bob> .
1865            <http://ex/Alice> <http://ex/age> "30"^^<http://www.w3.org/2001/XMLSchema#integer> .
1866            <http://ex/Bob> <http://ex/label> "Bob"@en .
1867            _:b0 <http://ex/p> "plain" .
1868        "#;
1869        let t = parse(input).unwrap();
1870        assert_eq!(t.len(), 4);
1871        assert_eq!(t[0].0, "<http://ex/Alice>");
1872        assert_eq!(t[0].2, "<http://ex/Bob>");
1873        assert_eq!(t[1].2, "\"30\"^^<http://www.w3.org/2001/XMLSchema#integer>");
1874        assert_eq!(t[2].2, "\"Bob\"@en");
1875        assert_eq!(t[3].0, "_:b0");
1876        assert_eq!(t[3].2, "\"plain\"");
1877    }
1878
1879    #[test]
1880    fn literal_with_spaces_and_escaped_quote() {
1881        let input = r#"<http://ex/s> <http://ex/p> "a \"quoted\" phrase here" ."#;
1882        let t = parse(input).unwrap();
1883        assert_eq!(t.len(), 1);
1884        assert_eq!(t[0].2, r#""a \"quoted\" phrase here""#);
1885    }
1886
1887    /// RDF-star: a quoted triple whose object is a **language-tagged literal**
1888    /// sits directly against the closing `>>` with no separating whitespace
1889    /// (`"name"@fr>>`). The language-tag scan must stop at `>` — a regression
1890    /// guard for the greedy scan-to-whitespace that swallowed `@fr>>` as the tag.
1891    #[test]
1892    fn quoted_triple_langtagged_object() {
1893        let s = r#"<<<http://ex/sp> <http://ex/name> "Hirondelle rustique"@fr>>"#;
1894        let (tok, rest) = take_term(s).unwrap();
1895        assert_eq!(
1896            tok,
1897            r#"<<<http://ex/sp> <http://ex/name> "Hirondelle rustique"@fr>>"#
1898        );
1899        assert_eq!(rest, "");
1900        // and it round-trips through a full annotation line
1901        let line = r#"<<<http://ex/sp> <http://ex/name> "Oreneta vulgar"@ca>> <http://purl.org/dc/terms/source> "Catalogue of Life" ."#;
1902        let t = parse(line).unwrap();
1903        assert_eq!(t.len(), 1);
1904        assert_eq!(
1905            t[0].0,
1906            r#"<<<http://ex/sp> <http://ex/name> "Oreneta vulgar"@ca>>"#
1907        );
1908        assert_eq!(t[0].2, r#""Catalogue of Life""#);
1909        // a plain lang-tagged object still terminates at whitespace
1910        assert_eq!(take_term(r#""x"@pt-BR ."#).unwrap().0, r#""x"@pt-BR"#);
1911    }
1912
1913    /// RDF 1.2 triple-term surface `<<( s p o )>>` parses to the SAME canonical
1914    /// token as the RDF-star `<< s p o >>`, so the two are interoperable (a query
1915    /// written either way matches data ingested either way).
1916    #[test]
1917    fn rdf12_triple_term_same_token_as_rdf_star() {
1918        let rdf12 = r#"<http://ex/r> <http://ex/p> <<( <http://ex/s> <http://ex/q> "o" )>> ."#;
1919        let star = r#"<http://ex/r> <http://ex/p> << <http://ex/s> <http://ex/q> "o" >> ."#;
1920        let a = parse(rdf12).unwrap();
1921        let b = parse(star).unwrap();
1922        assert_eq!(a.len(), 1);
1923        assert_eq!(a[0].2, r#"<<<http://ex/s> <http://ex/q> "o">>"#);
1924        assert_eq!(
1925            a[0].2, b[0].2,
1926            "RDF 1.2 and RDF-star must dedupe to one token"
1927        );
1928        // take_term consumes exactly the triple term (nothing left dangling).
1929        let (tok, rest) =
1930            take_term(r#"<<( <http://ex/s> <http://ex/q> <http://ex/o> )>>"#).unwrap();
1931        assert_eq!(tok, "<<<http://ex/s> <http://ex/q> <http://ex/o>>>");
1932        assert_eq!(rest, "");
1933    }
1934
1935    #[test]
1936    fn rejects_missing_dot() {
1937        assert!(parse("<a> <b> <c>").is_err());
1938    }
1939
1940    #[test]
1941    fn parses_quads_with_and_without_graph() {
1942        let input = "<http://ex/a> <http://ex/p> <http://ex/b> .\n\
1943                     <http://ex/a> <http://ex/p> <http://ex/c> <http://ex/g> .";
1944        let q = parse_quads(input).unwrap();
1945        assert_eq!(q.len(), 2);
1946        assert_eq!(q[0].3, None); // default graph
1947        assert_eq!(q[1].3, Some("<http://ex/g>".to_string()));
1948    }
1949
1950    #[test]
1951    fn parse_reader_matches_text_parse() {
1952        // The streaming reader must produce exactly the same quads as parsing the
1953        // whole text — including blank/comment skipping, a named graph, and CRLF.
1954        let nt = "<http://ex/a> <http://ex/p> <http://ex/b> .\r\n\
1955                  # comment\n\
1956                  \n\
1957                  _:b0 <http://ex/q> \"lit\"@en .\n";
1958        let via_text = parse_statements(nt, "nt").unwrap();
1959        let via_reader = parse_reader(std::io::Cursor::new(nt), "nt", 0).unwrap();
1960        assert_eq!(via_text, via_reader);
1961
1962        let nq = "<http://ex/a> <http://ex/p> <http://ex/b> .\n\
1963                  <http://ex/a> <http://ex/p> <http://ex/c> <http://ex/g> .\n";
1964        assert_eq!(
1965            parse_quads(nq).unwrap(),
1966            parse_reader(std::io::Cursor::new(nq), "nq", 0).unwrap()
1967        );
1968        // Turtle and TriG stream too — that is what lets the external build take a
1969        // multi-gigabyte `.ttl.gz` without expanding it to N-Triples first. Reading
1970        // a structured syntax from a reader must agree with parsing its whole text.
1971        let ttl = "@prefix ex: <http://ex/> .\nex:A ex:knows ex:B , ex:C .\n";
1972        assert_eq!(
1973            parse_statements(ttl, "ttl").unwrap(),
1974            parse_reader(std::io::Cursor::new(ttl), "ttl", 0).unwrap()
1975        );
1976        let trig = "@prefix ex: <http://ex/> .\nex:g { ex:A ex:knows ex:B . }\n";
1977        let streamed = parse_reader(std::io::Cursor::new(trig), "trig", 0).unwrap();
1978        assert_eq!(parse_statements(trig, "trig").unwrap(), streamed);
1979        assert_eq!(streamed[0].3.as_deref(), Some("<http://ex/g>"));
1980        // An unknown format is still a hard error, not a silent guess.
1981        assert!(parse_reader(std::io::Cursor::new(nt), "jsonld", 0).is_err());
1982    }
1983
1984    #[test]
1985    fn parse_statements_dispatches_by_format() {
1986        let nt = "<http://ex/a> <http://ex/p> <http://ex/b> .";
1987        assert_eq!(parse_statements(nt, "nt").unwrap().len(), 1);
1988        let ttl = "@prefix ex: <http://ex/> .\nex:A ex:knows ex:B , ex:C .";
1989        assert_eq!(parse_statements(ttl, "ttl").unwrap().len(), 2);
1990        // N-Triples is a subset of both Turtle and TriG, so the same text parses
1991        // under all three — and lands in the default graph under each.
1992        assert_eq!(parse_statements(nt, "trig").unwrap().len(), 1);
1993        assert!(parse_statements(nt, "trig").unwrap()[0].3.is_none());
1994        assert!(parse_statements(nt, "jsonld").is_err());
1995    }
1996
1997    /// A TriG graph block carries its graph name through as the same canonical
1998    /// token N-Quads would produce — so the two syntaxes agree on dictionary keys
1999    /// and a `.trig` shard federates with a `.nq` one.
2000    #[test]
2001    fn trig_graph_names_match_nquads() {
2002        let trig = "@prefix ex: <http://ex/> .\nex:g { ex:a ex:p ex:b . }\nex:a ex:p ex:c .\n";
2003        let nq = "<http://ex/a> <http://ex/p> <http://ex/b> <http://ex/g> .\n\
2004                  <http://ex/a> <http://ex/p> <http://ex/c> .\n";
2005        let mut from_trig = parse_statements(trig, "trig").unwrap();
2006        let mut from_nq = parse_statements(nq, "nq").unwrap();
2007        from_trig.sort();
2008        from_nq.sort();
2009        assert_eq!(from_trig, from_nq);
2010    }
2011
2012    /// RDF/XML (how most OWL ontologies ship) parses to the same canonical tokens,
2013    /// including the abbreviated typed-node syntax and `rdf:resource` references.
2014    #[test]
2015    fn parses_rdfxml_owl() {
2016        let xml = r#"<?xml version="1.0"?>
2017            <rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
2018                     xmlns:rdfs="http://www.w3.org/2000/01/rdf-schema#"
2019                     xmlns:owl="http://www.w3.org/2002/07/owl#">
2020              <owl:Class rdf:about="http://ex/Dog">
2021                <rdfs:subClassOf rdf:resource="http://ex/Animal"/>
2022                <rdfs:label>Dog</rdfs:label>
2023              </owl:Class>
2024            </rdf:RDF>"#;
2025        let triples = parse_rdfxml(xml).unwrap();
2026        // rdf:type owl:Class, rdfs:subClassOf, rdfs:label = 3 triples.
2027        assert_eq!(triples.len(), 3);
2028        assert!(triples.iter().any(|(s, p, o)| s == "<http://ex/Dog>"
2029            && p == "<http://www.w3.org/2000/01/rdf-schema#subClassOf>"
2030            && o == "<http://ex/Animal>"));
2031        // Same data through the format dispatcher (triples → default graph).
2032        assert_eq!(parse_statements(xml, "rdfxml").unwrap().len(), 3);
2033        // Malformed XML is a clear RdfXml error, not a silent empty parse.
2034        assert!(matches!(
2035            parse_statements("<rdf:RDF><not closed", "rdfxml"),
2036            Err(IngestError::RdfXml(_))
2037        ));
2038    }
2039
2040    /// Regression coverage for RUSTSEC-2026-0195: namespace declarations from
2041    /// an untrusted RDF/XML document must not make parsing panic or exhaust the
2042    /// bounded test process.
2043    #[test]
2044    fn rdfxml_namespace_fanout_is_bounded() {
2045        let declarations = (0..2_000)
2046            .map(|i| format!(r#" xmlns:p{i}="http://example.test/{i}/""#))
2047            .collect::<String>();
2048        let xml = format!(
2049            r#"<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"{declarations}/>"#
2050        );
2051        let result = parse_statements(&xml, "rdfxml");
2052        assert!(result.is_ok() || result.unwrap_err().to_string().contains("namespace"));
2053    }
2054
2055    /// Regression coverage for RUSTSEC-2026-0194: duplicate-name checking must
2056    /// complete for a large valid start tag without panicking or exhausting the
2057    /// bounded test process.
2058    #[test]
2059    fn rdfxml_attribute_fanout_completes() {
2060        let attributes = (0..2_000)
2061            .map(|i| format!(r#" p:a{i}="{i}""#))
2062            .collect::<String>();
2063        let xml = format!(
2064            r#"<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#" xmlns:p="http://example.test/"><rdf:Description rdf:about="http://example.test/s"{attributes}/></rdf:RDF>"#
2065        );
2066        assert!(parse_statements(&xml, "rdfxml").is_ok());
2067    }
2068
2069    /// Text in, queryable file image out — the whole in-memory build path the
2070    /// wasm `build()` binding uses, including a named graph.
2071    #[test]
2072    fn assemble_dataset_round_trips() {
2073        let text = "<http://ex/a> <http://ex/knows> <http://ex/b> .\n\
2074                    <http://ex/b> <http://ex/knows> <http://ex/c> .\n\
2075                    <http://ex/a> <http://ex/age> \"30\"^^<http://www.w3.org/2001/XMLSchema#integer> .\n\
2076                    <http://ex/a> <http://ex/p> <http://ex/d> <http://ex/g1> .";
2077        let quads = parse_statements(text, "nq").unwrap();
2078        let (bytes, stats) = assemble_dataset(quads, &[]);
2079        assert_eq!(stats.statements, 4);
2080        assert_eq!(stats.default_triples, 3);
2081        assert_eq!(stats.named_graphs, 1);
2082        assert!(stats.terms >= 7);
2083
2084        let rete = Rete::open(&bytes).unwrap();
2085        assert_eq!(
2086            rete.query(None, Some("<http://ex/knows>"), None).len(),
2087            2,
2088            "default-graph pattern query"
2089        );
2090        assert_eq!(rete.graph_names(), vec!["<http://ex/g1>"]);
2091        let out = crate::eval_query(
2092            &rete,
2093            "SELECT ?x WHERE { ?x <http://ex/knows> ?y . ?y <http://ex/knows> ?z }",
2094        )
2095        .unwrap();
2096        match out {
2097            crate::QueryOutput::Select(_, rows) => assert_eq!(rows.len(), 1),
2098            other => panic!("expected select result, got {other:?}"),
2099        }
2100    }
2101
2102    /// The exact minimal input the playground Build tab uses: two triples, no
2103    /// literals, no rdf:type, no named graph — the smallest graph that still
2104    /// builds a community pyramid. (The wasm `build()` panic this guards was a
2105    /// `std::time::Instant::now()` in the pyramid timing path; native has a clock
2106    /// so this passes here, while the playground harness exercises the wasm path.)
2107    #[test]
2108    fn assemble_minimal_typeless_graph() {
2109        let text = "<http://ex/A> <http://ex/knows> <http://ex/B> .\n\
2110                    <http://ex/B> <http://ex/knows> <http://ex/C> .\n";
2111        let quads = parse_statements(text, "nt").unwrap();
2112        let (bytes, stats) = assemble_dataset(quads, &[]);
2113        assert_eq!(stats.default_triples, 2);
2114        let rete = Rete::open(&bytes).unwrap();
2115        assert_eq!(rete.query(None, Some("<http://ex/knows>"), None).len(), 2);
2116    }
2117}