Permissive serde deserialization for partial json / stream

[kskalski]

Hi,
I was wondering about the following use-case

1. processing byte stream containing json value, so sometimes only a prefix of the full json text is available
2. parse as much content as is available to populate a serde-compatible struct
3. (optionally) continue parsing rest of the content once it's available, filling the remainder of the values for the struct

When looking at struson I think this capability is almost there, maybe except for (3.), which seems much harder in general. I guess the simplest way this could currently be done is:

• making as much of the struct fields, sub-objects, etc. #serde(default) such that serde would accept whatever partial content is there
• wrapping reader with capability of "closing open scopes" on error, such that deserializer will see incomplete-data basically as end of document

The last bullet is rather a hack, I suppose this could be handled in a more proper way inside the deserializer itself, possibly even opening a path towards (3.). Also, maybe this would allow better support for fields/sub-objects that need to be parsed as a whole and do not allow defaulting their parts.

What is your opinion about this kind of use-cases and whether attempting to still use serde for parsing partial content is worth the effort (the alternative I'm already considering is to use reader directly to manually populate the struct, though deserialize_in_place might be still a good way to integrate both worlds)?

[Marcono1234]

Due to mean the JSON stream could be cut off at some arbitrary position? In that case Struson should return SyntaxErrorKind::IncompleteDocument¹ so you could detect that and then as you proposed act on every end_array and end_object (maybe also has_next) call as if there are closing tokens.

However, there are probably a few issues with this:

• If for example a JSON string or number is cut off, there is no way to recover the incomplete start
• If the caller already started processing a JSON object member or a value (e.g. because has_next returned true), but then the end is reached you cannot properly recover from that and would have to return fake default values, e.g. "" for an incomplete JSON string

A problem with your point (3.) is also that JsonReader does not support recovering from most errors (see section "Error handling" in the documentation) because it might be an inconsistent state. I am not sure if this can easily be solved because it would probably require storing the incomplete values (e.g. an incomplete JSON string) which could become quite messy.

I assume you could probably implement something like you proposed by writing a custom JsonReader implementation which delegates most of the calls to an underlying JsonReader (e.g. JsonStreamReader), but has this special end-of-stream handling.

Just to clarify though; at the moment I don't think that I will be directly providing functionality for reading incomplete JSON values other than what is already supported² as part of Struson, instead of asking users implement their own JsonReader. I am afraid adding support for that would make the implementation too complex and error-prone.

Also, in which way is the end of the stream visible? Does Read::read really just return Ok(0), or does it block, or returns an error?

What is your opinion about this kind of use-cases and whether attempting to still use serde for parsing partial content is worth the effort

I am not actually that familiar with all the special features of Serde which might help here. It might be good to also ask this for example on Stack Overflow explicitly for Serde (and ideally link the question here); maybe someone has a good idea for that.

Hopefully that answered your questions; if not feel free to also give a short JSON or code example to demonstrate your use case. Let me know what you think.

And thanks also for considering Struson!

Footnotes

1. If there are some cases where it doesn't, I would probably consider that a bug. ↩

2. For example consuming incomplete JSON such as [1, 2 (missing ]) but simply not calling end_array or ignoring the error it would return. ↩

[Marcono1234]

Actually, ignore the issues I mentioned in my previous comment. This is probably possible, but the has_next implementation already has to peek at the next array item or object member to make sure it is complete before erroneously returning true.

So assuming your JSON is simply cut off at some point (i.e. Read::read returns Ok(0)) the following proof-of-concept shows how you could still deserialize that:

Proof-of-concept (click to expand)

Note: I haven't tested this extensively yet and it is probably not the most efficient implementation, but it should be functional.

#[derive(Debug, PartialEq)]
enum PeekedValue {
    Null,
    Bool(bool),
    Number(String),
    String(String),
    /// Peeked array start, but has not been consumed yet
    PeekedArray,
    /// Peeked object start, but has not been consumed yet
    PeekedObject,
}
impl PeekedValue {
    fn get_value_type(&self) -> ValueType {
        match self {
            PeekedValue::Null => ValueType::Null,
            PeekedValue::Bool(_) => ValueType::Boolean,
            PeekedValue::Number(_) => ValueType::Number,
            PeekedValue::String(_) => ValueType::String,
            PeekedValue::PeekedArray => ValueType::Array,
            PeekedValue::PeekedObject => ValueType::Object,
        }
    }
}

struct PartialJsonReader<J: JsonReader> {
    delegate: J,
    reached_eof: bool,
    /// Stack which is expanded every time an array or object is opened;
    /// values are `true` if object, `false` if array
    is_in_object: Vec<bool>,
    /// Temporarily holding string value or name to allow returning reference to it
    string_buf: String,
    peeked_name: Option<String>,
    peeked_value: Option<PeekedValue>,
}
impl<J: JsonReader> PartialJsonReader<J> {
    // TODO: Could probably provide real location in some cases, but for simplicty return unknown
    // location all the time
    fn get_unknown_location(&self) -> JsonErrorLocation {
        JsonErrorLocation {
            path: "?".to_owned(),
            line: 0,
            column: 0,
        }
    }

    fn peek_value(&mut self) -> Result<ValueType, ReaderError> {
        let peeked = self.delegate.peek()?;
        self.peeked_value = Some(match peeked {
            ValueType::Array => PeekedValue::PeekedArray,
            ValueType::Object => PeekedValue::PeekedObject,
            ValueType::String => PeekedValue::String(self.delegate.next_string()?),
            ValueType::Number => {
                let v = PeekedValue::Number(self.delegate.next_number_as_string()?);
                // For number must make sure complete number was processed; for example
                // `1` might actually be `1.2` or `12`

                // Only works for non-top-level value; for top-level value cannot know if number is complete
                if !self.is_in_object.is_empty() {
                    // Trigger EOF error in case nothing follows last char of number
                    self.delegate.has_next()?;
                }
                v
            }
            ValueType::Boolean => PeekedValue::Bool(self.delegate.next_bool()?),
            ValueType::Null => {
                self.delegate.next_null()?;
                PeekedValue::Null
            }
        });
        Ok(peeked)
    }

    fn has_next_impl(&mut self) -> Result<bool, ReaderError> {
        if self.delegate.has_next()? {
            // Must peek next array item / object member

            if let Some(true) = self.is_in_object.last() {
                self.peeked_name = Some(self.delegate.next_name_owned()?);
            }

            self.peek_value()?;
            Ok(true)
        } else {
            Ok(false)
        }
    }
}

impl<J: JsonReader> JsonReader for PartialJsonReader<J> {
    fn peek(&mut self) -> Result<ValueType, ReaderError> {
        // If called for top-level value and value has not peeked yet, peek at it here
        if self.is_in_object.is_empty() && self.peeked_value.is_none() {
            return self.peek_value();
        }

        if let Some(p) = &self.peeked_value {
            Ok(p.get_value_type())
        } else {
            panic!("should call `has_next` before peeking value")
        }
    }

    fn begin_object(&mut self) -> Result<(), ReaderError> {
        if let Some(p) = self.peeked_value.take() {
            if p == PeekedValue::PeekedObject {
                self.is_in_object.push(true);
                self.delegate.begin_object()
            } else {
                Err(ReaderError::UnexpectedValueType {
                    expected: ValueType::Object,
                    actual: p.get_value_type(),
                    location: self.get_unknown_location(),
                })
            }
        } else {
            panic!("should call `has_next` before consuming value")
        }
    }

    fn end_object(&mut self) -> Result<(), ReaderError> {
        self.is_in_object.pop();
        if self.reached_eof {
            Ok(())
        } else {
            self.delegate.end_object()
        }
    }

    fn begin_array(&mut self) -> Result<(), ReaderError> {
        if let Some(p) = self.peeked_value.take() {
            if p == PeekedValue::PeekedArray {
                self.is_in_object.push(false);
                self.delegate.begin_array()
            } else {
                Err(ReaderError::UnexpectedValueType {
                    expected: ValueType::Array,
                    actual: p.get_value_type(),
                    location: self.get_unknown_location(),
                })
            }
        } else {
            panic!("should call `has_next` before consuming value")
        }
    }

    fn end_array(&mut self) -> Result<(), ReaderError> {
        self.is_in_object.pop();
        if self.reached_eof {
            Ok(())
        } else {
            self.delegate.end_array()
        }
    }

    fn has_next(&mut self) -> Result<bool, ReaderError> {
        if self.reached_eof {
            Ok(false)
        } else if self.peeked_name.is_some() || self.peeked_value.is_some() {
            Ok(true)
        } else {
            match self.has_next_impl() {
                // JsonStreamReader currently reports not only `SyntaxErrorKind::IncompleteDocument`
                // on unexpected EOF, but also other errors, such as `InvalidLiteral`
                Err(ReaderError::SyntaxError(JsonSyntaxError { .. })) => {
                    self.reached_eof = true;
                    Ok(false)
                }
                // Propagate any other errors, or success result
                r => r,
            }
        }
    }

    fn next_name(&mut self) -> Result<&'_ str, ReaderError> {
        self.string_buf = self.next_name_owned()?;
        Ok(&self.string_buf)
    }

    fn next_name_owned(&mut self) -> Result<String, ReaderError> {
        if let Some(s) = self.peeked_name.take() {
            Ok(s)
        } else {
            panic!("should call `has_next` before consuming name")
        }
    }

    fn next_str(&mut self) -> Result<&'_ str, ReaderError> {
        self.string_buf = self.next_string()?;
        Ok(&self.string_buf)
    }

    fn next_string(&mut self) -> Result<String, ReaderError> {
        if let Some(p) = self.peeked_value.take() {
            if let PeekedValue::String(s) = p {
                Ok(s)
            } else {
                Err(ReaderError::UnexpectedValueType {
                    expected: ValueType::String,
                    actual: p.get_value_type(),
                    location: self.get_unknown_location(),
                })
            }
        } else {
            panic!("should call `has_next` before consuming value")
        }
    }

    fn next_string_reader(&mut self) -> Result<Box<dyn std::io::Read + '_>, ReaderError> {
        unimplemented!()
    }

    fn next_number_as_str(&mut self) -> Result<&'_ str, ReaderError> {
        self.string_buf = self.next_number_as_string()?;
        Ok(&self.string_buf)
    }

    fn next_number_as_string(&mut self) -> Result<String, ReaderError> {
        if let Some(p) = self.peeked_value.take() {
            if let PeekedValue::Number(s) = p {
                Ok(s)
            } else {
                Err(ReaderError::UnexpectedValueType {
                    expected: ValueType::Number,
                    actual: p.get_value_type(),
                    location: self.get_unknown_location(),
                })
            }
        } else {
            panic!("should call `has_next` before consuming value")
        }
    }

    fn next_bool(&mut self) -> Result<bool, ReaderError> {
        if let Some(p) = self.peeked_value.take() {
            if let PeekedValue::Bool(b) = p {
                Ok(b)
            } else {
                Err(ReaderError::UnexpectedValueType {
                    expected: ValueType::Boolean,
                    actual: p.get_value_type(),
                    location: self.get_unknown_location(),
                })
            }
        } else {
            panic!("should call `has_next` before consuming value")
        }
    }

    fn next_null(&mut self) -> Result<(), ReaderError> {
        if let Some(p) = self.peeked_value.take() {
            if p == PeekedValue::Null {
                Ok(())
            } else {
                Err(ReaderError::UnexpectedValueType {
                    expected: ValueType::Null,
                    actual: p.get_value_type(),
                    location: self.get_unknown_location(),
                })
            }
        } else {
            panic!("should call `has_next` before consuming value")
        }
    }

    fn deserialize_next<'de, D: Deserialize<'de>>(&mut self) -> Result<D, DeserializerError> {
        let mut deserializer = JsonReaderDeserializer::new(self);
        D::deserialize(&mut deserializer)
    }

    fn skip_name(&mut self) -> Result<(), ReaderError> {
        if self.peeked_name.take().is_some() {
            Ok(())
        } else {
            panic!("should call `has_next` before consuming name")
        }
    }

    // Important: This is implemented recursively; could lead to stack overflow for deeply nested JSON
    fn skip_value(&mut self) -> Result<(), ReaderError> {
        if let Some(p) = &self.peeked_value {
            // For array and object need to manually skip value here by delegating to other
            // methods to handle EOF properly; cannot delegate to underlying JSON reader
            if *p == PeekedValue::PeekedArray {
                self.begin_array()?;
                while self.has_next()? {
                    self.skip_value()?;
                }
                self.end_array()
            } else if *p == PeekedValue::PeekedObject {
                self.begin_object()?;
                while self.has_next()? {
                    self.skip_name()?;
                    self.skip_value()?;
                }
                self.end_object()
            } else {
                self.peeked_value.take();
                Ok(())
            }
        } else {
            panic!("should call `has_next` before skipping value")
        }
    }

    fn seek_to(&mut self, rel_json_path: &json_path::JsonPath) -> Result<(), ReaderError> {
        unimplemented!()
    }

    fn skip_to_top_level(&mut self) -> Result<(), ReaderError> {
        unimplemented!()
    }

    fn transfer_to<W: JsonWriter>(&mut self, json_writer: &mut W) -> Result<(), TransferError> {
        unimplemented!()
    }

    fn consume_trailing_whitespace(self) -> Result<(), ReaderError> {
        if self.reached_eof {
            Ok(())
        } else {
            self.delegate.consume_trailing_whitespace()
        }
    }
}

