oxirs_ttl/
streaming_parser.rs

1// Incremental/streaming Turtle parser for large files
2// Added in v1.1.0 Round 7
3
4use std::collections::HashMap;
5
6/// A parsed RDF term.
7#[derive(Debug, Clone, PartialEq)]
8pub enum ParsedTerm {
9    /// An absolute IRI enclosed in angle brackets.
10    Iri(String),
11    /// A prefixed name that has been expanded using the prefix map.
12    PrefixedName(String),
13    /// An RDF literal value.
14    Literal {
15        /// The lexical value of the literal.
16        value: String,
17        /// Optional XSD datatype IRI.
18        datatype: Option<String>,
19        /// Optional BCP 47 language tag.
20        lang: Option<String>,
21    },
22    /// An RDF blank node with its local identifier.
23    BlankNode(String),
24}
25
26/// A parsed RDF triple.
27#[derive(Debug, Clone, PartialEq)]
28pub struct ParsedTriple {
29    /// The subject term.
30    pub subject: ParsedTerm,
31    /// The predicate term.
32    pub predicate: ParsedTerm,
33    /// The object term.
34    pub object: ParsedTerm,
35}
36
37/// Errors from streaming parse.
38#[derive(Debug)]
39pub enum StreamParseError {
40    /// The Turtle syntax is invalid.
41    InvalidTurtle(String),
42    /// A prefixed name refers to an undeclared prefix.
43    UnknownPrefix(String),
44    /// Input ended unexpectedly mid-statement.
45    UnexpectedEof,
46    /// An IRI is malformed.
47    InvalidIri(String),
48}
49
50impl std::fmt::Display for StreamParseError {
51    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
52        match self {
53            StreamParseError::InvalidTurtle(s) => write!(f, "Invalid Turtle: {s}"),
54            StreamParseError::UnknownPrefix(p) => write!(f, "Unknown prefix: {p}"),
55            StreamParseError::UnexpectedEof => write!(f, "Unexpected end of input"),
56            StreamParseError::InvalidIri(s) => write!(f, "Invalid IRI: {s}"),
57        }
58    }
59}
60
61impl std::error::Error for StreamParseError {}
62
63/// An incremental Turtle parser that accepts chunks of text.
64pub struct StreamingParser {
65    prefixes: HashMap<String, String>,
66    buffer: String,
67    triples_parsed: u64,
68    base_iri: Option<String>,
69}
70
71impl StreamingParser {
72    /// Create a new parser with no base IRI.
73    pub fn new() -> Self {
74        Self {
75            prefixes: HashMap::new(),
76            buffer: String::new(),
77            triples_parsed: 0,
78            base_iri: None,
79        }
80    }
81
82    /// Create a new parser with a base IRI.
83    pub fn with_base(base_iri: impl Into<String>) -> Self {
84        Self {
85            prefixes: HashMap::new(),
86            buffer: String::new(),
87            triples_parsed: 0,
88            base_iri: Some(base_iri.into()),
89        }
90    }
91
92    /// Feed a chunk of Turtle text, returning all complete triples parsed.
93    pub fn feed(&mut self, chunk: &str) -> Result<Vec<ParsedTriple>, StreamParseError> {
94        self.buffer.push_str(chunk);
95        self.parse_buffer()
96    }
97
98    /// Flush any remaining buffered content, returning any remaining triples.
99    ///
100    /// `parse_buffer()` (invoked by every `feed()` call) only ever leaves the
101    /// internal buffer non-empty in two cases: it is waiting for more input
102    /// to complete the statement currently being parsed, or it is genuinely
103    /// empty — any statement that already parses cleanly is consumed and
104    /// removed from the buffer immediately, and any outright syntax error is
105    /// already surfaced as an `Err` from `feed()` itself. Therefore, if any
106    /// non-whitespace, non-comment content remains in the buffer once the
107    /// caller declares the stream finished, that content is by definition an
108    /// incomplete, truncated statement. Per this project's fail-loud
109    /// contract, that must be reported as an error rather than silently
110    /// discarded as an empty, successful result.
111    pub fn flush(&mut self) -> Result<Vec<ParsedTriple>, StreamParseError> {
112        let remaining = self
113            .skip_whitespace_and_comments(self.buffer.trim_start())
114            .trim_end()
115            .to_string();
116        self.buffer.clear();
117        if remaining.is_empty() {
118            Ok(vec![])
119        } else {
120            Err(StreamParseError::UnexpectedEof)
121        }
122    }
123
124    /// Return total triples parsed so far.
125    pub fn triples_parsed(&self) -> u64 {
126        self.triples_parsed
127    }
128
129    /// Return the current prefix map.
130    pub fn prefixes(&self) -> &HashMap<String, String> {
131        &self.prefixes
132    }
133
134    /// Set the base IRI for relative IRI resolution.
135    pub fn set_base(&mut self, base: impl Into<String>) {
136        self.base_iri = Some(base.into());
137    }
138
139    /// Reset the parser state.
140    pub fn reset(&mut self) {
141        self.prefixes.clear();
142        self.buffer.clear();
143        self.triples_parsed = 0;
144        self.base_iri = None;
145    }
146
147    /// Parse a complete Turtle document in one call.
148    pub fn parse_complete(input: &str) -> Result<Vec<ParsedTriple>, StreamParseError> {
149        let mut parser = Self::new();
150        let mut triples = parser.feed(input)?;
151        triples.extend(parser.flush()?);
152        Ok(triples)
153    }
154
155    /// Expand a prefixed name to a full IRI using current prefix map.
156    pub fn expand_prefix(&self, prefixed: &str) -> Result<String, StreamParseError> {
157        if let Some(colon) = prefixed.find(':') {
158            let prefix = &prefixed[..colon];
159            let local = &prefixed[colon + 1..];
160            match self.prefixes.get(prefix) {
161                Some(base) => Ok(format!("{base}{local}")),
162                None => Err(StreamParseError::UnknownPrefix(prefix.to_string())),
163            }
164        } else {
165            Err(StreamParseError::InvalidTurtle(format!(
166                "'{prefixed}' is not a prefixed name"
167            )))
168        }
169    }
170
171    // Parse as many complete statements as possible from the buffer.
172    fn parse_buffer(&mut self) -> Result<Vec<ParsedTriple>, StreamParseError> {
173        let mut triples = Vec::new();
174        loop {
175            // Clone the buffer into an owned string to avoid borrow conflict
176            let current = self.buffer.clone();
177            let trimmed = self
178                .skip_whitespace_and_comments(current.trim_start())
179                .to_string();
180            // Try to parse a single statement
181            match self.try_parse_statement(&trimmed) {
182                Ok(Some((stmt_triples, rest))) => {
183                    // Find how much of the original buffer was consumed
184                    let rest_owned = rest.to_string();
185                    // Reconstruct buffer: advance past what was consumed
186                    let consumed_len = trimmed.len() - rest.len();
187                    let leading_ws = current.len() - current.trim_start().len();
188                    let total_consumed = leading_ws + consumed_len;
189                    self.buffer = current[total_consumed..].to_string();
190                    triples.extend(stmt_triples);
191                    let _ = rest_owned;
192                }
193                Ok(None) => {
194                    // Need more input; leave buffer as-is
195                    break;
196                }
197                Err(e) => return Err(e),
198            }
199        }
200        Ok(triples)
201    }
202
203    fn skip_whitespace_and_comments<'a>(&self, input: &'a str) -> &'a str {
204        let mut s = input;
205        loop {
206            let trimmed = s.trim_start();
207            if trimmed.starts_with('#') {
208                // Skip to end of line
209                if let Some(nl) = trimmed.find('\n') {
210                    s = &trimmed[nl + 1..];
211                } else {
212                    return "";
213                }
214            } else {
215                return trimmed;
216            }
217        }
218    }
219
220    /// Try to parse one complete statement (prefix declaration or triple).
221    /// Returns None if there is not enough input yet.
222    fn try_parse_statement<'a>(
223        &mut self,
224        input: &'a str,
225    ) -> Result<Option<(Vec<ParsedTriple>, &'a str)>, StreamParseError> {
226        let s = self.skip_whitespace_and_comments(input);
227        if s.is_empty() {
228            return Ok(None);
229        }
230
231        // @prefix or PREFIX declaration
232        if s.starts_with("@prefix") || s.starts_with("@PREFIX") {
233            return self.parse_prefix_decl(s);
234        }
235        if s.to_uppercase().starts_with("PREFIX ") {
236            return self.parse_sparql_prefix_decl(s);
237        }
238        // @base declaration
239        if s.starts_with("@base") || s.starts_with("@BASE") {
240            return self.parse_base_decl(s);
241        }
242        if s.to_uppercase().starts_with("BASE ") {
243            return self.parse_sparql_base_decl(s);
244        }
245
246        // Try to parse a triple statement (ends with '.')
247        // Find the next '.' that ends a statement
248        self.try_parse_triple_statement(s)
249    }
250
251    fn parse_prefix_decl<'a>(
252        &mut self,
253        input: &'a str,
254    ) -> Result<Option<(Vec<ParsedTriple>, &'a str)>, StreamParseError> {
255        // @prefix prefix: <iri> .
256        let rest = input
257            .trim_start_matches("@prefix")
258            .trim_start_matches("@PREFIX");
259        let rest = rest.trim_start();
260        // Read prefix name (up to ':')
261        let colon_pos = rest.find(':').ok_or(StreamParseError::UnexpectedEof)?;
262        let prefix_name = rest[..colon_pos].trim().to_string();
263        let rest = rest[colon_pos + 1..].trim_start();
264        // Read IRI
265        let after_lt = match rest.strip_prefix('<') {
266            Some(r) => r,
267            None => {
268                if rest.is_empty() {
269                    return Ok(None);
270                }
271                return Err(StreamParseError::InvalidTurtle(
272                    "Expected '<' after prefix in @prefix declaration".to_string(),
273                ));
274            }
275        };
276        let iri_end = after_lt.find('>').ok_or(StreamParseError::UnexpectedEof)?;
277        let iri = after_lt[..iri_end].to_string();
278        let rest = after_lt[iri_end + 1..].trim_start();
279        // Consume optional '.'
280        let rest = rest.strip_prefix('.').unwrap_or(rest);
281        self.prefixes.insert(prefix_name, iri);
282        Ok(Some((vec![], rest)))
283    }
284
285    fn parse_sparql_prefix_decl<'a>(
286        &mut self,
287        input: &'a str,
288    ) -> Result<Option<(Vec<ParsedTriple>, &'a str)>, StreamParseError> {
289        // PREFIX prefix: <iri>  (no trailing dot)
290        // Safe to slice at byte 6 because "PREFIX" is ASCII
291        let rest = input[6..].trim_start(); // skip "PREFIX"
292        let colon_pos = rest.find(':').ok_or(StreamParseError::UnexpectedEof)?;
293        let prefix_name = rest[..colon_pos].trim().to_string();
294        let rest = rest[colon_pos + 1..].trim_start();
295        let after_lt2 = match rest.strip_prefix('<') {
296            Some(r) => r,
297            None => {
298                if rest.is_empty() {
299                    return Ok(None);
300                }
301                return Err(StreamParseError::InvalidTurtle(
302                    "Expected '<' in PREFIX declaration".to_string(),
303                ));
304            }
305        };
306        let iri_end = after_lt2.find('>').ok_or(StreamParseError::UnexpectedEof)?;
307        let iri = after_lt2[..iri_end].to_string();
308        let rest = &after_lt2[iri_end + 1..];
309        self.prefixes.insert(prefix_name, iri);
310        Ok(Some((vec![], rest)))
311    }
312
313    fn parse_base_decl<'a>(
314        &mut self,
315        input: &'a str,
316    ) -> Result<Option<(Vec<ParsedTriple>, &'a str)>, StreamParseError> {
317        // Strip @base or @BASE prefix (5 ASCII bytes)
318        let rest = &input[5..];
319        let rest = rest.trim_start();
320        let after_lt = match rest.strip_prefix('<') {
321            Some(r) => r,
322            None => return Ok(None),
323        };
324        let iri_end = after_lt.find('>').ok_or(StreamParseError::UnexpectedEof)?;
325        let iri = after_lt[..iri_end].to_string();
326        let rest = after_lt[iri_end + 1..].trim_start();
327        let rest = rest.strip_prefix('.').unwrap_or(rest);
328        self.base_iri = Some(iri);
329        Ok(Some((vec![], rest)))
330    }
331
332    fn parse_sparql_base_decl<'a>(
333        &mut self,
334        input: &'a str,
335    ) -> Result<Option<(Vec<ParsedTriple>, &'a str)>, StreamParseError> {
336        // Strip "BASE" (4 ASCII bytes)
337        let rest = input[4..].trim_start();
338        let after_lt = match rest.strip_prefix('<') {
339            Some(r) => r,
340            None => return Ok(None),
341        };
342        let iri_end = after_lt.find('>').ok_or(StreamParseError::UnexpectedEof)?;
343        let iri = after_lt[..iri_end].to_string();
344        let rest = &after_lt[iri_end + 1..];
345        self.base_iri = Some(iri);
346        Ok(Some((vec![], rest)))
347    }
348
349    fn try_parse_triple_statement<'a>(
350        &mut self,
351        input: &'a str,
352    ) -> Result<Option<(Vec<ParsedTriple>, &'a str)>, StreamParseError> {
353        // Find terminating '.' - but must not be inside a string literal or IRI
354        let dot_pos = find_statement_end(input);
355        let dot_pos = match dot_pos {
356            Some(p) => p,
357            None => return Ok(None), // incomplete
358        };
359
360        let statement = &input[..dot_pos];
361        let rest = &input[dot_pos + 1..]; // skip '.'
362
363        let triple = self.parse_triple_text(statement.trim())?;
364        self.triples_parsed += 1;
365        Ok(Some((vec![triple], rest)))
366    }
367
368    fn parse_triple_text(&mut self, text: &str) -> Result<ParsedTriple, StreamParseError> {
369        let (subject, rest) = self.parse_term(text)?;
370        let rest = self.skip_whitespace_and_comments(rest.trim_start());
371        let (predicate, rest) = self.parse_term(rest)?;
372        let rest = self.skip_whitespace_and_comments(rest.trim_start());
373        let (object, _rest) = self.parse_term(rest)?;
374        Ok(ParsedTriple {
375            subject,
376            predicate,
377            object,
378        })
379    }
380
381    fn parse_term<'a>(&self, input: &'a str) -> Result<(ParsedTerm, &'a str), StreamParseError> {
382        let s = input.trim_start();
383        if let Some(after_lt) = s.strip_prefix('<') {
384            // Absolute IRI
385            let end = after_lt
386                .find('>')
387                .ok_or(StreamParseError::InvalidIri("Unclosed '<'".to_string()))?;
388            let iri = after_lt[..end].to_string();
389            let term = ParsedTerm::Iri(iri);
390            return Ok((term, &after_lt[end + 1..]));
391        }
392        if s.starts_with('"') {
393            // String literal
394            return self.parse_literal(s);
395        }
396        if s.starts_with('\'') {
397            // Single-quoted literal
398            return self.parse_single_quoted_literal(s);
399        }
400        if let Some(after_bn) = s.strip_prefix("_:") {
401            // Blank node
402            let end = after_bn
403                .find(|c: char| c.is_whitespace() || c == '.' || c == ';' || c == ',')
404                .unwrap_or(after_bn.len());
405            let name = after_bn[..end].to_string();
406            return Ok((ParsedTerm::BlankNode(name), &after_bn[end..]));
407        }
408        if let Some(after_true) = s.strip_prefix("true") {
409            return Ok((
410                ParsedTerm::Literal {
411                    value: "true".to_string(),
412                    datatype: Some("http://www.w3.org/2001/XMLSchema#boolean".to_string()),
413                    lang: None,
414                },
415                after_true,
416            ));
417        }
418        if let Some(after_false) = s.strip_prefix("false") {
419            return Ok((
420                ParsedTerm::Literal {
421                    value: "false".to_string(),
422                    datatype: Some("http://www.w3.org/2001/XMLSchema#boolean".to_string()),
423                    lang: None,
424                },
425                after_false,
426            ));
427        }
428        // Prefixed name or keyword 'a'
429        if let Some(after_a) = s.strip_prefix('a') {
430            if after_a.is_empty() || after_a.starts_with(|c: char| c.is_whitespace()) {
431                // 'a' is short for rdf:type
432                let expanded = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type".to_string();
433                return Ok((ParsedTerm::Iri(expanded), after_a));
434            }
435        }
436        // Number literal
437        if s.starts_with(|c: char| c.is_ascii_digit() || c == '-') {
438            let end = s
439                .find(|c: char| c.is_whitespace() || c == '.' || c == ';' || c == ',')
440                .unwrap_or(s.len());
441            // Be careful with '.' — only consume it if it's not the statement terminator
442            let num_str = &s[..end];
443            if !num_str.is_empty() {
444                let datatype = if num_str.contains('.') {
445                    "http://www.w3.org/2001/XMLSchema#decimal"
446                } else {
447                    "http://www.w3.org/2001/XMLSchema#integer"
448                };
449                return Ok((
450                    ParsedTerm::Literal {
451                        value: num_str.to_string(),
452                        datatype: Some(datatype.to_string()),
453                        lang: None,
454                    },
455                    &s[end..],
456                ));
457            }
458        }
459        // Prefixed name: prefix:local
460        let name_end = s
461            .find(|c: char| c.is_whitespace() || c == '.' || c == ';' || c == ',' || c == ')')
462            .unwrap_or(s.len());
463        let name = &s[..name_end];
464        if name.is_empty() {
465            return Err(StreamParseError::InvalidTurtle(format!(
466                "Unexpected input: '{}'",
467                &s[..s.len().min(20)]
468            )));
469        }
470        if name.contains(':') {
471            let expanded = self.expand_prefix(name)?;
472            return Ok((ParsedTerm::PrefixedName(expanded), &s[name_end..]));
473        }
474        Err(StreamParseError::InvalidTurtle(format!(
475            "Cannot parse term from: '{}'",
476            &s[..s.len().min(30)]
477        )))
478    }
479
480    fn parse_literal<'a>(&self, input: &'a str) -> Result<(ParsedTerm, &'a str), StreamParseError> {
481        // Handle triple-quoted strings first
482        if let Some(after_tq) = input.strip_prefix("\"\"\"") {
483            let end = after_tq
484                .find("\"\"\"")
485                .ok_or(StreamParseError::InvalidTurtle(
486                    "Unclosed triple-quoted string".to_string(),
487                ))?;
488            let value = after_tq[..end].to_string();
489            let rest = &after_tq[end + 3..];
490            return self.parse_literal_suffix(value, rest);
491        }
492        // Double-quoted string: scan to find the closing '"' and record its byte position
493        let rest = input.strip_prefix('"').unwrap_or(&input[1..]); // skip opening '"'
494        let mut value = String::new();
495        let mut chars = rest.char_indices();
496        let closing_idx = loop {
497            match chars.next() {
498                Some((i, '"')) => break i,
499                Some((_, '\\')) => {
500                    match chars.next() {
501                        Some((_, 'n')) => value.push('\n'),
502                        Some((_, 't')) => value.push('\t'),
503                        Some((_, '"')) => value.push('"'),
504                        Some((_, '\\')) => value.push('\\'),
505                        Some((_, 'u')) => {
506                            // Unicode escape: \uXXXX
507                            let mut hex = String::new();
508                            for _ in 0..4 {
509                                if let Some((_, c)) = chars.next() {
510                                    hex.push(c);
511                                }
512                            }
513                            if let Ok(code) = u32::from_str_radix(&hex, 16) {
514                                if let Some(c) = char::from_u32(code) {
515                                    value.push(c);
516                                }
517                            }
518                        }
519                        Some((_, c)) => {
520                            value.push('\\');
521                            value.push(c);
522                        }
523                        None => return Err(StreamParseError::UnexpectedEof),
524                    }
525                }
526                Some((_, c)) => value.push(c),
527                None => {
528                    return Err(StreamParseError::InvalidTurtle(
529                        "Unclosed string literal".to_string(),
530                    ))
531                }
532            }
533        };
534        let rest = &rest[closing_idx + 1..]; // skip closing '"'
535        self.parse_literal_suffix(value, rest)
536    }
537
538    fn parse_single_quoted_literal<'a>(
539        &self,
540        input: &'a str,
541    ) -> Result<(ParsedTerm, &'a str), StreamParseError> {
542        let rest = input.strip_prefix('\'').unwrap_or(&input[1..]);
543        let mut value = String::new();
544        let mut chars = rest.char_indices();
545        let closing_idx = loop {
546            match chars.next() {
547                Some((i, '\'')) => break i,
548                Some((_, '\\')) => match chars.next() {
549                    Some((_, c)) => value.push(c),
550                    None => return Err(StreamParseError::UnexpectedEof),
551                },
552                Some((_, c)) => value.push(c),
553                None => {
554                    return Err(StreamParseError::InvalidTurtle(
555                        "Unclosed single-quoted literal".to_string(),
556                    ))
557                }
558            }
559        };
560        let rest = &rest[closing_idx + 1..];
561        self.parse_literal_suffix(value, rest)
562    }
563
564    fn parse_literal_suffix<'a>(
565        &self,
566        value: String,
567        rest: &'a str,
568    ) -> Result<(ParsedTerm, &'a str), StreamParseError> {
569        if let Some(after_hat) = rest.strip_prefix("^^") {
570            let rest = after_hat;
571            if let Some(after_lt) = rest.strip_prefix('<') {
572                let end = after_lt.find('>').ok_or(StreamParseError::InvalidIri(
573                    "Unclosed datatype IRI".to_string(),
574                ))?;
575                let datatype = after_lt[..end].to_string();
576                let rest = &after_lt[end + 1..];
577                return Ok((
578                    ParsedTerm::Literal {
579                        value,
580                        datatype: Some(datatype),
581                        lang: None,
582                    },
583                    rest,
584                ));
585            }
586            // Prefixed datatype
587            let end = rest
588                .find(|c: char| c.is_whitespace() || c == '.' || c == ';' || c == ',')
589                .unwrap_or(rest.len());
590            let pname = &rest[..end];
591            let datatype = self
592                .expand_prefix(pname)
593                .unwrap_or_else(|_| pname.to_string());
594            let rest = &rest[end..];
595            return Ok((
596                ParsedTerm::Literal {
597                    value,
598                    datatype: Some(datatype),
599                    lang: None,
600                },
601                rest,
602            ));
603        }
604        if let Some(after_at) = rest.strip_prefix('@') {
605            let end = after_at
606                .find(|c: char| c.is_whitespace() || c == '.' || c == ';' || c == ',')
607                .unwrap_or(after_at.len());
608            let lang = after_at[..end].to_string();
609            let rest = &after_at[end..];
610            return Ok((
611                ParsedTerm::Literal {
612                    value,
613                    datatype: None,
614                    lang: Some(lang),
615                },
616                rest,
617            ));
618        }
619        Ok((
620            ParsedTerm::Literal {
621                value,
622                datatype: None,
623                lang: None,
624            },
625            rest,
626        ))
627    }
628}
629
630impl Default for StreamingParser {
631    fn default() -> Self {
632        Self::new()
633    }
634}
635
636/// Find the position of the terminating '.' for a statement,
637/// ignoring '.' inside string literals and IRIs.
638fn find_statement_end(input: &str) -> Option<usize> {
639    let mut in_string = false;
640    let mut in_iri = false;
641    let mut triple_quote = false;
642    let chars: Vec<char> = input.chars().collect();
643    let n = chars.len();
644    let mut i = 0;
645    while i < n {
646        let c = chars[i];
647        if in_iri {
648            if c == '>' {
649                in_iri = false;
650            }
651            i += 1;
652            continue;
653        }
654        if in_string {
655            if triple_quote {
656                if i + 2 < n && c == '"' && chars[i + 1] == '"' && chars[i + 2] == '"' {
657                    in_string = false;
658                    triple_quote = false;
659                    i += 3;
660                    continue;
661                }
662            } else {
663                if c == '\\' {
664                    i += 2;
665                    continue;
666                }
667                if c == '"' {
668                    in_string = false;
669                }
670            }
671            i += 1;
672            continue;
673        }
674        // Not in string or IRI
675        if c == '#' {
676            // Comment to end of line — skip
677            while i < n && chars[i] != '\n' {
678                i += 1;
679            }
680            continue;
681        }
682        if c == '<' {
683            in_iri = true;
684            i += 1;
685            continue;
686        }
687        if c == '"' {
688            if i + 2 < n && chars[i + 1] == '"' && chars[i + 2] == '"' {
689                triple_quote = true;
690                in_string = true;
691                i += 3;
692                continue;
693            }
694            in_string = true;
695            i += 1;
696            continue;
697        }
698        if c == '.' {
699            // Make sure the dot is followed by whitespace or EOF (not a decimal)
700            let next = chars.get(i + 1);
701            match next {
702                None => return Some(i),
703                Some(nc) if nc.is_whitespace() || *nc == '#' => return Some(i),
704                _ => {}
705            }
706        }
707        i += 1;
708    }
709    None
710}
711
712#[cfg(test)]
713mod tests {
714    use super::*;
715
716    // ---- parse_complete ----
717
718    #[test]
719    fn test_parse_complete_simple() {
720        let input = "<http://s> <http://p> <http://o> .";
721        let triples = StreamingParser::parse_complete(input).expect("should succeed");
722        assert_eq!(triples.len(), 1);
723        assert_eq!(triples[0].subject, ParsedTerm::Iri("http://s".to_string()));
724        assert_eq!(
725            triples[0].predicate,
726            ParsedTerm::Iri("http://p".to_string())
727        );
728        assert_eq!(triples[0].object, ParsedTerm::Iri("http://o".to_string()));
729    }
730
731    #[test]
732    fn test_parse_complete_empty() {
733        let triples = StreamingParser::parse_complete("").expect("should succeed");
734        assert!(triples.is_empty());
735    }
736
737    #[test]
738    fn test_parse_complete_comment_only() {
739        let triples =
740            StreamingParser::parse_complete("# just a comment\n").expect("should succeed");
741        assert!(triples.is_empty());
742    }
743
744    #[test]
745    fn test_parse_complete_multiple_triples() {
746        let input = "<http://s1> <http://p> <http://o1> .\n<http://s2> <http://p> <http://o2> .";
747        let triples = StreamingParser::parse_complete(input).expect("should succeed");
748        assert_eq!(triples.len(), 2);
749    }
750
751    // ---- prefix declarations ----
752
753    #[test]
754    fn test_parse_prefix_declaration() {
755        let input = "@prefix ex: <http://example.org/> .\n<http://s> <http://p> <http://o> .";
756        let triples = StreamingParser::parse_complete(input).expect("should succeed");
757        assert_eq!(triples.len(), 1);
758    }
759
760    #[test]
761    fn test_parse_prefixed_name_expansion() {
762        let input = "@prefix ex: <http://example.org/> .\nex:Alice ex:knows ex:Bob .";
763        let triples = StreamingParser::parse_complete(input).expect("should succeed");
764        assert_eq!(triples.len(), 1);
765        assert_eq!(
766            triples[0].subject,
767            ParsedTerm::PrefixedName("http://example.org/Alice".to_string())
768        );
769    }
770
771    #[test]
772    fn test_parse_sparql_prefix() {
773        let input = "PREFIX ex: <http://example.org/>\n<http://s> <http://p> <http://o> .";
774        let triples = StreamingParser::parse_complete(input).expect("should succeed");
775        assert_eq!(triples.len(), 1);
776    }
777
778    // ---- expand_prefix ----
779
780    #[test]
781    fn test_expand_prefix_known() {
782        let mut parser = StreamingParser::new();
783        parser
784            .prefixes
785            .insert("ex".to_string(), "http://example.org/".to_string());
786        let expanded = parser.expand_prefix("ex:Alice").expect("should succeed");
787        assert_eq!(expanded, "http://example.org/Alice");
788    }
789
790    #[test]
791    fn test_expand_prefix_unknown_error() {
792        let parser = StreamingParser::new();
793        let result = parser.expand_prefix("unknown:Thing");
794        assert!(matches!(result, Err(StreamParseError::UnknownPrefix(_))));
795    }
796
797    #[test]
798    fn test_expand_prefix_no_colon_error() {
799        let parser = StreamingParser::new();
800        let result = parser.expand_prefix("nocolon");
801        assert!(matches!(result, Err(StreamParseError::InvalidTurtle(_))));
802    }
803
804    // ---- literals ----
805
806    #[test]
807    fn test_parse_string_literal() {
808        let input = "<http://s> <http://p> \"hello\" .";
809        let triples = StreamingParser::parse_complete(input).expect("should succeed");
810        assert_eq!(triples.len(), 1);
811        assert_eq!(
812            triples[0].object,
813            ParsedTerm::Literal {
814                value: "hello".to_string(),
815                datatype: None,
816                lang: None
817            }
818        );
819    }
820
821    #[test]
822    fn test_parse_literal_with_datatype() {
823        let input = "<http://s> <http://p> \"42\"^^<http://www.w3.org/2001/XMLSchema#integer> .";
824        let triples = StreamingParser::parse_complete(input).expect("should succeed");
825        assert_eq!(triples.len(), 1);
826        match &triples[0].object {
827            ParsedTerm::Literal {
828                value, datatype, ..
829            } => {
830                assert_eq!(value, "42");
831                assert!(datatype.as_deref().unwrap_or("").contains("integer"));
832            }
833            _ => panic!("expected literal"),
834        }
835    }
836
837    #[test]
838    fn test_parse_literal_with_lang() {
839        let input = "<http://s> <http://p> \"hello\"@en .";
840        let triples = StreamingParser::parse_complete(input).expect("should succeed");
841        assert_eq!(triples.len(), 1);
842        match &triples[0].object {
843            ParsedTerm::Literal { value, lang, .. } => {
844                assert_eq!(value, "hello");
845                assert_eq!(lang.as_deref(), Some("en"));
846            }
847            _ => panic!("expected literal"),
848        }
849    }
850
851    // ---- blank nodes ----
852
853    #[test]
854    fn test_parse_blank_node() {
855        let input = "_:b0 <http://p> <http://o> .";
856        let triples = StreamingParser::parse_complete(input).expect("should succeed");
857        assert_eq!(triples.len(), 1);
858        assert_eq!(triples[0].subject, ParsedTerm::BlankNode("b0".to_string()));
859    }
860
861    #[test]
862    fn test_parse_blank_node_object() {
863        let input = "<http://s> <http://p> _:b1 .";
864        let triples = StreamingParser::parse_complete(input).expect("should succeed");
865        assert_eq!(triples.len(), 1);
866        assert_eq!(triples[0].object, ParsedTerm::BlankNode("b1".to_string()));
867    }
868
869    // ---- chunked parsing ----
870
871    #[test]
872    fn test_parse_in_two_chunks_simple() {
873        let mut parser = StreamingParser::new();
874        let chunk1 = "<http://s> <http://p>";
875        let chunk2 = " <http://o> .";
876        let t1 = parser.feed(chunk1).expect("should succeed");
877        assert!(t1.is_empty()); // incomplete
878        let t2 = parser.feed(chunk2).expect("should succeed");
879        assert_eq!(t2.len(), 1);
880        assert_eq!(parser.triples_parsed(), 1);
881    }
882
883    #[test]
884    fn test_parse_prefix_then_triple_in_chunks() {
885        let mut parser = StreamingParser::new();
886        let chunk1 = "@prefix ex: <http://example.org/> .\n";
887        let chunk2 = "ex:s ex:p ex:o .";
888        let t1 = parser.feed(chunk1).expect("should succeed");
889        assert!(t1.is_empty());
890        let t2 = parser.feed(chunk2).expect("should succeed");
891        assert_eq!(t2.len(), 1);
892    }
893
894    #[test]
895    fn test_parse_10_triples_chunked() {
896        let mut input = String::new();
897        for i in 0..10 {
898            input.push_str(&format!("<http://s{i}> <http://p> <http://o{i}> .\n"));
899        }
900        let mut parser = StreamingParser::new();
901        let mid = input.len() / 2;
902        let t1 = parser.feed(&input[..mid]).expect("should succeed");
903        let t2 = parser.feed(&input[mid..]).expect("should succeed");
904        let total = t1.len() + t2.len();
905        assert_eq!(total, 10, "Expected 10 triples, got {total}");
906    }
907
908    // ---- triples_parsed counter ----
909
910    #[test]
911    fn test_triples_parsed_counter() {
912        let mut parser = StreamingParser::new();
913        parser
914            .feed("<http://s1> <http://p> <http://o1> .")
915            .expect("should succeed");
916        parser
917            .feed("<http://s2> <http://p> <http://o2> .")
918            .expect("should succeed");
919        assert_eq!(parser.triples_parsed(), 2);
920    }
921
922    // ---- prefixes accessor ----
923
924    #[test]
925    fn test_prefixes_accessor() {
926        let mut parser = StreamingParser::new();
927        parser
928            .feed("@prefix ex: <http://example.org/> .")
929            .expect("should succeed");
930        assert!(parser.prefixes().contains_key("ex"));
931    }
932
933    // ---- reset ----
934
935    #[test]
936    fn test_reset() {
937        let mut parser = StreamingParser::new();
938        parser
939            .feed("<http://s> <http://p> <http://o> .")
940            .expect("should succeed");
941        parser
942            .feed("@prefix ex: <http://example.org/> .")
943            .expect("should succeed");
944        parser.reset();
945        assert_eq!(parser.triples_parsed(), 0);
946        assert!(parser.prefixes().is_empty());
947    }
948
949    // ---- base IRI ----
950
951    #[test]
952    fn test_with_base_iri() {
953        let parser = StreamingParser::with_base("http://base.example.org/");
954        assert_eq!(
955            parser.base_iri,
956            Some("http://base.example.org/".to_string())
957        );
958    }
959
960    #[test]
961    fn test_set_base() {
962        let mut parser = StreamingParser::new();
963        parser.set_base("http://new-base.example.org/");
964        assert_eq!(
965            parser.base_iri,
966            Some("http://new-base.example.org/".to_string())
967        );
968    }
969
970    // ---- flush ----
971
972    #[test]
973    fn test_flush_empty_buffer() {
974        let mut parser = StreamingParser::new();
975        let result = parser.flush().expect("should succeed");
976        assert!(result.is_empty());
977    }
978
979    #[test]
980    fn test_flush_after_complete_triple() {
981        let mut parser = StreamingParser::new();
982        parser
983            .feed("<http://s> <http://p> <http://o> .")
984            .expect("should succeed");
985        let result = parser.flush().expect("should succeed");
986        assert!(result.is_empty()); // already parsed
987    }
988
989    #[test]
990    fn regression_flush_errors_on_truncated_trailing_statement() {
991        // Feeding a statement that never gets a trailing '.' before the
992        // stream ends must surface as an error at flush() time, not a
993        // silent, successful empty result (fail-loud contract).
994        let mut parser = StreamingParser::new();
995        parser
996            .feed("<http://s> <http://p> <http://o>")
997            .expect("feed of an incomplete statement should not itself error");
998        let result = parser.flush();
999        assert!(
1000            matches!(result, Err(StreamParseError::UnexpectedEof)),
1001            "flush() on a truncated trailing statement must return Err(UnexpectedEof), got {result:?}"
1002        );
1003    }
1004
1005    #[test]
1006    fn regression_flush_errors_on_partial_iri_token() {
1007        let mut parser = StreamingParser::new();
1008        parser
1009            .feed("<http://s> <http://p> <http://incomplete")
1010            .expect("feed of a partial token should not itself error");
1011        let result = parser.flush();
1012        assert!(
1013            result.is_err(),
1014            "a dangling, unterminated token must error at flush"
1015        );
1016    }
1017
1018    #[test]
1019    fn regression_parse_complete_errors_on_truncated_document() {
1020        // parse_complete() = feed() + flush(); a document that ends
1021        // mid-statement must propagate an error rather than silently
1022        // dropping the trailing (incomplete) statement.
1023        let input = "<http://s1> <http://p> <http://o1> .\n<http://s2> <http://p> <http://o2>";
1024        let result = StreamingParser::parse_complete(input);
1025        assert!(
1026            result.is_err(),
1027            "a document truncated mid-statement must fail rather than succeed with a silently dropped triple"
1028        );
1029    }
1030
1031    #[test]
1032    fn regression_flush_still_ok_for_trailing_comment_only() {
1033        // A trailing comment-only remainder (no real dangling content) must
1034        // still flush cleanly, not be misclassified as truncated input.
1035        let mut parser = StreamingParser::new();
1036        parser
1037            .feed("<http://s> <http://p> <http://o> .\n# trailing comment")
1038            .expect("should succeed");
1039        let result = parser
1040            .flush()
1041            .expect("trailing comment-only remainder should not error");
1042        assert!(result.is_empty());
1043    }
1044
1045    // ---- errors ----
1046
1047    #[test]
1048    fn test_stream_parse_error_display() {
1049        let e = StreamParseError::InvalidTurtle("oops".to_string());
1050        assert!(format!("{e}").contains("oops"));
1051        let e2 = StreamParseError::UnknownPrefix("foo".to_string());
1052        assert!(format!("{e2}").contains("foo"));
1053        let e3 = StreamParseError::UnexpectedEof;
1054        assert!(!format!("{e3}").is_empty());
1055        let e4 = StreamParseError::InvalidIri("bad".to_string());
1056        assert!(format!("{e4}").contains("bad"));
1057    }
1058
1059    // ---- default ----
1060
1061    #[test]
1062    fn test_default() {
1063        let parser = StreamingParser::default();
1064        assert_eq!(parser.triples_parsed(), 0);
1065        assert!(parser.prefixes().is_empty());
1066    }
1067
1068    // ---- rdf:type shorthand 'a' ----
1069
1070    #[test]
1071    fn test_rdf_type_shorthand() {
1072        let input = "<http://s> a <http://Type> .";
1073        let triples = StreamingParser::parse_complete(input).expect("should succeed");
1074        assert_eq!(triples.len(), 1);
1075        assert_eq!(
1076            triples[0].predicate,
1077            ParsedTerm::Iri("http://www.w3.org/1999/02/22-rdf-syntax-ns#type".to_string())
1078        );
1079    }
1080}
oxirs_ttl/streaming_parser.rs

oxirs_ttl/
streaming_parser.rs
