1use 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#[derive(Clone)]
57pub struct VortexOpenOptions {
58 session: VortexSession,
60 segment_cache: Option<Arc<dyn SegmentCache>>,
62 initial_read_size: usize,
64 file_size: Option<u64>,
66 dtype: Option<DType>,
68 footer: Option<Footer>,
70 include_metadata: bool,
72 metrics_registry: Option<Arc<dyn MetricsRegistry>>,
74 labels: Vec<Label>,
76 cache_layout_reader: bool,
78}
79
80pub trait OpenOptionsSessionExt:
82 ArraySessionExt + LayoutSessionExt + RuntimeSessionExt + MemorySessionExt
83{
84 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 pub fn session(&self) -> &VortexSession {
108 &self.session
109 }
110
111 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 pub fn with_layout_reader_cache(mut self) -> Self {
126 self.cache_layout_reader = true;
127 self
128 }
129
130 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 pub fn without_segment_cache(mut self) -> Self {
144 self.segment_cache = None;
145 self
146 }
147
148 pub fn with_file_size(mut self, file_size: u64) -> Self {
153 self.file_size = Some(file_size);
154 self
155 }
156
157 pub fn with_some_file_size(mut self, file_size: Option<u64>) -> Self {
162 self.file_size = file_size;
163 self
164 }
165
166 pub fn with_dtype(mut self, dtype: DType) -> Self {
172 self.dtype = Some(dtype);
173 self
174 }
175
176 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 pub fn include_metadata(mut self) -> Self {
192 self.include_metadata = true;
193 self
194 }
195
196 pub fn with_include_metadata(mut self, include_metadata: bool) -> Self {
201 self.include_metadata = include_metadata;
202 self
203 }
204
205 pub fn with_metrics_registry(mut self, metrics: Arc<dyn MetricsRegistry>) -> Self {
207 self.metrics_registry = Some(metrics);
208 self
209 }
210
211 pub fn with_labels(mut self, labels: Vec<Label>) -> Self {
213 self.labels.extend(labels);
214 self
215 }
216
217 pub async fn open(self, source: Arc<dyn VortexReadAt>) -> VortexResult<VortexFile> {
224 self.open_read(source).await
225 }
226
227 #[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 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 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 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 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 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 .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 footer.validate_file_size(file_size)?;
392
393 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 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 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 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 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 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 let mut buf = ByteBufferMut::empty();
594
595 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 let _file = session.open_options().open_read(reader).await.unwrap();
633
634 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 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 #[tokio::test]
800 async fn collect_initial_segments_rejects_out_of_bounds_segment() -> VortexResult<()> {
801 let session = test_session();
802
803 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 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}