macro_rules! deserialize_partial {
    ($reader:expr, |$deserializer:ident| $deserializing_function:expr) => {{
        let mut json_reader = PartialJsonReader {
            delegate: JsonStreamReader::new($reader),
            reached_eof: false,
            is_in_object: Vec::new(),
            string_buf: String::new(),
            peeked_name: None,
            peeked_value: None,
        };
        let mut d = JsonReaderDeserializer::new(&mut json_reader);
        let $deserializer = &mut d;
        $deserializing_function
    }};
}

And then you can use it like this:

#[derive(Debug, Default, Deserialize)]
#[serde(default)]
struct Outer {
    a: u32,
    b: bool,
    c: Option<u32>,
    d: Vec<Inner>,
}
#[derive(Debug, Default, Deserialize)]
#[serde(default)]
struct Inner {
    e: String,
    f: f32,
}

let full_json = r#"{"a":2,"b":true,"c":null,"d":[{"e":"str\"","f":1.2e3}]}"#;
let mut json = String::new();
let mut outer = Outer::default();

for c in full_json.chars() {
    json.push(c);
    deserialize_partial!(json.as_bytes(), |d| Outer::deserialize_in_place(
        d, &mut outer
    ))
    .unwrap();
    println!("{json}\n  {outer:?}\n");
}

