Function deserialize_stream

datafusion/datasource/src/decoder.rs:174–193

Deserializes a stream of bytes into a stream of [`RecordBatch`] objects using the provided deserializer. Returns a boxed stream of `Result `. The stream yields [`RecordBatch`] objects as they are produced by the deserializer, or an [`ArrowError`] if an error occurs while polling the input or deserializing.

(
    mut input: impl Stream<Item = Result<Bytes>> + Unpin + Send + 'a,
    mut deserializer: impl BatchDeserializer<Bytes> + 'a,
)

/// objects as they are produced by the deserializer, or an [`ArrowError`] if an error
/// occurs while polling the input or deserializing.
pub fn deserialize_stream<'a>(
    mut input: impl Stream<Item = Result<Bytes>> + Unpin + Send + 'a,
    mut deserializer: impl BatchDeserializer<Bytes> + 'a,
) -> BoxStream<'a, Result<RecordBatch, ArrowError>> {
    futures::stream::poll_fn(move |cx| {
        loop {
            match ready!(input.poll_next_unpin(cx)).transpose()? {
                Some(b) => _ = deserializer.digest(b),
                None => deserializer.finish(),
            };

            return match deserializer.next()? {
                DeserializerOutput::RecordBatch(rb) => Poll::Ready(Some(Ok(rb))),
                DeserializerOutput::InputExhausted => Poll::Ready(None),
                DeserializerOutput::RequiresMoreData => continue,
            };
        }
    })
    .boxed()
}