do not merge: Sub seg read

Cargo.toml

@@ -167,6 +167,7 @@ geoarrow = "0.8.0"
 geoarrow-cast = "0.8.0"
 get_dir = "0.5.0"
 glob = "0.3.2"
+io-uring = "0.7.13"
 goldenfile = "1"
 half = { version = "2.7.1", features = ["std", "num-traits"] }
 hashbrown = "0.17.1"

benchmarks/datafusion-bench/src/main.rs

@@ -289,10 +289,14 @@ async fn register_v2_tables<B: Benchmark + ?Sized>(
             .runtime_env()
             .object_store(table_url.object_store())?;
 
-        let fs: FileSystemRef = Arc::new(ObjectStoreFileSystem::new(
-            Arc::clone(&store),
-            SESSION.handle(),
-        ));
+        let fs: FileSystemRef = if benchmark_base.scheme() == "file" {
+            Arc::new(ObjectStoreFileSystem::local(SESSION.handle()))
+        } else {
+            Arc::new(ObjectStoreFileSystem::new(
+                Arc::clone(&store),
+                SESSION.handle(),
+            ))
+        };
         let base_prefix = benchmark_base.path().trim_start_matches('/').to_string();
         let fs = fs.with_prefix(base_prefix);
 

encodings/alp/src/alp_rd/array.rs

@@ -71,6 +71,33 @@ pub struct ALPRDMetadata {
     patches: Option<PatchesMetadata>,
 }
 
+impl ALPRDMetadata {
+    pub fn right_bit_width(&self) -> VortexResult<u8> {
+        u8::try_from(self.right_bit_width).map_err(|_| {
+            vortex_err!(
+                "right bit width {} does not fit in u8",
+                self.right_bit_width
+            )
+        })
+    }
+
+    pub fn left_parts_dictionary(&self) -> VortexResult<Buffer<u16>> {
+        self.dict
+            .get(..usize::try_from(self.dict_len)?)
+            .ok_or_else(|| vortex_err!("ALPRD dictionary length is out of bounds"))?
+            .iter()
+            .map(|&value| {
+                u16::try_from(value)
+                    .map_err(|_| vortex_err!("ALPRD dictionary value {value} does not fit in u16"))
+            })
+            .collect()
+    }
+
+    pub fn patches(&self) -> Option<&PatchesMetadata> {
+        self.patches.as_ref()
+    }
+}
+
 impl ArrayHash for ALPRDData {
     fn array_hash<H: Hasher>(&self, state: &mut H, accuracy: EqMode) {
         self.left_parts_dictionary.array_hash(state, accuracy);

encodings/fastlanes/src/bitpacking/mod.rs

@@ -19,6 +19,7 @@ mod vtable;
 pub(crate) use plugin::BitPackedPatchedPlugin;
 pub use vtable::BitPacked;
 pub use vtable::BitPackedArray;
+pub use vtable::BitPackedMetadata;
 
 pub(crate) fn initialize(session: &vortex_session::VortexSession) {
     vtable::initialize(session);

encodings/fastlanes/src/bitpacking/vtable/mod.rs

@@ -72,6 +72,22 @@ pub struct BitPackedMetadata {
     pub(crate) patches: Option<PatchesMetadata>,
 }
 
+impl BitPackedMetadata {
+    pub fn bit_width(&self) -> VortexResult<u8> {
+        u8::try_from(self.bit_width)
+            .map_err(|_| vortex_err!("bit width {} does not fit in u8", self.bit_width))
+    }
+
+    pub fn offset(&self) -> VortexResult<u16> {
+        u16::try_from(self.offset)
+            .map_err(|_| vortex_err!("bit-packed offset {} does not fit in u16", self.offset))
+    }
+
+    pub fn patches(&self) -> Option<&PatchesMetadata> {
+        self.patches.as_ref()
+    }
+}
+
 impl ArrayHash for BitPackedData {
     fn array_hash<H: Hasher>(&self, state: &mut H, accuracy: EqMode) {
         self.offset.hash(state);

vortex-array/src/arrays/list/mod.rs

@@ -14,6 +14,7 @@ pub(crate) mod compute;
 
 mod vtable;
 pub use vtable::List;
+pub use vtable::ListMetadata;
 
 pub(crate) fn initialize(session: &vortex_session::VortexSession) {
     compute::initialize(session);

vortex-array/src/arrays/list/vtable/mod.rs

@@ -52,6 +52,12 @@ pub struct ListMetadata {
     offset_ptype: i32,
 }
 
+impl ListMetadata {
+    pub fn elements_len(&self) -> u64 {
+        self.elements_len
+    }
+}
+
 impl ArrayHash for ListData {
     fn array_hash<H: Hasher>(&self, _state: &mut H, _accuracy: EqMode) {}
 }

vortex-array/src/mask_future.rs

@@ -20,6 +20,9 @@ use vortex_mask::Mask;
 pub struct MaskFuture {
     inner: Shared<BoxFuture<'static, SharedVortexResult<Mask>>>,
     len: usize,
+    upper_bound: Option<Mask>,
+    upper_bound_is_exact: bool,
+    partial_reads_allowed: bool,
 }
 
 impl MaskFuture {
@@ -40,6 +43,9 @@ impl MaskFuture {
                 .boxed()
                 .shared(),
             len,
+            upper_bound: None,
+            upper_bound_is_exact: false,
+            partial_reads_allowed: false,
         }
     }
 
@@ -55,7 +61,12 @@ impl MaskFuture {
 
     /// Create a MaskFuture from a ready mask.
     pub fn ready(mask: Mask) -> Self {
-        Self::new(mask.len(), async move { Ok(mask) })
+        let upper_bound = mask.clone();
+        let mut future = Self::new(mask.len(), async move { Ok(mask) });
+        future.upper_bound = Some(upper_bound);
+        future.upper_bound_is_exact = true;
+        future.partial_reads_allowed = true;
+        future
     }
 
     /// Create a MaskFuture that resolves to a mask with all values set to true.
@@ -72,7 +83,57 @@ impl MaskFuture {
         }
 
         let inner = self.inner.clone();
-        Self::new(range.len(), async move { Ok(inner.await?.slice(range)) })
+        let upper_bound = self
+            .upper_bound
+            .as_ref()
+            .map(|upper_bound| upper_bound.slice(range.clone()));
+        let mut sliced = Self::new(range.len(), async move { Ok(inner.await?.slice(range)) });
+        sliced.upper_bound = upper_bound;
+        sliced.upper_bound_is_exact = self.upper_bound_is_exact;
+        sliced.partial_reads_allowed = self.partial_reads_allowed;
+        sliced
+    }
+
+    /// Attach a conservative upper bound for the mask resolved by this future.
+    ///
+    /// Readers can use this to register I/O eagerly without waiting for filter evaluation. The
+    /// resolved mask must not contain a true row that is false in `upper_bound`.
+    pub fn with_upper_bound(mut self, upper_bound: Mask) -> Self {
+        assert_eq!(
+            upper_bound.len(),
+            self.len,
+            "MaskFuture upper bound length mismatch"
+        );
+        self.upper_bound = Some(upper_bound);
+        self.upper_bound_is_exact = false;
+        self
+    }
+
+    /// Return the conservative upper bound for this future, when one is known.
+    pub fn upper_bound(&self) -> Option<&Mask> {
+        self.upper_bound.as_ref()
+    }
+
+    /// Return whether the upper bound is the exact mask returned by this future.
+    pub fn upper_bound_is_exact(&self) -> bool {
+        self.upper_bound_is_exact
+    }
+
+    /// Permit readers to satisfy this selection using partial segment reads.
+    pub fn with_partial_reads(mut self) -> Self {
+        self.partial_reads_allowed = true;
+        self
+    }
+
+    /// Prevent readers from turning this mask into partial segment reads.
+    pub fn without_partial_reads(mut self) -> Self {
+        self.partial_reads_allowed = false;
+        self
+    }
+
+    /// Return whether readers may satisfy this selection using partial segment reads.
+    pub fn partial_reads_allowed(&self) -> bool {
+        self.partial_reads_allowed
     }
 
     pub fn inspect(
@@ -84,6 +145,9 @@ impl MaskFuture {
         Self {
             inner: self.inner.inspect(f).boxed().shared(),
             len,
+            upper_bound: self.upper_bound,
+            upper_bound_is_exact: self.upper_bound_is_exact,
+            partial_reads_allowed: self.partial_reads_allowed,
         }
     }
 }
@@ -119,8 +183,26 @@ mod tests {
 
             let partial = fut.slice(0..mask.len() - 1);
             assert_eq!(partial.len(), mask.len() - 1);
+            assert_eq!(partial.upper_bound(), Some(&mask.slice(0..mask.len() - 1)));
             assert_eq!(partial.await?, mask.slice(0..mask.len() - 1));
             Ok(())
         })
     }
+
+    #[test]
+    fn new_future_has_no_upper_bound_until_attached() {
+        let future = MaskFuture::new(3, async { Ok(Mask::new_false(3)) });
+        assert!(future.upper_bound().is_none());
+
+        let upper_bound = Mask::from_indices(3, [0, 2]);
+        let future = future.with_upper_bound(upper_bound.clone());
+        assert_eq!(future.upper_bound(), Some(&upper_bound));
+        assert!(!future.upper_bound_is_exact());
+    }
+
+    #[test]
+    fn ready_future_has_an_exact_upper_bound() {
+        let future = MaskFuture::ready(Mask::from_indices(3, [0, 2]));
+        assert!(future.upper_bound_is_exact());
+    }
 }

vortex-array/src/serde.rs

@@ -5,6 +5,7 @@ use std::borrow::Cow;
 use std::fmt::Debug;
 use std::fmt::Formatter;
 use std::iter;
+use std::ops::Range;
 use std::sync::Arc;
 
 use flatbuffers::FlatBufferBuilder;
@@ -294,6 +295,31 @@ pub struct SerializedArray {
     buffers: Arc<[BufferHandle]>,
 }
 
+/// Location and alignment of one serialized array buffer within its containing segment.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct SerializedBuffer {
+    index: usize,
+    range: Range<usize>,
+    alignment: Alignment,
+}
+
+impl SerializedBuffer {
+    /// Return this buffer's index in the serialized array's global buffer table.
+    pub fn index(&self) -> usize {
+        self.index
+    }
+
+    /// Return this buffer's byte range within the containing segment.
+    pub fn range(&self) -> &Range<usize> {
+        &self.range
+    }
+
+    /// Return the alignment required when materializing this buffer independently.
+    pub fn alignment(&self) -> Alignment {
+        self.alignment
+    }
+}
+
 impl Debug for SerializedArray {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("SerializedArray")
@@ -516,6 +542,46 @@ impl SerializedArray {
             .unwrap_or_default()
     }
 
+    /// Return the global buffer indices referenced by this array node.
+    pub fn buffer_indices(&self) -> Vec<usize> {
+        self.flatbuffer()
+            .buffers()
+            .map(|buffers| buffers.iter().map(usize::from).collect())
+            .unwrap_or_default()
+    }
+
+    /// Return validated locations for all data buffers in their serialized segment.
+    pub fn buffer_descriptors(&self) -> VortexResult<Vec<SerializedBuffer>> {
+        let fb_array = root::<fba::Array>(self.flatbuffer.as_ref())?;
+        let mut offset = 0usize;
+        fb_array
+            .buffers()
+            .unwrap_or_default()
+            .iter()
+            .enumerate()
+            .map(|(index, buffer)| {
+                if buffer.compression() != Compression::None {
+                    vortex_bail!(
+                        "Partial reads do not support serialized buffer compression {:?}",
+                        buffer.compression()
+                    );
+                }
+                let start = offset
+                    .checked_add(usize::from(buffer.padding()))
+                    .ok_or_else(|| vortex_err!("Buffer {index} padding overflows"))?;
+                let end = start
+                    .checked_add(buffer.length() as usize)
+                    .ok_or_else(|| vortex_err!("Buffer {index} length overflows"))?;
+                offset = end;
+                Ok(SerializedBuffer {
+                    index,
+                    range: start..end,
+                    alignment: Alignment::try_from_untrusted_exponent(buffer.alignment_exponent())?,
+                })
+            })
+            .collect()
+    }
+
     /// Validate and align the array tree flatbuffer, returning the aligned buffer and root location.
     fn validate_array_tree(array_tree: impl Into<ByteBuffer>) -> VortexResult<(FlatBuffer, usize)> {
         let fb_buffer = FlatBuffer::align_from(array_tree.into());

vortex-file/Cargo.toml

@@ -17,6 +17,7 @@ version = { workspace = true }
 all-features = true
 
 [dependencies]
+async-stream = { workspace = true }
 async-trait = { workspace = true }
 bytes = { workspace = true }
 flatbuffers = { workspace = true }