Is this what you had in mind?

[Marcono1234]

I have now included a slightly adjusted variant of this as integration test: https://github.com/Marcono1234/struson/blob/main/tests/partial_reader.rs

[kskalski]

Thanks

[Marcono1234]

Small follow-up: I have adjusted the implementation in the partial_reader.rs integration test a bit to reduce panics and to follow the JsonReader requirements more closely.
That should make it less error-prone to use, but it still doesn't fully implement the JsonReader trait.

[kskalski]

Oh, nice! Yes, I was a bit concerned with the problems you mentioned and thought that some kind of buffering of already read data should be enough to solve those.
I'm glad this was indeed possible and with a fixed amount of memory to remember peeked element.

Looking at the code and your test case this is pretty much what I needed. :-)
I will give it a try with the schema/objects that I'm processing.

One thing I need to add is some kind of tracking for successfully parsed top-level values. My actual use-case is processing websocket where I want to start analyzing data even before the FIN frame / full message is delivered. I already got websocket impl that hides the framing and basically exposes continuous stream of bytes as Read. The idea will be to read from that stream and every time some chunk of bytes is delivered attempt to parse (possibly partial) message out of it, so I will track the position of buffer that I successfully processed as full message (returned as Full(T)), if eof was reached by the reader, then return a Partial(T) and once more bytes are delivered re-try to deliver full message parsing again from after the last full message.

