Add file-level user metadata

docs/specs/file-format.md

@@ -54,6 +54,13 @@ The postscript contains the locations of:
 2. a `layout` segment containing the root `Layout`
 3. a `statistics` segment containing file-level per-field statistics (e.g., minima and maxima of each field/column, for whole-file pruning)
 4. a `footer` segment containing a dictionary-encoded _segment map_, and other shared configuration such as compression and encryption schemes
+5. up to 16 user-defined `metadata` segments, each identified by a unique, non-empty UTF-8 key of at most 64 bytes
+
+The postscript carries a locator (offset, length, and alignment) for each metadata segment that is
+present; a file written without user metadata (and any file predating this feature) carries none.
+Readers do not load the opaque metadata values by default. Opt-in metadata reads resolve each
+locator separately, allowing values outside the initial file-tail read to be fetched without reading
+the intervening file contents.
 
 :::{literalinclude} ../../vortex-flatbuffers/flatbuffers/vortex-file/footer.fbs
 :start-after: [postscript]

vortex-buffer/src/alignment.rs

@@ -117,9 +117,14 @@ impl Alignment {
     ///
     /// ## Panics
     ///
-    /// Panics if `1 << exponent` overflows `usize`.
+    /// Panics if `1 << exponent` overflows `usize`. Use [`Self::try_from_exponent`] when parsing
+    /// untrusted input.
     #[inline]
     pub const fn from_exponent(exponent: u8) -> Self {
+        assert!(
+            (exponent as u32) < usize::BITS,
+            "Alignment exponent must fit in usize"
+        );
         Self::new(1 << exponent)
     }
 

vortex-file/src/file.rs

@@ -15,6 +15,7 @@ use vortex_array::ArrayRef;
 use vortex_array::dtype::DType;
 use vortex_array::dtype::FieldMask;
 use vortex_array::expr::Expression;
+use vortex_buffer::ByteBuffer;
 use vortex_error::VortexResult;
 use vortex_layout::LayoutReader;
 use vortex_layout::scan::layout::LayoutReaderDataSource;
@@ -23,6 +24,7 @@ use vortex_layout::scan::split_by::SplitBy;
 use vortex_layout::segments::SegmentSource;
 use vortex_scan::DataSourceRef;
 use vortex_session::VortexSession;
+use vortex_utils::aliases::hash_map::HashMap;
 
 use crate::FileStatistics;
 use crate::footer::Footer;
@@ -42,6 +44,8 @@ pub struct VortexFile {
     segment_source: Arc<dyn SegmentSource>,
     /// The Vortex session used to open this file.
     session: VortexSession,
+    /// User-defined metadata values resolved for this file open.
+    metadata: Arc<HashMap<String, ByteBuffer>>,
     /// None id LayoutReader caching is turned off
     layout_reader_cache: Option<OnceLock<Arc<dyn LayoutReader>>>,
 }
@@ -78,10 +82,16 @@ impl VortexFile {
             footer,
             segment_source,
             session,
+            metadata: Arc::new(HashMap::new()),
             layout_reader_cache: None,
         }
     }
 
+    pub(crate) fn with_metadata(mut self, metadata: Arc<HashMap<String, ByteBuffer>>) -> Self {
+        self.metadata = metadata;
+        self
+    }
+
     /// Enable layout reader caching.
     ///
     /// Repeated calls to [`layout_reader`](Self::layout_reader), [`scan`](Self::scan), and
@@ -91,6 +101,7 @@ impl VortexFile {
             footer: self.footer,
             segment_source: self.segment_source,
             session: self.session,
+            metadata: self.metadata,
             layout_reader_cache: Some(OnceLock::new()),
         }
     }
@@ -117,6 +128,22 @@ impl VortexFile {
         self.footer.statistics()
     }
 
