arrow_ipc/reader/
stream.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::collections::HashMap;
19use std::fmt::Debug;
20use std::sync::Arc;
21
22use arrow_array::{ArrayRef, RecordBatch};
23use arrow_buffer::{Buffer, MutableBuffer};
24use arrow_data::UnsafeFlag;
25use arrow_schema::{ArrowError, SchemaRef};
26
27use crate::convert::MessageBuffer;
28use crate::reader::{RecordBatchDecoder, read_dictionary_impl};
29use crate::{CONTINUATION_MARKER, MessageHeader};
30
31/// A low-level interface for reading [`RecordBatch`] data from a stream of bytes
32///
33/// See [StreamReader](crate::reader::StreamReader) for a higher-level interface
34#[derive(Debug, Default)]
35pub struct StreamDecoder {
36    /// The schema of this decoder, if read
37    schema: Option<SchemaRef>,
38    /// Lookup table for dictionaries by ID
39    dictionaries: HashMap<i64, ArrayRef>,
40    /// The decoder state
41    state: DecoderState,
42    /// A scratch buffer when a read is split across multiple `Buffer`
43    buf: MutableBuffer,
44    /// Whether or not array data in input buffers are required to be aligned
45    require_alignment: bool,
46    /// Should validation be skipped when reading data? Defaults to false.
47    ///
48    /// See [`StreamDecoder::with_skip_validation`] for details.
49    ///
50    skip_validation: UnsafeFlag,
51}
52
53#[derive(Debug)]
54enum DecoderState {
55    /// Decoding the message header
56    Header {
57        /// Temporary buffer
58        buf: [u8; 4],
59        /// Number of bytes read into buf
60        read: u8,
61        /// If we have read a continuation token
62        continuation: bool,
63    },
64    /// Decoding the message flatbuffer
65    Message {
66        /// The size of the message flatbuffer
67        size: u32,
68    },
69    /// Decoding the message body
70    Body {
71        /// The message flatbuffer
72        message: MessageBuffer,
73    },
74    /// Reached the end of the stream
75    Finished,
76}
77
78impl Default for DecoderState {
79    fn default() -> Self {
80        Self::Header {
81            buf: [0; 4],
82            read: 0,
83            continuation: false,
84        }
85    }
86}
87
88impl StreamDecoder {
89    /// Create a new [`StreamDecoder`]
90    pub fn new() -> Self {
91        Self::default()
92    }
93
94    /// Specifies whether or not array data in input buffers is required to be properly aligned.
95    ///
96    /// If `require_alignment` is true, this decoder will return an error if any array data in the
97    /// input `buf` is not properly aligned.
98    /// Under the hood it will use [`arrow_data::ArrayDataBuilder::build`] to construct
99    /// [`arrow_data::ArrayData`].
100    ///
101    /// If `require_alignment` is false (the default), this decoder will automatically allocate a
102    /// new aligned buffer and copy over the data if any array data in the input `buf` is not
103    /// properly aligned. (Properly aligned array data will remain zero-copy.)
104    /// Under the hood it will use [`arrow_data::ArrayDataBuilder::align_buffers`] to construct
105    /// [`arrow_data::ArrayData`].
106    pub fn with_require_alignment(mut self, require_alignment: bool) -> Self {
107        self.require_alignment = require_alignment;
108        self
109    }
110
111    /// Return the schema if decoded, else None.
112    pub fn schema(&self) -> Option<SchemaRef> {
113        self.schema.as_ref().map(|schema| schema.clone())
114    }
115
116    /// Specifies if validation should be skipped when reading data (defaults to `false`)
117    ///
118    /// # Safety
119    ///
120    /// This flag must only be set to `true` when you trust the input data and are
121    /// sure the data you are reading is valid Arrow IPC stream data, otherwise
122    /// undefined behavior may result.
123    ///
124    /// For example, DataFusion uses this when reading spill files it wrote itself.
125    pub unsafe fn with_skip_validation(mut self, skip_validation: bool) -> Self {
126        unsafe { self.skip_validation.set(skip_validation) };
127        self
128    }
129
130    /// Try to read the next [`RecordBatch`] from the provided [`Buffer`]
131    ///
132    /// [`Buffer::advance`] will be called on `buffer` for any consumed bytes.
133    ///
134    /// The push-based interface facilitates integration with sources that yield arbitrarily
135    /// delimited bytes ranges, such as a chunked byte stream received from object storage
136    ///
137    /// ```
138    /// # use arrow_array::RecordBatch;
139    /// # use arrow_buffer::Buffer;
140    /// # use arrow_ipc::reader::StreamDecoder;
141    /// # use arrow_schema::ArrowError;
142    /// #
143    /// fn print_stream<I>(src: impl Iterator<Item = Buffer>) -> Result<(), ArrowError> {
144    ///     let mut decoder = StreamDecoder::new();
145    ///     for mut x in src {
146    ///         while !x.is_empty() {
147    ///             if let Some(x) = decoder.decode(&mut x)? {
148    ///                 println!("{x:?}");
149    ///             }
150    ///             if let Some(schema) = decoder.schema() {
151    ///                 println!("Schema: {schema:?}");
152    ///             }
153    ///         }
154    ///     }
155    ///     decoder.finish().unwrap();
156    ///     Ok(())
157    /// }
158    /// ```
159    pub fn decode(&mut self, buffer: &mut Buffer) -> Result<Option<RecordBatch>, ArrowError> {
160        while !buffer.is_empty() {
161            match &mut self.state {
162                DecoderState::Header {
163                    buf,
164                    read,
165                    continuation,
166                } => {
167                    let offset_buf = &mut buf[*read as usize..];
168                    let to_read = buffer.len().min(offset_buf.len());
169                    offset_buf[..to_read].copy_from_slice(&buffer[..to_read]);
170                    *read += to_read as u8;
171                    buffer.advance(to_read);
172                    if *read == 4 {
173                        if !*continuation && buf == &CONTINUATION_MARKER {
174                            *continuation = true;
175                            *read = 0;
176                            continue;
177                        }
178                        let size = u32::from_le_bytes(*buf);
179
180                        if size == 0 {
181                            self.state = DecoderState::Finished;
182                            continue;
183                        }
184                        self.state = DecoderState::Message { size };
185                    }
186                }
187                DecoderState::Message { size } => {
188                    let len = *size as usize;
189                    if self.buf.is_empty() && buffer.len() > len {
190                        let message = MessageBuffer::try_new(buffer.slice_with_length(0, len))?;
191                        self.state = DecoderState::Body { message };
192                        buffer.advance(len);
193                        continue;
194                    }
195
196                    let to_read = buffer.len().min(len - self.buf.len());
197                    self.buf
198                        .try_extend_from_slice(&buffer[..to_read])
199                        .map_err(|e| ArrowError::MemoryError(e.to_string()))?;
200                    buffer.advance(to_read);
201                    if self.buf.len() == len {
202                        let message = MessageBuffer::try_new(std::mem::take(&mut self.buf).into())?;
203                        self.state = DecoderState::Body { message };
204                    }
205                }
206                DecoderState::Body { message } => {
207                    let message = message.as_ref();
208                    let body_length = message.bodyLength() as usize;
209
210                    let body = if self.buf.is_empty() && buffer.len() >= body_length {
211                        let body = buffer.slice_with_length(0, body_length);
212                        buffer.advance(body_length);
213                        body
214                    } else {
215                        let to_read = buffer.len().min(body_length - self.buf.len());
216                        self.buf
217                            .try_extend_from_slice(&buffer[..to_read])
218                            .map_err(|e| ArrowError::MemoryError(e.to_string()))?;
219                        buffer.advance(to_read);
220
221                        if self.buf.len() != body_length {
222                            continue;
223                        }
224                        std::mem::take(&mut self.buf).into()
225                    };
226
227                    let version = message.version();
228                    match message.header_type() {
229                        MessageHeader::Schema => {
230                            if self.schema.is_some() {
231                                return Err(ArrowError::IpcError(
232                                    "Not expecting a schema when messages are read".to_string(),
233                                ));
234                            }
235
236                            let ipc_schema = message.header_as_schema().ok_or_else(|| {
237                                ArrowError::ParseError(
238                                    "Unable to read IPC message as schema".to_string(),
239                                )
240                            })?;
241                            let schema = crate::convert::try_fb_to_schema(ipc_schema)?;
242                            self.state = DecoderState::default();
243                            self.schema = Some(Arc::new(schema));
244                        }
245                        MessageHeader::RecordBatch => {
246                            let batch = message.header_as_record_batch().unwrap();
247                            let schema = self.schema.clone().ok_or_else(|| {
248                                ArrowError::IpcError("Missing schema".to_string())
249                            })?;
250                            let batch = RecordBatchDecoder::try_new(
251                                &body,
252                                batch,
253                                schema,
254                                &self.dictionaries,
255                                &version,
256                            )?
257                            .with_require_alignment(self.require_alignment)
258                            .read_record_batch()?;
259                            self.state = DecoderState::default();
260                            return Ok(Some(batch));
261                        }
262                        MessageHeader::DictionaryBatch => {
263                            let dictionary = message.header_as_dictionary_batch().unwrap();
264                            let schema = self.schema.as_deref().ok_or_else(|| {
265                                ArrowError::IpcError("Missing schema".to_string())
266                            })?;
267                            read_dictionary_impl(
268                                &body,
269                                dictionary,
270                                schema,
271                                &mut self.dictionaries,
272                                &version,
273                                self.require_alignment,
274                                self.skip_validation.clone(),
275                            )?;
276                            self.state = DecoderState::default();
277                        }
278                        MessageHeader::NONE => {
279                            self.state = DecoderState::default();
280                        }
281                        t => {
282                            return Err(ArrowError::IpcError(format!(
283                                "Message type unsupported by StreamDecoder: {t:?}"
284                            )));
285                        }
286                    }
287                }
288                DecoderState::Finished => {
289                    return Err(ArrowError::IpcError("Unexpected EOS".to_string()));
290                }
291            }
292        }
293        Ok(None)
294    }
295
296    /// Signal the end of stream
297    ///
298    /// Returns an error if any partial data remains in the stream
299    pub fn finish(&mut self) -> Result<(), ArrowError> {
300        match self.state {
301            DecoderState::Finished
302            | DecoderState::Header {
303                read: 0,
304                continuation: false,
305                ..
306            } => Ok(()),
307            _ => Err(ArrowError::IpcError("Unexpected End of Stream".to_string())),
308        }
309    }
310}
311
312#[cfg(test)]
313mod tests {
314    use super::*;
315    use crate::writer::{IpcWriteOptions, StreamWriter};
316    use arrow_array::{
317        DictionaryArray, Int32Array, Int64Array, RecordBatch, RunArray, types::Int32Type,
318    };
319    use arrow_schema::{DataType, Field, Schema};
320
321    // Further tests in arrow-integration-testing/tests/ipc_reader.rs
322
323    #[test]
324    fn test_eos() {
325        let schema = Arc::new(Schema::new(vec![
326            Field::new("int32", DataType::Int32, false),
327            Field::new("int64", DataType::Int64, false),
328        ]));
329
330        let input = RecordBatch::try_new(
331            schema.clone(),
332            vec![
333                Arc::new(Int32Array::from(vec![1, 2, 3])) as _,
334                Arc::new(Int64Array::from(vec![1, 2, 3])) as _,
335            ],
336        )
337        .unwrap();
338
339        let mut buf = Vec::with_capacity(1024);
340        let mut s = StreamWriter::try_new(&mut buf, &schema).unwrap();
341        s.write(&input).unwrap();
342        s.finish().unwrap();
343        drop(s);
344
345        let buffer = Buffer::from_vec(buf);
346
347        let mut b = buffer.slice_with_length(0, buffer.len() - 1);
348        let mut decoder = StreamDecoder::new();
349        let output = decoder.decode(&mut b).unwrap().unwrap();
350        assert_eq!(output, input);
351        assert_eq!(b.len(), 7); // 8 byte EOS truncated by 1 byte
352        assert!(decoder.decode(&mut b).unwrap().is_none());
353
354        let err = decoder.finish().unwrap_err().to_string();
355        assert_eq!(err, "Ipc error: Unexpected End of Stream");
356    }
357
358    #[test]
359    fn test_schema() {
360        let schema = Arc::new(Schema::new(vec![
361            Field::new("int32", DataType::Int32, false),
362            Field::new("int64", DataType::Int64, false),
363        ]));
364
365        let mut buf = Vec::with_capacity(1024);
366        let mut s = StreamWriter::try_new(&mut buf, &schema).unwrap();
367        s.finish().unwrap();
368        drop(s);
369
370        let buffer = Buffer::from_vec(buf);
371
372        let mut b = buffer.slice_with_length(0, buffer.len() - 1);
373        let mut decoder = StreamDecoder::new();
374        let output = decoder.decode(&mut b).unwrap();
375        assert!(output.is_none());
376        let decoded_schema = decoder.schema().unwrap();
377        assert_eq!(schema, decoded_schema);
378
379        let err = decoder.finish().unwrap_err().to_string();
380        assert_eq!(err, "Ipc error: Unexpected End of Stream");
381    }
382
383    #[test]
384    fn test_read_ree_dict_record_batches_from_buffer() {
385        let schema = Schema::new(vec![Field::new(
386            "test1",
387            DataType::RunEndEncoded(
388                Arc::new(Field::new("run_ends".to_string(), DataType::Int32, false)),
389                #[expect(deprecated)]
390                Arc::new(Field::new_dict(
391                    "values".to_string(),
392                    DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)),
393                    true,
394                    0,
395                    false,
396                )),
397            ),
398            true,
399        )]);
400        let batch = RecordBatch::try_new(
401            schema.clone().into(),
402            vec![Arc::new(
403                RunArray::try_new(
404                    &Int32Array::from(vec![1, 2, 3]),
405                    &vec![Some("a"), None, Some("a")]
406                        .into_iter()
407                        .collect::<DictionaryArray<Int32Type>>(),
408                )
409                .expect("Failed to create RunArray"),
410            )],
411        )
412        .expect("Failed to create RecordBatch");
413
414        let mut buffer = vec![];
415        {
416            let mut writer = StreamWriter::try_new_with_options(
417                &mut buffer,
418                &schema,
419                IpcWriteOptions::default(),
420            )
421            .expect("Failed to create StreamWriter");
422            writer.write(&batch).expect("Failed to write RecordBatch");
423            writer.finish().expect("Failed to finish StreamWriter");
424        }
425
426        let mut decoder = StreamDecoder::new();
427        let buf = &mut Buffer::from(buffer.as_slice());
428        while let Some(batch) = decoder
429            .decode(buf)
430            .map_err(|e| {
431                ArrowError::ExternalError(format!("Failed to decode record batch: {e}").into())
432            })
433            .expect("Failed to decode record batch")
434        {
435            assert_eq!(batch, batch);
436        }
437
438        decoder.finish().expect("Failed to finish decoder");
439    }
440}
arrow_ipc/reader/stream.rs

arrow_ipc/reader/
stream.rs