[kskalski]

I played with the above code and it works well for me. However I think that recovery from the partial parse is going to be tricky:

• ideally I would like to resume parsing (i.e. case (3)), but I don't believe this is going to be a viable solution due to corner cases and need of some kind of continuation api from serde
• but it should be also fine to recover by re-setting state of PartialJsonReader (and JsonStreamReader),
  • the main problem is that I need additional API to know exact position in underlying Read input where to start from and since JsonStreamReder has its internal buffer, I can't just rely on whatever is the read position on the object I provided to the reader - possibly JsonStreamReader should expose some kind of current_location() accessor, such that when PartialJsonReader found the top level closing token, then that location would point to the next character offset after it
  • clearly this approach requires me to keep the buffer from previous top level value end up to the current eof, but this should be fine for me

[Marcono1234]

possibly JsonStreamReader should expose some kind of current_location() accessor

#14 would cover that right?

Edit: Though the current JsonErrorLocation struct does not expose the byte position, only line and column number. I have added exposing the byte position as open question to #15.

[kskalski]

Access to reader helps, but only when I actually know where is the location that the top-level value ends. Normally with std::io::Read trait json reader might actually read more data beyond the top-level value (when it has available space in its buffer and when more bytes of input are available), so in the most general case looking at the pattern of reads is not enough (unless BufRead is used, since it has explicit consume function).