+    /// Returns the user-defined metadata segments loaded for this file.
+    ///
+    /// Metadata is only loaded when requested during open. Iteration order is unspecified.
+    pub fn metadata_segments(&self) -> impl Iterator<Item = (&str, &ByteBuffer)> {
+        self.metadata
+            .iter()
+            .map(|(key, metadata)| (key.as_str(), metadata))
+    }
+
+    /// Returns the loaded user-defined metadata segment for the given key.
+    ///
+    /// Returns `None` when the key is absent or metadata was not loaded.
+    pub fn metadata_segment(&self, key: &str) -> Option<&ByteBuffer> {
+        self.metadata.get(key)
+    }
+
     /// Create a new segment source for reading from the file.
     ///
     /// This may spawn a background I/O driver that will exit when the returned segment source

vortex-file/src/footer/deserializer.rs

@@ -1,6 +1,8 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
+use std::sync::Arc;
+
 use flatbuffers::root;
 use vortex_array::dtype::DType;
 use vortex_buffer::ByteBuffer;
@@ -18,6 +20,7 @@ use crate::Footer;
 use crate::MAGIC_BYTES;
 use crate::VERSION;
 use crate::footer::FileStatistics;
+use crate::footer::SegmentSpec;
 use crate::footer::postscript::Postscript;
 use crate::footer::postscript::PostscriptSegment;
 
@@ -177,14 +180,50 @@ impl FooterDeserializer {
                 )
             })
             .transpose()?;
+        let metadata = postscript
+            .metadata
+            .iter()
+            .map(|metadata| {
+                let segment = SegmentSpec {
+                    offset: metadata.segment.offset,
+                    length: metadata.segment.length,
+                    alignment: metadata.segment.alignment,
+                };
+                let end = segment
+                    .offset
+                    .checked_add(u64::from(segment.length))
+                    .ok_or_else(|| {
+                        vortex_err!("Metadata segment {} range overflowed u64", metadata.key)
+                    })?;
+                if end > file_size {
+                    vortex_bail!(
+                        "Metadata segment {} range {}..{} exceeds file size {}",
+                        metadata.key,
+                        segment.offset,
+                        end,
+                        file_size
+                    );
+                }
+                let offset = usize::try_from(segment.offset)?;
+                if !segment.alignment.is_offset_aligned(offset) {
+                    vortex_bail!(
+                        "Metadata segment {} offset {} is not aligned to {}",
+                        metadata.key,
+                        segment.offset,
+                        segment.alignment
+                    );
+                }
+                Ok((metadata.key.clone(), segment))
+            })
+            .collect::<VortexResult<Arc<[_]>>>()?;
 
         Ok(DeserializeStep::Done(self.parse_footer(
             initial_offset,
             &self.buffer,
-            &postscript.footer,
-            &postscript.layout,
+            postscript,
             dtype,
             file_stats,
+            metadata,
         )?))
     }
 
@@ -269,19 +308,29 @@ impl FooterDeserializer {
         &self,
         initial_offset: u64,
         initial_read: &[u8],
-        footer_segment: &PostscriptSegment,
-        layout_segment: &PostscriptSegment,
+        postscript: &Postscript,
         dtype: DType,
         file_stats: Option<FileStatistics>,
+        metadata: Arc<[(String, SegmentSpec)]>,
     ) -> VortexResult<Footer> {
+        let footer_segment = &postscript.footer;
         let footer_bytes = checked_segment_slice(initial_read, initial_offset, footer_segment)?;
+
+        let layout_segment = &postscript.layout;
         let layout_bytes = FlatBuffer::copy_from(checked_segment_slice(
             initial_read,
             initial_offset,
             layout_segment,
         )?);
 
-        Footer::from_flatbuffer(footer_bytes, layout_bytes, dtype, file_stats, &self.session)
+        Footer::from_flatbuffer(
+            footer_bytes,
+            layout_bytes,
+            dtype,
+            file_stats,
+            metadata,
+            &self.session,
+        )
     }
 }
 
@@ -374,6 +423,7 @@ mod tests {
             layout: segment(0, 1),
             statistics: None,
             footer: footer_segment,
+            metadata: Vec::new(),
         };
         let buffer = eof_buffer(&postscript)?;
         let file_size = buffer.len() as u64;
@@ -393,6 +443,7 @@ mod tests {
             layout: segment(0, 1),
             statistics: None,
             footer: segment(1, 1),
+            metadata: Vec::new(),
         };
         let buffer = eof_buffer(&postscript)?;
         let declared_size = buffer.len() as u64 - 1;

