Skip to main content

vortex_array/
serde.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4use std::borrow::Cow;
5use std::fmt::Debug;
6use std::fmt::Formatter;
7use std::iter;
8use std::sync::Arc;
9
10use flatbuffers::FlatBufferBuilder;
11use flatbuffers::Follow;
12use flatbuffers::WIPOffset;
13use flatbuffers::root;
14use vortex_buffer::Alignment;
15use vortex_buffer::ByteBuffer;
16use vortex_error::VortexError;
17use vortex_error::VortexExpect;
18use vortex_error::VortexResult;
19use vortex_error::vortex_bail;
20use vortex_error::vortex_err;
21use vortex_error::vortex_panic;
22use vortex_flatbuffers::FlatBuffer;
23use vortex_flatbuffers::WriteFlatBuffer;
24use vortex_flatbuffers::array as fba;
25use vortex_flatbuffers::array::Compression;
26use vortex_session::VortexSession;
27use vortex_session::registry::ReadContext;
28use vortex_utils::aliases::hash_map::HashMap;
29
30use crate::ArrayContext;
31use crate::ArrayRef;
32use crate::ArraySlots;
33use crate::array::ArrayId;
34use crate::array::new_foreign_array;
35use crate::buffer::BufferHandle;
36use crate::dtype::DType;
37use crate::dtype::TryFromBytes;
38use crate::session::ArraySessionExt;
39use crate::stats::StatsSet;
40
41/// Options for serializing an array.
42#[derive(Default, Debug)]
43pub struct SerializeOptions {
44    /// The starting position within an external stream or file. This offset is used to compute
45    /// appropriate padding to enable zero-copy reads.
46    pub offset: usize,
47    /// Whether to include sufficient zero-copy padding.
48    pub include_padding: bool,
49}
50
51impl ArrayRef {
52    /// Serialize the array into a sequence of byte buffers that should be written contiguously.
53    /// This function returns a vec to avoid copying data buffers.
54    ///
55    /// Optionally, padding can be included to guarantee buffer alignment and ensure zero-copy
56    /// reads within the context of an external file or stream. In this case, the alignment of
57    /// the first byte buffer should be respected when writing the buffers to the stream or file.
58    ///
59    /// The format of this blob is a sequence of data buffers, possible with prefixed padding,
60    /// followed by a flatbuffer containing an [`fba::Array`] message, and ending with a
61    /// little-endian u32 describing the length of the flatbuffer message.
62    pub fn serialize(
63        &self,
64        ctx: &ArrayContext,
65        session: &VortexSession,
66        options: &SerializeOptions,
67    ) -> VortexResult<Vec<ByteBuffer>> {
68        // Collect all array buffers
69        let array_buffers = self
70            .depth_first_traversal()
71            .flat_map(|f| f.buffers())
72            .collect::<Vec<_>>();
73
74        // Allocate result buffers, including a possible padding buffer for each.
75        let mut buffers = vec![];
76        let mut fb_buffers = Vec::with_capacity(buffers.capacity());
77
78        // If we're including padding, we need to find the maximum required buffer alignment.
79        let max_alignment = array_buffers
80            .iter()
81            .map(|buf| buf.alignment())
82            .chain(iter::once(FlatBuffer::alignment()))
83            .max()
84            .unwrap_or_else(FlatBuffer::alignment);
85
86        // Create a shared buffer of zeros we can use for padding
87        let zeros = ByteBuffer::zeroed(*max_alignment);
88
89        // We push an empty buffer with the maximum alignment, so then subsequent buffers
90        // will be aligned. For subsequent buffers, we always push a 1-byte alignment.
91        buffers.push(ByteBuffer::zeroed_aligned(0, max_alignment));
92
93        // Keep track of where we are in the "file" to calculate padding.
94        let mut pos = options.offset;
95
96        // Push all the array buffers with padding as necessary.
97        for buffer in array_buffers {
98            let padding = if options.include_padding {
99                let padding = pos.next_multiple_of(*buffer.alignment()) - pos;
100                if padding > 0 {
101                    pos += padding;
102                    buffers.push(zeros.slice(0..padding));
103                }
104                padding
105            } else {
106                0
107            };
108
109            fb_buffers.push(fba::Buffer::new(
110                u16::try_from(padding).vortex_expect("padding fits into u16"),
111                buffer.alignment().exponent(),
112                Compression::None,
113                u32::try_from(buffer.len())
114                    .map_err(|_| vortex_err!("All buffers must fit into u32 for serialization"))?,
115            ));
116
117            pos += buffer.len();
118            buffers.push(buffer.aligned(Alignment::none()));
119        }
120
121        // Set up the flatbuffer builder
122        let mut fbb = FlatBufferBuilder::new();
123
124        let root = ArrayNodeFlatBuffer::try_new(ctx, session, self)?;
125        let fb_root = root.try_write_flatbuffer(&mut fbb)?;
126
127        let fb_buffers = fbb.create_vector(&fb_buffers);
128        let fb_array = fba::Array::create(
129            &mut fbb,
130            &fba::ArrayArgs {
131                root: Some(fb_root),
132                buffers: Some(fb_buffers),
133            },
134        );
135        fbb.finish_minimal(fb_array);
136        let (fb_vec, fb_start) = fbb.collapse();
137        let fb_end = fb_vec.len();
138        let fb_buffer = ByteBuffer::from(fb_vec).slice(fb_start..fb_end);
139        let fb_length = fb_buffer.len();
140
141        if options.include_padding {
142            let padding = pos.next_multiple_of(*FlatBuffer::alignment()) - pos;
143            if padding > 0 {
144                buffers.push(zeros.slice(0..padding));
145            }
146        }
147        buffers.push(fb_buffer);
148
149        // Finally, we write down the u32 length for the flatbuffer.
150        buffers.push(ByteBuffer::from(
151            u32::try_from(fb_length)
152                .map_err(|_| vortex_err!("Array metadata flatbuffer must fit into u32 for serialization. Array encoding tree is too large."))?
153                .to_le_bytes()
154                .to_vec(),
155        ));
156
157        Ok(buffers)
158    }
159}
160
161/// A utility struct for creating an [`fba::ArrayNode`] flatbuffer.
162pub struct ArrayNodeFlatBuffer<'a> {
163    ctx: &'a ArrayContext,
164    session: &'a VortexSession,
165    array: &'a ArrayRef,
166    buffer_idx: u16,
167}
168
169impl<'a> ArrayNodeFlatBuffer<'a> {
170    pub fn try_new(
171        ctx: &'a ArrayContext,
172        session: &'a VortexSession,
173        array: &'a ArrayRef,
174    ) -> VortexResult<Self> {
175        let n_buffers_recursive = array.nbuffers_recursive();
176        if n_buffers_recursive > u16::MAX as usize {
177            vortex_bail!(
178                "Array and all descendent arrays can have at most u16::MAX buffers: {}",
179                n_buffers_recursive
180            );
181        };
182        Ok(Self {
183            ctx,
184            session,
185            array,
186            buffer_idx: 0,
187        })
188    }
189
190    pub fn try_write_flatbuffer<'fb>(
191        &self,
192        fbb: &mut FlatBufferBuilder<'fb>,
193    ) -> VortexResult<WIPOffset<fba::ArrayNode<'fb>>> {
194        let encoding_idx = self
195            .ctx
196            .intern(&self.array.encoding_id())
197            // TODO(ngates): write_flatbuffer should return a result if this can fail.
198            .ok_or_else(|| {
199                vortex_err!(
200                    "Array encoding {} not permitted by ctx",
201                    self.array.encoding_id()
202                )
203            })?;
204
205        let metadata_bytes = self.session.array_serialize(self.array)?.ok_or_else(|| {
206            vortex_err!(
207                "Array {} does not support serialization",
208                self.array.encoding_id()
209            )
210        })?;
211        let metadata = Some(fbb.create_vector(metadata_bytes.as_slice()));
212
213        // Assign buffer indices for all child arrays.
214        let nbuffers = u16::try_from(self.array.nbuffers())
215            .map_err(|_| vortex_err!("Array can have at most u16::MAX buffers"))?;
216        let mut child_buffer_idx = self.buffer_idx + nbuffers;
217
218        let children = self
219            .array
220            .children()
221            .iter()
222            .map(|child| {
223                // Update the number of buffers required.
224                let msg = ArrayNodeFlatBuffer {
225                    ctx: self.ctx,
226                    session: self.session,
227                    array: child,
228                    buffer_idx: child_buffer_idx,
229                }
230                .try_write_flatbuffer(fbb)?;
231
232                child_buffer_idx = u16::try_from(child.nbuffers_recursive())
233                    .ok()
234                    .and_then(|nbuffers| nbuffers.checked_add(child_buffer_idx))
235                    .ok_or_else(|| vortex_err!("Too many buffers (u16) for Array"))?;
236
237                Ok(msg)
238            })
239            .collect::<VortexResult<Vec<_>>>()?;
240        let children = Some(fbb.create_vector(&children));
241
242        let buffers = Some(fbb.create_vector_from_iter((0..nbuffers).map(|i| i + self.buffer_idx)));
243        let stats = Some(self.array.statistics().write_flatbuffer(fbb)?);
244
245        Ok(fba::ArrayNode::create(
246            fbb,
247            &fba::ArrayNodeArgs {
248                encoding: encoding_idx,
249                metadata,
250                children,
251                buffers,
252                stats,
253            },
254        ))
255    }
256}
257
258/// To minimize the serialized form, arrays do not persist their own dtype and length. Instead,
259/// parent arrays pass this information down during deserialization.
260pub trait ArrayChildren {
261    /// Returns the nth child of the array with the given dtype and length.
262    fn get(&self, index: usize, dtype: &DType, len: usize) -> VortexResult<ArrayRef>;
263
264    /// The number of children.
265    fn len(&self) -> usize;
266
267    /// Returns true if there are no children.
268    fn is_empty(&self) -> bool {
269        self.len() == 0
270    }
271}
272
273impl<T: AsRef<[ArrayRef]>> ArrayChildren for T {
274    fn get(&self, index: usize, dtype: &DType, len: usize) -> VortexResult<ArrayRef> {
275        let array = self.as_ref()[index].clone();
276        assert_eq!(array.len(), len);
277        assert_eq!(array.dtype(), dtype);
278        Ok(array)
279    }
280
281    fn len(&self) -> usize {
282        self.as_ref().len()
283    }
284}
285
286/// [`SerializedArray`] represents a parsed but not-yet-decoded deserialized array.
287/// It contains all the information from the serialized form, without anything extra. i.e.
288/// it is missing a [`DType`] and `len`, and the `encoding_id` is not yet resolved to a concrete
289/// vtable.
290///
291/// An [`SerializedArray`] can be fully decoded into an [`ArrayRef`] using the `decode` function.
292#[derive(Clone)]
293pub struct SerializedArray {
294    // Typed as fb::ArrayNode
295    flatbuffer: FlatBuffer,
296    // The location of the current fb::ArrayNode
297    flatbuffer_loc: usize,
298    buffers: Arc<[BufferHandle]>,
299}
300
301impl Debug for SerializedArray {
302    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
303        f.debug_struct("SerializedArray")
304            .field("encoding_id", &self.encoding_id())
305            .field("children", &(0..self.nchildren()).map(|i| self.child(i)))
306            .field(
307                "buffers",
308                &(0..self.nbuffers()).map(|i| self.buffer(i).ok()),
309            )
310            .field("metadata", &self.metadata())
311            .finish()
312    }
313}
314
315impl SerializedArray {
316    /// Decode an [`SerializedArray`] into an [`ArrayRef`].
317    pub fn decode(
318        &self,
319        dtype: &DType,
320        len: usize,
321        ctx: &ReadContext,
322        session: &VortexSession,
323    ) -> VortexResult<ArrayRef> {
324        let encoding_idx = self.flatbuffer().encoding();
325        let encoding_id = ctx
326            .resolve(encoding_idx)
327            .ok_or_else(|| vortex_err!("Unknown encoding index: {}", encoding_idx))?;
328        let Some(plugin) = session.arrays().registry().find(&encoding_id) else {
329            if session.allows_unknown() {
330                return self.decode_foreign(encoding_id, dtype, len, ctx);
331            }
332            vortex_bail!("Unknown encoding: {}", encoding_id);
333        };
334
335        let children = SerializedArrayChildren {
336            ser: self,
337            ctx,
338            session,
339        };
340
341        let buffers = self.collect_buffers()?;
342
343        let decoded =
344            plugin.deserialize(dtype, len, self.metadata(), &buffers, &children, session)?;
345
346        assert_eq!(
347            decoded.len(),
348            len,
349            "Array decoded from {} has incorrect length {}, expected {}",
350            encoding_id,
351            decoded.len(),
352            len
353        );
354        assert_eq!(
355            decoded.dtype(),
356            dtype,
357            "Array decoded from {} has incorrect dtype {}, expected {}",
358            encoding_id,
359            decoded.dtype(),
360            dtype,
361        );
362
363        assert!(
364            plugin.is_supported_encoding(&decoded.encoding_id()),
365            "Array decoded from {} has incorrect encoding {}",
366            encoding_id,
367            decoded.encoding_id(),
368        );
369
370        // Populate statistics from the serialized array.
371        if let Some(stats) = self.flatbuffer().stats() {
372            decoded
373                .statistics()
374                .set_iter(StatsSet::from_flatbuffer(&stats, dtype, session)?.into_iter());
375        }
376
377        Ok(decoded)
378    }
379
380    fn decode_foreign(
381        &self,
382        encoding_id: ArrayId,
383        dtype: &DType,
384        len: usize,
385        ctx: &ReadContext,
386    ) -> VortexResult<ArrayRef> {
387        let children = (0..self.nchildren())
388            .map(|idx| {
389                let child = self.child(idx);
390                let child_encoding_idx = child.flatbuffer().encoding();
391                let child_encoding_id = ctx
392                    .resolve(child_encoding_idx)
393                    .ok_or_else(|| vortex_err!("Unknown encoding index: {}", child_encoding_idx))?;
394                child
395                    .decode_foreign(child_encoding_id, dtype, len, ctx)
396                    .map(Some)
397            })
398            .collect::<VortexResult<ArraySlots>>()?;
399
400        new_foreign_array(
401            encoding_id,
402            dtype.clone(),
403            len,
404            self.metadata().to_vec(),
405            self.collect_buffers()?.into_owned(),
406            children,
407        )
408    }
409
410    /// Returns the array encoding.
411    pub fn encoding_id(&self) -> u16 {
412        self.flatbuffer().encoding()
413    }
414
415    /// Returns the array metadata bytes.
416    pub fn metadata(&self) -> &[u8] {
417        self.flatbuffer()
418            .metadata()
419            .map(|metadata| metadata.bytes())
420            .unwrap_or(&[])
421    }
422
423    /// Returns the number of children.
424    pub fn nchildren(&self) -> usize {
425        self.flatbuffer()
426            .children()
427            .map_or(0, |children| children.len())
428    }
429
430    /// Returns the nth child of the array.
431    pub fn child(&self, idx: usize) -> SerializedArray {
432        let children = self
433            .flatbuffer()
434            .children()
435            .vortex_expect("Expected array to have children");
436        if idx >= children.len() {
437            vortex_panic!(
438                "Invalid child index {} for array with {} children",
439                idx,
440                children.len()
441            );
442        }
443        self.with_root(children.get(idx))
444    }
445
446    /// Returns the number of buffers.
447    pub fn nbuffers(&self) -> usize {
448        self.flatbuffer()
449            .buffers()
450            .map_or(0, |buffers| buffers.len())
451    }
452
453    /// Returns the nth buffer of the current array.
454    pub fn buffer(&self, idx: usize) -> VortexResult<BufferHandle> {
455        let buffer_idx = self
456            .flatbuffer()
457            .buffers()
458            .ok_or_else(|| vortex_err!("Array has no buffers"))?
459            .get(idx);
460        self.buffers
461            .get(buffer_idx as usize)
462            .cloned()
463            .ok_or_else(|| {
464                vortex_err!(
465                    "Invalid buffer index {} for array with {} buffers",
466                    buffer_idx,
467                    self.nbuffers()
468                )
469            })
470    }
471
472    /// Returns all buffers for the current array node.
473    ///
474    /// If buffer indices are contiguous, returns a zero-copy borrowed slice.
475    /// Otherwise falls back to collecting each buffer individually.
476    fn collect_buffers(&self) -> VortexResult<Cow<'_, [BufferHandle]>> {
477        let Some(fb_buffers) = self.flatbuffer().buffers() else {
478            return Ok(Cow::Borrowed(&[]));
479        };
480        let count = fb_buffers.len();
481        if count == 0 {
482            return Ok(Cow::Borrowed(&[]));
483        }
484        let start = fb_buffers.get(0) as usize;
485        let contiguous = fb_buffers
486            .iter()
487            .enumerate()
488            .all(|(i, idx)| idx as usize == start + i);
489        if contiguous {
490            self.buffers.get(start..start + count).map_or_else(
491                || {
492                    vortex_bail!(
493                        "buffer indices {}..{} out of range for {} buffers",
494                        start,
495                        start + count,
496                        self.buffers.len()
497                    )
498                },
499                |slice| Ok(Cow::Borrowed(slice)),
500            )
501        } else {
502            (0..count)
503                .map(|idx| self.buffer(idx))
504                .collect::<VortexResult<Vec<_>>>()
505                .map(Cow::Owned)
506        }
507    }
508
509    /// Returns the buffer lengths as stored in the flatbuffer metadata.
510    ///
511    /// This reads the buffer descriptors from the flatbuffer, which contain the
512    /// serialized length of each buffer. This is useful for displaying buffer sizes
513    /// without needing to access the actual buffer data.
514    pub fn buffer_lengths(&self) -> Vec<usize> {
515        let fb_array = root::<fba::Array>(self.flatbuffer.as_ref())
516            .vortex_expect("SerializedArray flatbuffer must be a valid Array");
517        fb_array
518            .buffers()
519            .map(|buffers| buffers.iter().map(|b| b.length() as usize).collect())
520            .unwrap_or_default()
521    }
522
523    /// Validate and align the array tree flatbuffer, returning the aligned buffer and root location.
524    fn validate_array_tree(array_tree: impl Into<ByteBuffer>) -> VortexResult<(FlatBuffer, usize)> {
525        let fb_buffer = FlatBuffer::align_from(array_tree.into());
526        let fb_array = root::<fba::Array>(fb_buffer.as_ref())?;
527        let fb_root = fb_array
528            .root()
529            .ok_or_else(|| vortex_err!("Array must have a root node"))?;
530        let flatbuffer_loc = fb_root._tab.loc();
531        Ok((fb_buffer, flatbuffer_loc))
532    }
533
534    /// Create an [`SerializedArray`] from a pre-existing array tree flatbuffer and pre-resolved buffer
535    /// handles.
536    ///
537    /// The caller is responsible for resolving buffers from whatever source (device segments, host
538    /// overrides, or a mix). The buffers must be in the same order as the `Array.buffers` descriptor
539    /// list in the flatbuffer.
540    pub fn from_flatbuffer_with_buffers(
541        array_tree: impl Into<ByteBuffer>,
542        buffers: Vec<BufferHandle>,
543    ) -> VortexResult<Self> {
544        let (flatbuffer, flatbuffer_loc) = Self::validate_array_tree(array_tree)?;
545        Ok(SerializedArray {
546            flatbuffer,
547            flatbuffer_loc,
548            buffers: buffers.into(),
549        })
550    }
551
552    /// Create an [`SerializedArray`] from a raw array tree flatbuffer (metadata only).
553    ///
554    /// This constructor creates a `SerializedArray` with no buffer data, useful for
555    /// inspecting the metadata when the actual buffer data is not needed
556    /// (e.g., displaying buffer sizes from inlined array tree metadata).
557    ///
558    /// Note: Calling `buffer()` on the returned `SerializedArray` will fail since
559    /// no actual buffer data is available.
560    pub fn from_array_tree(array_tree: impl Into<ByteBuffer>) -> VortexResult<Self> {
561        let (flatbuffer, flatbuffer_loc) = Self::validate_array_tree(array_tree)?;
562        Ok(SerializedArray {
563            flatbuffer,
564            flatbuffer_loc,
565            buffers: Arc::new([]),
566        })
567    }
568
569    /// Returns the root ArrayNode flatbuffer.
570    fn flatbuffer(&self) -> fba::ArrayNode<'_> {
571        unsafe { fba::ArrayNode::follow(self.flatbuffer.as_ref(), self.flatbuffer_loc) }
572    }
573
574    /// Returns a new [`SerializedArray`] with the given node as the root
575    // TODO(ngates): we may want a wrapper that avoids this clone.
576    fn with_root(&self, root: fba::ArrayNode) -> Self {
577        let mut this = self.clone();
578        this.flatbuffer_loc = root._tab.loc();
579        this
580    }
581
582    /// Create an [`SerializedArray`] from a pre-existing flatbuffer (ArrayNode) and a segment containing
583    /// only the data buffers (without the flatbuffer suffix).
584    ///
585    /// This is used when the flatbuffer is stored separately in layout metadata (e.g., when
586    /// `FLAT_LAYOUT_INLINE_ARRAY_NODE` is enabled).
587    pub fn from_flatbuffer_and_segment(
588        array_tree: ByteBuffer,
589        segment: BufferHandle,
590    ) -> VortexResult<Self> {
591        // HashMap::new doesn't allocate when empty, so this has no overhead
592        Self::from_flatbuffer_and_segment_with_overrides(array_tree, segment, &HashMap::new())
593    }
594
595    /// Create an [`SerializedArray`] from a pre-existing flatbuffer (ArrayNode) and a segment,
596    /// substituting host-resident buffer overrides for specific buffer indices.
597    ///
598    /// Buffers whose index appears in `buffer_overrides` are resolved from the provided
599    /// host data instead of the segment. All other buffers are sliced from the segment
600    /// using the padding and alignment described in the flatbuffer.
601    pub fn from_flatbuffer_and_segment_with_overrides(
602        array_tree: ByteBuffer,
603        segment: BufferHandle,
604        buffer_overrides: &HashMap<u32, ByteBuffer>,
605    ) -> VortexResult<Self> {
606        // We align each buffer individually, so we remove alignment requirements on the segment
607        // for host-resident buffers. Device buffers are sliced directly.
608        let segment = segment.ensure_aligned(Alignment::none())?;
609
610        // this can't return the validated array because there is no lifetime to give it, so we
611        // need to cast it below, which is safe.
612        let (fb_buffer, flatbuffer_loc) = Self::validate_array_tree(array_tree)?;
613        // SAFETY: fb_buffer was already validated by validate_array_tree above.
614        let fb_array = unsafe { fba::root_as_array_unchecked(fb_buffer.as_ref()) };
615
616        let mut offset = 0usize;
617        let buffers = fb_array
618            .buffers()
619            .unwrap_or_default()
620            .iter()
621            .enumerate()
622            .map(|(idx, fb_buf)| {
623                let idx = u32::try_from(idx).vortex_expect("buffer count must fit in u32");
624
625                // The padding, length, and resulting offsets all come from the flatbuffer, which
626                // may be corrupt. Use checked arithmetic so malformed metadata returns a
627                // `VortexError` rather than panicking (see issue #8819).
628                let buffer_len = fb_buf.length() as usize;
629                let start = offset
630                    .checked_add(fb_buf.padding() as usize)
631                    .ok_or_else(|| {
632                        vortex_err!("Buffer {idx} offset overflows when adding its padding")
633                    })?;
634                let end = start.checked_add(buffer_len).ok_or_else(|| {
635                    vortex_err!("Buffer {idx} offset overflows when adding its length")
636                })?;
637
638                // The alignment exponent comes from the flatbuffer and may be corrupt, so validate
639                // it rather than panicking on a too-large shift (see issue #8819).
640                let alignment = Alignment::try_from_exponent(fb_buf.alignment_exponent())?;
641                let handle = if let Some(host_data) = buffer_overrides.get(&idx) {
642                    BufferHandle::new_host(host_data.clone()).ensure_aligned(alignment)?
643                } else {
644                    // Bounds-check against the segment so an out-of-range buffer returns a
645                    // `VortexError` rather than panicking when slicing (see issue #8819).
646                    if end > segment.len() {
647                        vortex_bail!(
648                            "Buffer {idx} at offset {start} with length {buffer_len} is out of \
649                             bounds of the {}-byte segment",
650                            segment.len(),
651                        );
652                    }
653                    segment.slice(start..end).ensure_aligned(alignment)?
654                };
655
656                offset = end;
657                Ok(handle)
658            })
659            .collect::<VortexResult<Arc<[_]>>>()?;
660
661        Ok(SerializedArray {
662            flatbuffer: fb_buffer,
663            flatbuffer_loc,
664            buffers,
665        })
666    }
667}
668
669struct SerializedArrayChildren<'a> {
670    ser: &'a SerializedArray,
671    ctx: &'a ReadContext,
672    session: &'a VortexSession,
673}
674
675impl ArrayChildren for SerializedArrayChildren<'_> {
676    fn get(&self, index: usize, dtype: &DType, len: usize) -> VortexResult<ArrayRef> {
677        self.ser
678            .child(index)
679            .decode(dtype, len, self.ctx, self.session)
680    }
681
682    fn len(&self) -> usize {
683        self.ser.nchildren()
684    }
685}
686
687impl TryFrom<ByteBuffer> for SerializedArray {
688    type Error = VortexError;
689
690    fn try_from(value: ByteBuffer) -> Result<Self, Self::Error> {
691        // The final 4 bytes contain the length of the flatbuffer.
692        if value.len() < 4 {
693            vortex_bail!("SerializedArray buffer is too short");
694        }
695
696        // We align each buffer individually, so we remove alignment requirements on the buffer.
697        let value = value.aligned(Alignment::none());
698
699        let fb_length = u32::try_from_le_bytes(&value.as_slice()[value.len() - 4..])? as usize;
700        if value.len() < 4 + fb_length {
701            vortex_bail!("SerializedArray buffer is too short for flatbuffer");
702        }
703
704        let fb_offset = value.len() - 4 - fb_length;
705        let array_tree = value.slice(fb_offset..fb_offset + fb_length);
706        let segment = BufferHandle::new_host(value.slice(0..fb_offset));
707
708        Self::from_flatbuffer_and_segment(array_tree, segment)
709    }
710}
711
712impl TryFrom<BufferHandle> for SerializedArray {
713    type Error = VortexError;
714
715    fn try_from(value: BufferHandle) -> Result<Self, Self::Error> {
716        Self::try_from(value.try_to_host_sync()?)
717    }
718}
719
720#[cfg(test)]
721mod tests {
722    use vortex_buffer::ByteBufferMut;
723
724    use super::*;
725    use crate::IntoArray;
726    use crate::array_session;
727    use crate::arrays::PrimitiveArray;
728
729    /// A corrupt array tree can declare a buffer that extends past the backing segment. Slicing
730    /// such a buffer must return a [`VortexError`] rather than panicking (see issue #8819).
731    #[test]
732    fn from_flatbuffer_and_segment_rejects_out_of_bounds_buffer() -> VortexResult<()> {
733        let session = array_session();
734        let array_ctx = ArrayContext::empty();
735
736        // Serialize a simple array so we have a valid array tree flatbuffer whose declared buffer
737        // lengths describe the trailing data segment.
738        let serialized = PrimitiveArray::from_iter([1i32, 2, 3, 4])
739            .into_array()
740            .serialize(&array_ctx, &session, &SerializeOptions::default())?;
741
742        let mut concat = ByteBufferMut::empty();
743        for buf in serialized {
744            concat.extend_from_slice(buf.as_ref());
745        }
746        let value = concat.freeze().aligned(Alignment::none());
747
748        // Split the blob into the trailing flatbuffer and the leading data segment, mirroring
749        // `SerializedArray::try_from`.
750        let fb_length = u32::try_from_le_bytes(&value.as_slice()[value.len() - 4..])? as usize;
751        let fb_offset = value.len() - 4 - fb_length;
752        assert!(
753            fb_offset > 0,
754            "the array must have at least one data buffer"
755        );
756        let array_tree = value.slice(fb_offset..fb_offset + fb_length);
757
758        // Truncate the data segment by one byte so the declared buffer no longer fits.
759        let truncated = BufferHandle::new_host(value.slice(0..fb_offset - 1));
760
761        let Some(err) = SerializedArray::from_flatbuffer_and_segment(array_tree, truncated).err()
762        else {
763            vortex_bail!("out-of-bounds buffer must be rejected");
764        };
765        assert!(
766            err.to_string().contains("out of bounds"),
767            "unexpected error: {err}"
768        );
769
770        Ok(())
771    }
772}