I was concerned about line/column in current location struct (maybe I could assume column is a byte position, but as 32-bit it could overflow and any new-line would prevent using it), but yeah, #15 suggests a byte position and that would be useful (alternatively location since top-level value started could be used). If such information is available on demand and I could "checkpoint" location in the input reader after top-level value is parsed, it would be great.

[Marcono1234]

but as 32-bit it could overflow

Good point; I have changed it to u64 in 17f0545 (for consistency also for the line number even though I assume in reality it is pretty unlikely that it will ever overflow).

If such information is available on demand and I could "checkpoint" location in the input reader after top-level value is parsed, it would be great.

This information will be provided as JsonReaderPosition::data_pos, which for JsonStreamReader represents the byte position. The JSON reader position will be available through JsonReader::current_position(...).

[kskalski]

One potentially useful idea here: maybe JsonStreamReader could have an implementation for reader being BufRead (or even use that as default providing an adapter constructor(?) for plain Read with https://doc.rust-lang.org/std/io/struct.BufReader.html). This way:

• any other Read implementation that is also BufRead would get a performance boost thanks to avoiding extra copying
• my use-case of rewind could be hacked as a specialized BufRead implementation that tracks fill_buf vs consume calls, but keep additional accounting to only free up relevant part of the buffer after full element from the stream got de-serialized

[Marcono1234]

Currently JsonStreamReader also uses its internal buffer to serve name and string value references from it (in case they don't contain escape sequences). It seems that would not be possible when using only BufRead since it will only refill the buffer when its empty, not when the caller simply needs more data.

Maybe that could be changed, and for example always an internal Vec<u8> buffer could be used for name and and string value references. But I am not sure yet if that is a direction in which I would like to change this.

[kskalski]

I think BufRead can be asked to re-fill the buffer by calling consume(buf.len()), though of course then the previous contents of the buf can no longer be used.

How does stream reader handle strings / names longer than its internal buffer? Clearly it cannot only rely on the contents of the buffer to create the final value (without growing it), so does it switch to storing data on the side only when it hits this case? It appears to me there is no big difference between:

• using BufRead and hitting a case where end of the slice returned by it is in the middle of a string
• using internal fixed-size buffer and hitting similar case where all of the buffer is used (by not-yet processed content), but the last read finished in the middle of a string

For optimal performance I suppose cases where whole string lands in the available buffer could be detected and leveraged to create the final whole string, but otherwise the side-buffer might be necessary anyway. There is a small difference - assuming internal buffer is a kind of ring-buffer, it might be possible to delay copying data to the side-buffer until really the entirety of internal buffer is filled - this avoids copying at the cost of doing extra read with a smaller slice, but arguably this trade-off could be both for good or bad.

[Marcono1234]

Sorry for my late response.

so does it switch to storing data on the side only when it hits this case?

Yes JsonStreamReader has a value_bytes_buf which is used in this case and also when escape sequences are encountered.

One option would be to always use a Vec<u8> / String to collect the value and not try to serve the data directly from the reader buffer at all. This might simplify some of the code and also might allow using BufRead.
I am just not sure at the moment whether that is a direction I want to take, and I will probably have to properly measure performance at some point. The existing benchmarks could be a bit helpful for this but to me they seemed to not be completely deterministic.

I have created #32 for tracking that idea.

[kskalski]

I see, it sounds good. With the position accessor added I think the ability to use BufRead would primarily be as performance improvement, so clearly it shouldn't come at a cost of degrading performance. I imagine ideally the reader would try to use borrowed data from the provided read buffer and only switch to its own buffers when forced (dealing with escaped characters or too lengthy strings), but it does in fact require careful choices how to structure the whole code base to do it.