Skip to main content

vortex_file/
open.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4use std::sync::Arc;
5
6use futures::executor::block_on;
7use vortex_array::dtype::DType;
8use vortex_array::memory::MemorySessionExt;
9use vortex_array::session::ArraySessionExt;
10use vortex_buffer::Alignment;
11use vortex_buffer::ByteBuffer;
12use vortex_error::VortexError;
13use vortex_error::VortexExpect;
14use vortex_error::VortexResult;
15use vortex_error::vortex_err;
16use vortex_io::VortexReadAt;
17use vortex_io::session::RuntimeSessionExt;
18use vortex_layout::segments::InstrumentedSegmentCache;
19use vortex_layout::segments::NoOpSegmentCache;
20use vortex_layout::segments::SegmentCache;
21use vortex_layout::segments::SegmentCacheSourceAdapter;
22use vortex_layout::segments::SegmentId;
23use vortex_layout::segments::SegmentSource;
24use vortex_layout::segments::SharedSegmentSource;
25use vortex_layout::session::LayoutSessionExt;
26use vortex_metrics::DefaultMetricsRegistry;
27use vortex_metrics::Label;
28use vortex_metrics::MetricsRegistry;
29use vortex_session::VortexSession;
30use vortex_utils::aliases::hash_map::HashMap;
31
32use crate::DeserializeStep;
33use crate::EOF_SIZE;
34use crate::MAX_POSTSCRIPT_SIZE;
35use crate::VortexFile;
36use crate::footer::Footer;
37use crate::segments::BufferSegmentSource;
38use crate::segments::FileSegmentSource;
39use crate::segments::InitialReadSegmentCache;
40use crate::segments::RequestMetrics;
41
42const INITIAL_READ_SIZE: usize = MAX_POSTSCRIPT_SIZE as usize + EOF_SIZE;
43
44struct FooterRead {
45    footer: Footer,
46    initial_segments: HashMap<SegmentId, ByteBuffer>,
47}
48
49/// Open options for a Vortex file reader.
50///
51/// Options are session-bound because opening a file may need array, layout, dtype, runtime, and
52/// memory registries. Construct with [`OpenOptionsSessionExt::open_options`] for the common path.
53///
54/// The opener first resolves a [`Footer`], then creates a segment source for later scans. Known
55/// file metadata can be supplied up front to avoid IO during footer discovery.
56#[derive(Clone)]
57pub struct VortexOpenOptions {
58    /// The session to use for opening the file.
59    session: VortexSession,
60    /// Cache to use for file segments.
61    segment_cache: Option<Arc<dyn SegmentCache>>,
62    /// The number of bytes to read when parsing the footer.
63    initial_read_size: usize,
64    /// An optional, externally provided, file size.
65    file_size: Option<u64>,
66    /// An optional, externally provided, DType.
67    dtype: Option<DType>,
68    /// An optional, externally provided, file layout.
69    footer: Option<Footer>,
70    /// Whether to include user-defined metadata segments when opening the file.
71    include_metadata: bool,
72    /// A metrics registry for the file.
73    metrics_registry: Option<Arc<dyn MetricsRegistry>>,
74    /// Default labels applied to all the file's metrics
75    labels: Vec<Label>,
76    /// Whether to cache file's LayoutReader between scans
77    cache_layout_reader: bool,
78}
79
80/// Extension trait for constructing [`VortexOpenOptions`] from a session.
81pub trait OpenOptionsSessionExt:
82    ArraySessionExt + LayoutSessionExt + RuntimeSessionExt + MemorySessionExt
83{
84    /// Create a new [`VortexOpenOptions`] using the provided session to open a file.
85    fn open_options(&self) -> VortexOpenOptions {
86        VortexOpenOptions {
87            session: self.session(),
88            segment_cache: None,
89            initial_read_size: INITIAL_READ_SIZE,
90            file_size: None,
91            dtype: None,
92            footer: None,
93            include_metadata: false,
94            metrics_registry: None,
95            labels: Vec::default(),
96            cache_layout_reader: false,
97        }
98    }
99}
100impl<S: ArraySessionExt + LayoutSessionExt + RuntimeSessionExt + MemorySessionExt>
101    OpenOptionsSessionExt for S
102{
103}
104
105impl VortexOpenOptions {
106    /// Return the session this opener is bound to.
107    pub fn session(&self) -> &VortexSession {
108        &self.session
109    }
110
111    /// Configure how many bytes to read from the end of the file before parsing the footer.
112    ///
113    /// The actual read is at least large enough to contain the maximum postscript and EOF marker,
114    /// and no larger than the file. Increase this when you expect footer segments to be near the
115    /// end and want to avoid a second footer read.
116    pub fn with_initial_read_size(mut self, initial_read_size: usize) -> Self {
117        self.initial_read_size = initial_read_size;
118        self
119    }
120
121    /// Cache the file's [`LayoutReader`](vortex_layout::LayoutReader) between scans.
122    ///
123    /// This avoids rebuilding the reader tree for repeated scans of the same [`VortexFile`], at the
124    /// cost of keeping reader state alive for the lifetime of the file handle.
125    pub fn with_layout_reader_cache(mut self) -> Self {
126        self.cache_layout_reader = true;
127        self
128    }
129
130    /// Configure a custom [`SegmentCache`].
131    ///
132    /// The cache is checked before the underlying file segment source. Segments covered by the
133    /// initial footer read are also inserted into an internal first-read cache.
134    pub fn with_segment_cache(mut self, segment_cache: Arc<dyn SegmentCache>) -> Self {
135        self.segment_cache = Some(segment_cache);
136        self
137    }
138
139    /// Disable the configured segment cache.
140    ///
141    /// This is useful when deriving an opener for a source whose buffers have different memory
142    /// placement requirements from the configured host cache.
143    pub fn without_segment_cache(mut self) -> Self {
144        self.segment_cache = None;
145        self
146    }
147
148    /// Configure a known file size.
149    ///
150    /// This helps to prevent an I/O request to discover the size of the file.
151    /// Of course, all bets are off if you pass an incorrect value.
152    pub fn with_file_size(mut self, file_size: u64) -> Self {
153        self.file_size = Some(file_size);
154        self
155    }
156
157    /// Configure a known file size.
158    ///
159    /// This helps to prevent an I/O request to discover the size of the file.
160    /// Of course, all bets are off if you pass an incorrect value.
161    pub fn with_some_file_size(mut self, file_size: Option<u64>) -> Self {
162        self.file_size = file_size;
163        self
164    }
165
166    /// Configure a known DType.
167    ///
168    /// If this is provided, then the Vortex file may be opened with fewer I/O requests.
169    ///
170    /// For Vortex files that do not contain a `DType`, this is required.
171    pub fn with_dtype(mut self, dtype: DType) -> Self {
172        self.dtype = Some(dtype);
173        self
174    }
175
176    /// Configure a known file footer.
177    ///
178    /// If this is provided, then the Vortex file can be opened without performing any I/O.
179    /// Once open, the [`Footer`] can be accessed via [`crate::VortexFile::footer`].
180    pub fn with_footer(mut self, footer: Footer) -> Self {
181        self.dtype = Some(footer.layout().dtype().clone());
182        self.footer = Some(footer);
183        self
184    }
185
186    /// Include user-defined metadata segments when opening the file.
187    ///
188    /// By default, opening a file reads only the metadata required to interpret the layout.
189    /// Enabling this option loads all metadata segments named by the postscript, which may require
190    /// additional reads before the footer is returned.
191    pub fn include_metadata(mut self) -> Self {
192        self.include_metadata = true;
193        self
194    }
195
196    /// Configure whether user-defined metadata segments are included when opening the file.
197    ///
198    /// Enabling this option loads all metadata segments named by the postscript, which may require
199    /// additional reads before the footer is returned.
200    pub fn with_include_metadata(mut self, include_metadata: bool) -> Self {
201        self.include_metadata = include_metadata;
202        self
203    }
204
205    /// Configure a custom [`MetricsRegistry`] implementation.
206    pub fn with_metrics_registry(mut self, metrics: Arc<dyn MetricsRegistry>) -> Self {
207        self.metrics_registry = Some(metrics);
208        self
209    }
210
211    /// Adds labels to all the file's metrics.
212    pub fn with_labels(mut self, labels: Vec<Label>) -> Self {
213        self.labels.extend(labels);
214        self
215    }
216
217    /// Open a Vortex file using the provided I/O source.
218    ///
219    /// This is the most common way to open a [`VortexFile`] and tends to provide the best
220    /// out-of-the-box performance. The underlying I/O system will continue to be optimised for
221    /// different file systems and object stores so we encourage users to use this method
222    /// whenever possible and file issues if they encounter problems.
223    pub async fn open(self, source: Arc<dyn VortexReadAt>) -> VortexResult<VortexFile> {
224        self.open_read(source).await
225    }
226
227    /// Open a Vortex file from a filesystem path.
228    #[cfg(not(target_arch = "wasm32"))]
229    pub async fn open_path(self, path: impl AsRef<std::path::Path>) -> VortexResult<VortexFile> {
230        use vortex_io::std_file::FileReadAt;
231        let handle = self.session.handle();
232        let allocator = self.session.allocator();
233        let source = Arc::new(FileReadAt::open_with_allocator(path, handle, allocator)?);
234        self.open(source).await
235    }
236
237    /// Open a Vortex file from an in-memory buffer.
238    ///
239    /// This uses a `BufferSegmentSource` that resolves segments synchronously
240    /// by slicing the buffer directly, bypassing the async I/O pipeline.
241    ///
242    /// Segment cache and metrics registry settings are ignored for this path.
243    pub fn open_buffer<B: Into<ByteBuffer>>(self, buffer: B) -> VortexResult<VortexFile> {
244        let buffer: ByteBuffer = buffer.into();
245
246        if self.segment_cache.is_some() {
247            tracing::warn!("segment cache is ignored for in-memory `open_buffer`");
248        }
249        if self.metrics_registry.is_some() {
250            tracing::warn!("metrics registry is ignored for in-memory `open_buffer`");
251        }
252
253        let cache_layout_reader = self.cache_layout_reader;
254        let include_metadata = self.include_metadata;
255        let mut opts = self.with_initial_read_size(0);
256
257        let footer = match opts.footer.take() {
258            Some(footer) => footer,
259            None => block_on(opts.read_footer(&buffer))?.footer,
260        };
261        footer.validate_file_size(buffer.len() as u64)?;
262
263        let segment_source: Arc<dyn SegmentSource> = Arc::new(BufferSegmentSource::new(
264            buffer,
265            footer.segment_specs_with_metadata(),
266        ));
267        let metadata = if include_metadata {
268            block_on(resolve_metadata(&footer, Arc::clone(&segment_source)))?
269        } else {
270            Arc::new(HashMap::new())
271        };
272        let file = VortexFile::new(footer, segment_source, opts.session).with_metadata(metadata);
273        Ok(if cache_layout_reader {
274            file.with_caching()
275        } else {
276            file
277        })
278    }
279
280    /// Open a [`VortexFile`] using any [`VortexReadAt`] implementation.
281    ///
282    /// This is the common path for files, object stores, and custom random-access sources.
283    pub async fn open_read<R: VortexReadAt + Clone>(self, reader: R) -> VortexResult<VortexFile> {
284        let segment_cache = self
285            .segment_cache
286            .clone()
287            .unwrap_or_else(|| Arc::new(NoOpSegmentCache));
288
289        let metrics_registry = self
290            .metrics_registry
291            .clone()
292            .unwrap_or_else(|| Arc::new(DefaultMetricsRegistry::default()));
293
294        let FooterRead {
295            footer,
296            initial_segments,
297        } = if let Some(footer) = self.footer {
298            if let Some(file_size) = self.file_size {
299                footer.validate_file_size(file_size)?;
300            }
301            FooterRead {
302                footer,
303                initial_segments: HashMap::default(),
304            }
305        } else {
306            self.read_footer(&reader).await?
307        };
308
309        let segment_cache = Arc::new(InstrumentedSegmentCache::new(
310            InitialReadSegmentCache {
311                initial: initial_segments,
312                fallback: segment_cache,
313            },
314            metrics_registry.as_ref(),
315            self.labels.clone(),
316        ));
317
318        let metrics = RequestMetrics::new(metrics_registry.as_ref(), self.labels);
319
320        // Create a segment source backed by the VortexRead implementation.
321        let segment_source = Arc::new(SharedSegmentSource::new(FileSegmentSource::open(
322            footer.segment_specs_with_metadata(),
323            reader,
324            self.session.handle(),
325            metrics,
326        )));
327
328        // Wrap up the segment source to first resolve segments from the initial read cache.
329        let segment_source: Arc<dyn SegmentSource> = Arc::new(SegmentCacheSourceAdapter::new(
330            segment_cache,
331            segment_source,
332        ));
333
334        let metadata = if self.include_metadata {
335            resolve_metadata(&footer, Arc::clone(&segment_source)).await?
336        } else {
337            Arc::new(HashMap::new())
338        };
339        let file =
340            VortexFile::new(footer, segment_source, self.session.clone()).with_metadata(metadata);
341        Ok(if self.cache_layout_reader {
342            file.with_caching()
343        } else {
344            file
345        })
346    }
347
348    async fn read_footer(&self, read: &dyn VortexReadAt) -> VortexResult<FooterRead> {
349        // Fetch the file size and perform the initial read.
350        let file_size = match self.file_size {
351            None => read.size().await?,
352            Some(file_size) => file_size,
353        };
354        let mut initial_read_size = self
355            .initial_read_size
356            // Make sure we read enough to cover the postscript
357            .max(MAX_POSTSCRIPT_SIZE as usize + EOF_SIZE);
358        if let Ok(file_size) = usize::try_from(file_size) {
359            initial_read_size = initial_read_size.min(file_size);
360        }
361
362        let initial_offset = file_size - initial_read_size as u64;
363        let initial_read: ByteBuffer = read
364            .read_at(initial_offset, initial_read_size, Alignment::none())
365            .await?
366            .try_into_host()?
367            .await?;
368
369        let mut deserializer = Footer::deserializer(initial_read, self.session.clone())
370            .with_size(file_size)
371            .with_some_dtype(self.dtype.clone());
372
373        let footer = loop {
374            match deserializer.deserialize()? {
375                DeserializeStep::NeedMoreData { offset, len } => {
376                    let more_data = read
377                        .read_at(offset, len, Alignment::none())
378                        .await?
379                        .try_into_host()?
380                        .await?;
381                    deserializer.prefix_data(more_data);
382                }
383                DeserializeStep::NeedFileSize => unreachable!("We passed file_size above"),
384                DeserializeStep::Done(footer) => break Ok::<_, VortexError>(footer),
385            }
386        }?;
387
388        // Segment specs describe the data file, not necessarily the byte stream used to
389        // deserialize a standalone cached footer. Validate them here, where we know this is the
390        // size of the actual data source (see issue #8819).
391        footer.validate_file_size(file_size)?;
392
393        // If the initial read happened to cover any segments, then we can populate the
394        // segment cache
395        let initial_offset = file_size - (deserializer.buffer().len() as u64);
396        let initial_segments =
397            Self::collect_initial_segments(initial_offset, deserializer.buffer(), &footer)?;
398
399        Ok(FooterRead {
400            footer,
401            initial_segments,
402        })
403    }
404
405    /// Collect segments that were covered by the initial read.
406    fn collect_initial_segments(
407        initial_offset: u64,
408        initial_read: &ByteBuffer,
409        footer: &Footer,
410    ) -> VortexResult<HashMap<SegmentId, ByteBuffer>> {
411        let mut initial_read_segments = HashMap::default();
412
413        // Iterate `segment_specs_with_metadata` (not just the segment map) so metadata segments
414        // covered by the initial read are cached too. Metadata segments are appended and not
415        // offset-sorted, so we skip per-segment rather than partition on offset.
416        for (idx, segment) in footer.segment_specs_with_metadata().iter().enumerate() {
417            if segment.offset < initial_offset {
418                continue;
419            }
420            let segment_id =
421                SegmentId::from(u32::try_from(idx).vortex_expect("Invalid segment ID"));
422            let offset =
423                usize::try_from(segment.offset - initial_offset).vortex_expect("Invalid offset");
424            // The segment map is validated against the file size before this method is called, but
425            // still bounds-check here so slicing never depends on that distant validation for panic
426            // safety (see issue #8819).
427            let end = offset
428                .checked_add(segment.length as usize)
429                .filter(|end| *end <= initial_read.len())
430                .ok_or_else(|| {
431                    vortex_err!(
432                        "Segment at offset {} with length {} is out of bounds of the \
433                         {}-byte initial read",
434                        segment.offset,
435                        segment.length,
436                        initial_read.len(),
437                    )
438                })?;
439            let buffer = initial_read.slice(offset..end).aligned(segment.alignment);
440            initial_read_segments.insert(segment_id, buffer);
441        }
442
443        Ok(initial_read_segments)
444    }
445}
446
447async fn resolve_metadata(
448    footer: &Footer,
449    segment_source: Arc<dyn SegmentSource>,
450) -> VortexResult<Arc<HashMap<String, ByteBuffer>>> {
451    let first_metadata_id = footer.segment_map().len();
452    let requests = footer
453        .metadata_segments()
454        .enumerate()
455        .map(|(index, (key, locator))| {
456            let id = u32::try_from(first_metadata_id + index).map(SegmentId::from);
457            let key = key.to_string();
458            let alignment = locator.alignment;
459            let segment_source = Arc::clone(&segment_source);
460            async move {
461                let handle = segment_source.request(id?).await?;
462                let buffer = handle.try_into_host()?.await?;
463                Ok::<_, VortexError>((
464                    key,
465                    ByteBuffer::copy_from_aligned(buffer.as_slice(), alignment),
466                ))
467            }
468        });
469    let metadata = futures::future::try_join_all(requests)
470        .await?
471        .into_iter()
472        .collect::<HashMap<_, _>>();
473
474    Ok(Arc::new(metadata))
475}
476
477#[cfg(feature = "object_store")]
478impl VortexOpenOptions {
479    /// Open a Vortex file from an `object_store` backend and path.
480    pub async fn open_object_store(
481        self,
482        object_store: &Arc<dyn object_store::ObjectStore>,
483        path: &str,
484    ) -> VortexResult<VortexFile> {
485        use vortex_io::object_store::ObjectStoreReadAt;
486
487        let handle = self.session.handle();
488        let allocator = self.session.allocator();
489        let source = Arc::new(ObjectStoreReadAt::new_with_allocator(
490            Arc::clone(object_store),
491            path.into(),
492            handle,
493            allocator,
494        ));
495        self.open(source).await
496    }
497}
498
499#[cfg(test)]
500mod tests {
501    use std::sync::atomic::AtomicUsize;
502    use std::sync::atomic::Ordering;
503
504    use futures::future::BoxFuture;
505    use parking_lot::Mutex;
506    use vortex_array::IntoArray;
507    use vortex_array::buffer::BufferHandle;
508    use vortex_array::memory::DefaultHostAllocator;
509    use vortex_array::memory::HostAllocator;
510    use vortex_array::memory::MemorySessionExt;
511    use vortex_array::memory::WritableHostBuffer;
512    use vortex_buffer::Alignment;
513    use vortex_buffer::Buffer;
514    use vortex_buffer::ByteBuffer;
515    use vortex_buffer::ByteBufferMut;
516    use vortex_error::vortex_bail;
517    use vortex_io::session::RuntimeSession;
518    use vortex_layout::session::LayoutSession;
519    use vortex_session::registry::Id;
520    use vortex_session::registry::ReadContext;
521
522    use super::*;
523    use crate::WriteOptionsSessionExt;
524    use crate::footer::SegmentSpec;
525
526    fn test_session() -> VortexSession {
527        let session = vortex_array::array_session()
528            .with::<LayoutSession>()
529            .with::<RuntimeSession>();
530        crate::register_default_encodings(&session);
531        crate::enable_all_registered_array_encodings(&session);
532        session
533    }
534
535    #[derive(Clone)]
536    // Define CountingRead struct
537    struct CountingRead<R> {
538        inner: R,
539        total_read: Arc<AtomicUsize>,
540        first_read_len: Arc<AtomicUsize>,
541        reads: Arc<Mutex<Vec<(u64, usize)>>>,
542    }
543
544    impl<R: VortexReadAt + Clone> VortexReadAt for CountingRead<R> {
545        fn size(&self) -> BoxFuture<'static, VortexResult<u64>> {
546            self.inner.size()
547        }
548
549        fn read_at(
550            &self,
551            offset: u64,
552            length: usize,
553            alignment: Alignment,
554        ) -> BoxFuture<'static, VortexResult<BufferHandle>> {
555            self.total_read.fetch_add(length, Ordering::Relaxed);
556            self.reads.lock().push((offset, length));
557            let _ = self.first_read_len.compare_exchange(
558                0,
559                length,
560                Ordering::Relaxed,
561                Ordering::Relaxed,
562            );
563            self.inner.read_at(offset, length, alignment)
564        }
565
566        fn concurrency(&self) -> usize {
567            self.inner.concurrency()
568        }
569    }
570
571    #[derive(Debug)]
572    struct CountingAllocator {
573        allocations: Arc<AtomicUsize>,
574    }
575
576    impl HostAllocator for CountingAllocator {
577        fn allocate(&self, len: usize, alignment: Alignment) -> VortexResult<WritableHostBuffer> {
578            self.allocations.fetch_add(1, Ordering::Relaxed);
579            DefaultHostAllocator.allocate(len, alignment)
580        }
581    }
582
583    #[tokio::test]
584    async fn test_initial_read_size() {
585        let session = vortex_array::array_session()
586            .with::<LayoutSession>()
587            .with::<RuntimeSession>();
588
589        crate::register_default_encodings(&session);
590        crate::enable_all_registered_array_encodings(&session);
591
592        // Create a large file (> 1MB)
593        let mut buf = ByteBufferMut::empty();
594
595        // 1.5M integers -> ~6MB. We use high-entropy (pseudo-random) values so the data does not
596        // compress well under any encoding (Sequence, RunEnd, Delta, ...), keeping the written
597        // file comfortably above 1MB.
598        let mut state = 0x9E37_79B9u32;
599        let array = Buffer::from(
600            (0i32..1_500_000)
601                .map(|_| {
602                    state = state.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
603                    state as i32
604                })
605                .collect::<Vec<i32>>(),
606        )
607        .into_array();
608
609        session
610            .write_options()
611            .write(&mut buf, array.to_array_stream())
612            .await
613            .unwrap();
614
615        let buffer = ByteBuffer::from(buf);
616        assert!(
617            buffer.len() > 1024 * 1024,
618            "Buffer length is only {} bytes",
619            buffer.len()
620        );
621
622        let total_read = Arc::new(AtomicUsize::new(0));
623        let first_read_len = Arc::new(AtomicUsize::new(0));
624        let reader = CountingRead {
625            inner: buffer,
626            total_read: Arc::clone(&total_read),
627            first_read_len: Arc::clone(&first_read_len),
628            reads: Arc::new(Mutex::new(Vec::new())),
629        };
630
631        // Open the file
632        let _file = session.open_options().open_read(reader).await.unwrap();
633
634        // Assert that we read approximately the postscript size, not 1MB
635        let first = first_read_len.load(Ordering::Relaxed);
636        assert_eq!(
637            first,
638            MAX_POSTSCRIPT_SIZE as usize + EOF_SIZE,
639            "Read exactly the postscript size"
640        );
641        let read = total_read.load(Ordering::Relaxed);
642        assert!(read < 1024 * 1024, "Read {} bytes, expected < 1MB", read);
643    }
644
645    #[tokio::test]
646    async fn test_metadata_outside_initial_read_uses_targeted_read() -> VortexResult<()> {
647        let session = vortex_array::array_session()
648            .with::<LayoutSession>()
649            .with::<RuntimeSession>();
650        crate::register_default_encodings(&session);
651        crate::enable_all_registered_array_encodings(&session);
652
653        let metadata = ByteBuffer::copy_from(vec![0x5a; INITIAL_READ_SIZE * 2]);
654        let mut output = ByteBufferMut::empty();
655        let summary = session
656            .write_options()
657            .with_metadata_segment("outside", metadata.clone())
658            .write(
659                &mut output,
660                Buffer::from(vec![1u32]).into_array().to_array_stream(),
661            )
662            .await?;
663        let locator = *summary
664            .footer()
665            .metadata_segment("outside")
666            .vortex_expect("metadata locator");
667        let bytes = ByteBuffer::from(output);
668        assert!(locator.offset < bytes.len() as u64 - INITIAL_READ_SIZE as u64);
669
670        let default_total = Arc::new(AtomicUsize::new(0));
671        let default_reads = Arc::new(Mutex::new(Vec::new()));
672        let default_reader = CountingRead {
673            inner: bytes.clone(),
674            total_read: Arc::clone(&default_total),
675            first_read_len: Arc::new(AtomicUsize::new(0)),
676            reads: Arc::clone(&default_reads),
677        };
678        let default_file = session.open_options().open_read(default_reader).await?;
679        assert!(default_file.metadata_segment("outside").is_none());
680        assert!(
681            !default_reads
682                .lock()
683                .contains(&(locator.offset, locator.length as usize))
684        );
685        // A default open must not amplify into the metadata: total bytes read stay
686        // below the metadata segment's size (itself 2x the initial read).
687        assert!(
688            default_total.load(Ordering::Relaxed) < metadata.len(),
689            "default open read {} bytes; metadata segment is {}",
690            default_total.load(Ordering::Relaxed),
691            metadata.len()
692        );
693
694        let metadata_reads = Arc::new(Mutex::new(Vec::new()));
695        let metadata_reader = CountingRead {
696            inner: bytes,
697            total_read: Arc::new(AtomicUsize::new(0)),
698            first_read_len: Arc::new(AtomicUsize::new(0)),
699            reads: Arc::clone(&metadata_reads),
700        };
701        let file = session
702            .open_options()
703            .include_metadata()
704            .open_read(metadata_reader)
705            .await?;
706        assert_eq!(
707            file.metadata_segment("outside").map(ByteBuffer::as_slice),
708            Some(metadata.as_slice())
709        );
710        assert!(
711            metadata_reads
712                .lock()
713                .contains(&(locator.offset, locator.length as usize))
714        );
715
716        Ok(())
717    }
718
719    #[tokio::test]
720    async fn test_with_footer_include_metadata_open_buffer_resolves() -> VortexResult<()> {
721        let session = vortex_array::array_session()
722            .with::<LayoutSession>()
723            .with::<RuntimeSession>();
724        crate::register_default_encodings(&session);
725        crate::enable_all_registered_array_encodings(&session);
726
727        let value = ByteBuffer::copy_from(b"supplied-footer metadata");
728        let mut output = ByteBufferMut::empty();
729        let summary = session
730            .write_options()
731            .with_metadata_segment("key", value.clone())
732            .write(
733                &mut output,
734                Buffer::from(vec![1u32]).into_array().to_array_stream(),
735            )
736            .await?;
737        let footer = summary.footer().clone();
738        let bytes = ByteBuffer::from(output);
739
740        let file = session
741            .open_options()
742            .with_footer(footer.clone())
743            .include_metadata()
744            .open_buffer(bytes.clone())?;
745        assert_eq!(
746            file.metadata_segment("key").map(ByteBuffer::as_slice),
747            Some(value.as_slice())
748        );
749
750        let default = session
751            .open_options()
752            .with_footer(footer)
753            .open_buffer(bytes)?;
754        assert!(default.metadata_segment("key").is_none());
755
756        Ok(())
757    }
758
759    #[cfg(not(target_arch = "wasm32"))]
760    #[tokio::test]
761    async fn test_open_path_uses_memory_session_allocator() {
762        let session = vortex_array::array_session()
763            .with::<LayoutSession>()
764            .with::<RuntimeSession>();
765
766        crate::register_default_encodings(&session);
767        crate::enable_all_registered_array_encodings(&session);
768
769        let mut buf = ByteBufferMut::empty();
770        let array = Buffer::from((0i32..16_384).collect::<Vec<i32>>()).into_array();
771        session
772            .write_options()
773            .write(&mut buf, array.to_array_stream())
774            .await
775            .unwrap();
776
777        let file_path = std::env::temp_dir().join(format!(
778            "vortex-open-memory-session-{}.vx",
779            std::process::id()
780        ));
781        std::fs::write(&file_path, ByteBuffer::from(buf).as_slice()).unwrap();
782
783        let allocations = Arc::new(AtomicUsize::new(0));
784        let session = session.with_allocator(Arc::new(CountingAllocator {
785            allocations: Arc::clone(&allocations),
786        }));
787
788        let _file = session.open_options().open_path(&file_path).await.unwrap();
789        std::fs::remove_file(&file_path).unwrap();
790
791        assert!(
792            allocations.load(Ordering::Relaxed) > 0,
793            "expected at least one host allocation from MemorySession"
794        );
795    }
796
797    /// `collect_initial_segments` must bounds-check the segment map against the initial read rather
798    /// than slicing unchecked, so a segment larger than the read returns an error (see issue #8819).
799    #[tokio::test]
800    async fn collect_initial_segments_rejects_out_of_bounds_segment() -> VortexResult<()> {
801        let session = test_session();
802
803        // A valid root layout is obtained by writing and parsing a small file.
804        // `collect_initial_segments` only consults the segment map, so the layout is irrelevant.
805        let mut buf = ByteBufferMut::empty();
806        let array = Buffer::from((0i32..16).collect::<Vec<i32>>()).into_array();
807        session
808            .write_options()
809            .write(&mut buf, array.to_array_stream())
810            .await?;
811        let footer = session
812            .open_options()
813            .read_footer(&ByteBuffer::from(buf))
814            .await?
815            .footer;
816
817        // Build a footer whose sole segment is far larger than the initial read below.
818        let bad_segments: Arc<[SegmentSpec]> = Arc::from([SegmentSpec {
819            offset: 0,
820            length: 1024,
821            alignment: Alignment::none(),
822        }]);
823        let bad_footer = Footer::new(
824            Arc::clone(footer.layout()),
825            bad_segments,
826            None,
827            ReadContext::new(Vec::<Id>::new()),
828        );
829
830        let initial_read = ByteBuffer::zeroed(16);
831        let Err(err) = VortexOpenOptions::collect_initial_segments(0, &initial_read, &bad_footer)
832        else {
833            vortex_bail!("collecting an out-of-bounds segment must return an error");
834        };
835        assert!(
836            err.to_string().contains("out of bounds"),
837            "unexpected error: {err}"
838        );
839
840        Ok(())
841    }
842
843    #[tokio::test]
844    async fn standalone_footer_round_trip() -> VortexResult<()> {
845        let session = test_session();
846
847        let mut file_bytes = ByteBufferMut::empty();
848        let values = (0i32..65_536)
849            .map(|value| value.wrapping_mul(1_664_525).wrapping_add(1_013_904_223))
850            .collect::<Vec<_>>();
851        let array = Buffer::from(values).into_array();
852        session
853            .write_options()
854            .write(&mut file_bytes, array.to_array_stream())
855            .await?;
856        let file_bytes = ByteBuffer::from(file_bytes);
857
858        let footer = session
859            .open_options()
860            .open_buffer(file_bytes.clone())?
861            .footer()
862            .clone();
863        let last_segment_end = footer
864            .segment_map()
865            .iter()
866            .map(|segment| segment.offset + u64::from(segment.length))
867            .max()
868            .unwrap_or_default();
869        let serialized_footer = footer.into_serializer().serialize()?;
870        let serialized_footer_size = serialized_footer.iter().map(ByteBuffer::len).sum();
871        assert!(last_segment_end > serialized_footer_size as u64);
872        let mut footer_bytes = ByteBufferMut::with_capacity(serialized_footer_size);
873        for buffer in serialized_footer {
874            footer_bytes.extend_from_slice(&buffer);
875        }
876
877        let mut deserializer = Footer::deserializer(footer_bytes.freeze(), session.clone())
878            .with_size(serialized_footer_size as u64);
879        let DeserializeStep::Done(cached_footer) = deserializer.deserialize()? else {
880            vortex_bail!("standalone footer bytes must be sufficient for deserialization");
881        };
882
883        session
884            .open_options()
885            .with_footer(cached_footer)
886            .open_buffer(file_bytes)?;
887
888        Ok(())
889    }
890}