vortex-file/src/footer/mod.rs

@@ -39,12 +39,27 @@ use vortex_layout::layout_from_flatbuffer_with_options;
 use vortex_session::VortexSession;
 use vortex_session::registry::ReadContext;
 
+/// Maximum number of user-defined metadata segments. Keeps postscript bookkeeping small so the
+/// footer and required segments still fit the initial tail read.
+pub(crate) const MAX_METADATA_SEGMENTS: usize = 16;
+
+/// Maximum length, in UTF-8 bytes, of a user-defined metadata key (keys live in the postscript).
+///
+/// 64 bytes covers reverse-DNS query-engine keys, not just short Iceberg-style keys: e.g.
+/// `org.apache.spark.sql.parquet.row.metadata` (41 bytes), which Spark writes into every Parquet
+/// file. With [`MAX_METADATA_SEGMENTS`] keys this bounds the postscript key budget at 1 KiB.
+pub(crate) const MAX_METADATA_KEY_BYTES: usize = 64;
+
+/// User-defined metadata segment locators stored as `(key, locator)` pairs.
+pub(crate) type MetadataSegments = Arc<[(String, SegmentSpec)]>;
+
 /// Captures the layout information of a Vortex file.
 #[derive(Debug, Clone)]
 pub struct Footer {
     root_layout: LayoutRef,
     segments: Arc<[SegmentSpec]>,
     statistics: Option<FileStatistics>,
+    metadata: Arc<[(String, SegmentSpec)]>,
     // The specific arrays used within the file, in the order they were registered.
     array_read_ctx: ReadContext,
     // The approximate size of the footer in bytes, used for caching and memory management.
@@ -62,6 +77,7 @@ impl Footer {
             root_layout,
             segments,
             statistics,
+            metadata: Arc::from([]),
             array_read_ctx,
             approx_byte_size: None,
         }
@@ -78,9 +94,14 @@ impl Footer {
         layout_bytes: FlatBuffer,
         dtype: DType,
         statistics: Option<FileStatistics>,
+        metadata: Arc<[(String, SegmentSpec)]>,
         session: &VortexSession,
     ) -> VortexResult<Self> {
-        let approx_byte_size = footer_bytes.len() + layout_bytes.len();
+        let metadata_bytes: usize = metadata
+            .iter()
+            .map(|(key, _segment)| key.len() + size_of::<SegmentSpec>())
+            .sum();
+        let approx_byte_size = footer_bytes.len() + layout_bytes.len() + metadata_bytes;
         let fb_footer = root::<fb::Footer>(footer_bytes)?;
 
         // Create a LayoutContext from the registry.
@@ -128,6 +149,7 @@ impl Footer {
             root_layout,
             segments,
             statistics,
+            metadata,
             array_read_ctx,
             approx_byte_size: Some(approx_byte_size),
         })
@@ -148,6 +170,33 @@ impl Footer {
         self.statistics.as_ref()
     }
 
+    /// Returns the user-defined metadata segment locators stored in the postscript.
+    pub fn metadata_segments(&self) -> impl Iterator<Item = (&str, &SegmentSpec)> {
+        self.metadata
+            .iter()
+            .map(|(key, segment)| (key.as_str(), segment))
+    }
+
+    /// Returns the user-defined metadata segment locator for the given key.
+    pub fn metadata_segment(&self, key: &str) -> Option<&SegmentSpec> {
+        self.metadata
+            .iter()
+            .find_map(|(candidate, segment)| (candidate == key).then_some(segment))
+    }
+
+    pub(crate) fn with_metadata_segments(mut self, metadata: Arc<[(String, SegmentSpec)]>) -> Self {
+        self.metadata = metadata;
+        self
+    }
+
+    pub(crate) fn segment_specs_with_metadata(&self) -> Arc<[SegmentSpec]> {
+        self.segments
+            .iter()
+            .copied()
+            .chain(self.metadata.iter().map(|(_, segment)| *segment))
+            .collect()
+    }
+
     /// Computes the compressed size in bytes of every field in the file, keyed by field path.
     ///
     /// Sizes are derived by attributing each segment in the [segment map][Self::segment_map] to a