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        session
532    }
533
534    #[derive(Clone)]
535    // Define CountingRead struct
536    struct CountingRead<R> {
537        inner: R,
538        total_read: Arc<AtomicUsize>,
539        first_read_len: Arc<AtomicUsize>,
540        reads: Arc<Mutex<Vec<(u64, usize)>>>,
541    }
542
543    impl<R: VortexReadAt + Clone> VortexReadAt for CountingRead<R> {
544        fn size(&self) -> BoxFuture<'static, VortexResult<u64>> {
545            self.inner.size()
546        }
547
548        fn read_at(
549            &self,
550            offset: u64,
551            length: usize,
552            alignment: Alignment,
553        ) -> BoxFuture<'static, VortexResult<BufferHandle>> {
554            self.total_read.fetch_add(length, Ordering::Relaxed);
555            self.reads.lock().push((offset, length));
556            let _ = self.first_read_len.compare_exchange(
557                0,
558                length,
559                Ordering::Relaxed,
560                Ordering::Relaxed,
561            );
562            self.inner.read_at(offset, length, alignment)
563        }
564
565        fn concurrency(&self) -> usize {
566            self.inner.concurrency()
567        }
568    }
569
570    #[derive(Debug)]
571    struct CountingAllocator {
572        allocations: Arc<AtomicUsize>,
573    }
574
575    impl HostAllocator for CountingAllocator {
576        fn allocate(&self, len: usize, alignment: Alignment) -> VortexResult<WritableHostBuffer> {
577            self.allocations.fetch_add(1, Ordering::Relaxed);
578            DefaultHostAllocator.allocate(len, alignment)
579        }
580    }
581
582    #[tokio::test]
583    async fn test_initial_read_size() {
584        let session = vortex_array::array_session()
585            .with::<LayoutSession>()
586            .with::<RuntimeSession>();
587
588        crate::register_default_encodings(&session);
589
590        // Create a large file (> 1MB)
591        let mut buf = ByteBufferMut::empty();
592
593        // 1.5M integers -> ~6MB. We use high-entropy (pseudo-random) values so the data does not
594        // compress well under any encoding (Sequence, RunEnd, Delta, ...), keeping the written
595        // file comfortably above 1MB.
596        let mut state = 0x9E37_79B9u32;
597        let array = Buffer::from(
598            (0i32..1_500_000)
599                .map(|_| {
600                    state = state.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
601                    state as i32
602                })
603                .collect::<Vec<i32>>(),
604        )
605        .into_array();
606
607        session
608            .write_options()
609            .write(&mut buf, array.to_array_stream())
610            .await
611            .unwrap();
612
613        let buffer = ByteBuffer::from(buf);
614        assert!(
615            buffer.len() > 1024 * 1024,
616            "Buffer length is only {} bytes",
617            buffer.len()
618        );
619
620        let total_read = Arc::new(AtomicUsize::new(0));
621        let first_read_len = Arc::new(AtomicUsize::new(0));
622        let reader = CountingRead {
623            inner: buffer,
624            total_read: Arc::clone(&total_read),
625            first_read_len: Arc::clone(&first_read_len),
626            reads: Arc::new(Mutex::new(Vec::new())),
627        };
628
629        // Open the file
630        let _file = session.open_options().open_read(reader).await.unwrap();
631
632        // Assert that we read approximately the postscript size, not 1MB
633        let first = first_read_len.load(Ordering::Relaxed);
634        assert_eq!(
635            first,
636            MAX_POSTSCRIPT_SIZE as usize + EOF_SIZE,
637            "Read exactly the postscript size"
638        );
639        let read = total_read.load(Ordering::Relaxed);
640        assert!(read < 1024 * 1024, "Read {} bytes, expected < 1MB", read);
641    }
642
643    #[tokio::test]
644    async fn test_metadata_outside_initial_read_uses_targeted_read() -> VortexResult<()> {
645        let session = vortex_array::array_session()
646            .with::<LayoutSession>()
647            .with::<RuntimeSession>();
648        crate::register_default_encodings(&session);
649
650        let metadata = ByteBuffer::copy_from(vec![0x5a; INITIAL_READ_SIZE * 2]);
651        let mut output = ByteBufferMut::empty();
652        let summary = session
653            .write_options()
654            .with_metadata_segment("outside", metadata.clone())
655            .write(
656                &mut output,
657                Buffer::from(vec![1u32]).into_array().to_array_stream(),
658            )
659            .await?;
660        let locator = *summary
661            .footer()
662            .metadata_segment("outside")
663            .vortex_expect("metadata locator");
664        let bytes = ByteBuffer::from(output);
665        assert!(locator.offset < bytes.len() as u64 - INITIAL_READ_SIZE as u64);
666
667        let default_total = Arc::new(AtomicUsize::new(0));
668        let default_reads = Arc::new(Mutex::new(Vec::new()));
669        let default_reader = CountingRead {
670            inner: bytes.clone(),
671            total_read: Arc::clone(&default_total),
672            first_read_len: Arc::new(AtomicUsize::new(0)),
673            reads: Arc::clone(&default_reads),
674        };
675        let default_file = session.open_options().open_read(default_reader).await?;
676        assert!(default_file.metadata_segment("outside").is_none());
677        assert!(
678            !default_reads
679                .lock()
680                .contains(&(locator.offset, locator.length as usize))
681        );
682        // A default open must not amplify into the metadata: total bytes read stay
683        // below the metadata segment's size (itself 2x the initial read).
684        assert!(
685            default_total.load(Ordering::Relaxed) < metadata.len(),
686            "default open read {} bytes; metadata segment is {}",
687            default_total.load(Ordering::Relaxed),
688            metadata.len()
689        );
690
691        let metadata_reads = Arc::new(Mutex::new(Vec::new()));
692        let metadata_reader = CountingRead {
693            inner: bytes,
694            total_read: Arc::new(AtomicUsize::new(0)),
695            first_read_len: Arc::new(AtomicUsize::new(0)),
696            reads: Arc::clone(&metadata_reads),
697        };
698        let file = session
699            .open_options()
700            .include_metadata()
701            .open_read(metadata_reader)
702            .await?;
703        assert_eq!(
704            file.metadata_segment("outside").map(ByteBuffer::as_slice),
705            Some(metadata.as_slice())
706        );
707        assert!(
708            metadata_reads
709                .lock()
710                .contains(&(locator.offset, locator.length as usize))
711        );
712
713        Ok(())
714    }
715
716    #[tokio::test]
717    async fn test_with_footer_include_metadata_open_buffer_resolves() -> VortexResult<()> {
718        let session = vortex_array::array_session()
719            .with::<LayoutSession>()
720            .with::<RuntimeSession>();
721        crate::register_default_encodings(&session);
722
723        let value = ByteBuffer::copy_from(b"supplied-footer metadata");
724        let mut output = ByteBufferMut::empty();
725        let summary = session
726            .write_options()
727            .with_metadata_segment("key", value.clone())
728            .write(
729                &mut output,
730                Buffer::from(vec![1u32]).into_array().to_array_stream(),
731            )
732            .await?;
733        let footer = summary.footer().clone();
734        let bytes = ByteBuffer::from(output);
735
736        let file = session
737            .open_options()
738            .with_footer(footer.clone())
739            .include_metadata()
740            .open_buffer(bytes.clone())?;
741        assert_eq!(
742            file.metadata_segment("key").map(ByteBuffer::as_slice),
743            Some(value.as_slice())
744        );
745
746        let default = session
747            .open_options()
748            .with_footer(footer)
749            .open_buffer(bytes)?;
750        assert!(default.metadata_segment("key").is_none());
751
752        Ok(())
753    }
754
755    #[cfg(not(target_arch = "wasm32"))]
756    #[tokio::test]
757    async fn test_open_path_uses_memory_session_allocator() {
758        let session = vortex_array::array_session()
759            .with::<LayoutSession>()
760            .with::<RuntimeSession>();
761
762        crate::register_default_encodings(&session);
763
764        let mut buf = ByteBufferMut::empty();
765        let array = Buffer::from((0i32..16_384).collect::<Vec<i32>>()).into_array();
766        session
767            .write_options()
768            .write(&mut buf, array.to_array_stream())
769            .await
770            .unwrap();
771
772        let file_path = std::env::temp_dir().join(format!(
773            "vortex-open-memory-session-{}.vx",
774            std::process::id()
775        ));
776        std::fs::write(&file_path, ByteBuffer::from(buf).as_slice()).unwrap();
777
778        let allocations = Arc::new(AtomicUsize::new(0));
779        let session = session.with_allocator(Arc::new(CountingAllocator {
780            allocations: Arc::clone(&allocations),
781        }));
782
783        let _file = session.open_options().open_path(&file_path).await.unwrap();
784        std::fs::remove_file(&file_path).unwrap();
785
786        assert!(
787            allocations.load(Ordering::Relaxed) > 0,
788            "expected at least one host allocation from MemorySession"
789        );
790    }
791
792    /// `collect_initial_segments` must bounds-check the segment map against the initial read rather
793    /// than slicing unchecked, so a segment larger than the read returns an error (see issue #8819).
794    #[tokio::test]
795    async fn collect_initial_segments_rejects_out_of_bounds_segment() -> VortexResult<()> {
796        let session = test_session();
797
798        // A valid root layout is obtained by writing and parsing a small file.
799        // `collect_initial_segments` only consults the segment map, so the layout is irrelevant.
800        let mut buf = ByteBufferMut::empty();
801        let array = Buffer::from((0i32..16).collect::<Vec<i32>>()).into_array();
802        session
803            .write_options()
804            .write(&mut buf, array.to_array_stream())
805            .await?;
806        let footer = session
807            .open_options()
808            .read_footer(&ByteBuffer::from(buf))
809            .await?
810            .footer;
811
812        // Build a footer whose sole segment is far larger than the initial read below.
813        let bad_segments: Arc<[SegmentSpec]> = Arc::from([SegmentSpec {
814            offset: 0,
815            length: 1024,
816            alignment: Alignment::none(),
817        }]);
818        let bad_footer = Footer::new(
819            Arc::clone(footer.layout()),
820            bad_segments,
821            None,
822            ReadContext::new(Vec::<Id>::new()),
823        );
824
825        let initial_read = ByteBuffer::zeroed(16);
826        let Err(err) = VortexOpenOptions::collect_initial_segments(0, &initial_read, &bad_footer)
827        else {
828            vortex_bail!("collecting an out-of-bounds segment must return an error");
829        };
830        assert!(
831            err.to_string().contains("out of bounds"),
832            "unexpected error: {err}"
833        );
834
835        Ok(())
836    }
837
838    #[tokio::test]
839    async fn standalone_footer_round_trip() -> VortexResult<()> {
840        let session = test_session();
841
842        let mut file_bytes = ByteBufferMut::empty();
843        let values = (0i32..65_536)
844            .map(|value| value.wrapping_mul(1_664_525).wrapping_add(1_013_904_223))
845            .collect::<Vec<_>>();
846        let array = Buffer::from(values).into_array();
847        session
848            .write_options()
849            .write(&mut file_bytes, array.to_array_stream())
850            .await?;
851        let file_bytes = ByteBuffer::from(file_bytes);
852
853        let footer = session
854            .open_options()
855            .open_buffer(file_bytes.clone())?
856            .footer()
857            .clone();
858        let last_segment_end = footer
859            .segment_map()
860            .iter()
861            .map(|segment| segment.offset + u64::from(segment.length))
862            .max()
863            .unwrap_or_default();
864        let serialized_footer = footer.into_serializer().serialize()?;
865        let serialized_footer_size = serialized_footer.iter().map(ByteBuffer::len).sum();
866        assert!(last_segment_end > serialized_footer_size as u64);
867        let mut footer_bytes = ByteBufferMut::with_capacity(serialized_footer_size);
868        for buffer in serialized_footer {
869            footer_bytes.extend_from_slice(&buffer);
870        }
871
872        let mut deserializer = Footer::deserializer(footer_bytes.freeze(), session.clone())
873            .with_size(serialized_footer_size as u64);
874        let DeserializeStep::Done(cached_footer) = deserializer.deserialize()? else {
875            vortex_bail!("standalone footer bytes must be sufficient for deserialization");
876        };
877
878        session
879            .open_options()
880            .with_footer(cached_footer)
881            .open_buffer(file_bytes)?;
882
883        Ok(())
884    }
885}