Add FlatLayout range read for sub-segment IO

When a FlatLayout has its array_tree metadata inlined in the footer, we can figure out exactly which bytes of the segment are needed for a given row range without any IO. This lets us issue a single small read instead of fetching the entire segment, which is a big win for point lookups and narrow scans on wide tables.

The range read planner walks the encoding tree (Primitive, Bool, BitPacked, Delta, FoR, ZigZag, ALP, ALPRD, Dict, FixedSizeList, Constant) and computes the minimal contiguous byte range covering the needed buffers. If that range is less than 50% of the full segment, we issue the targeted read; otherwise we fall back to reading the whole segment.

To make Delta work with sub-ranged buffers, Delta::build() now derives child array lengths from `len + offset` instead of metadata.deltas_len. On disk, offset is always 0 so this is a no-op for the normal decode path, but it lets the range read pass a smaller decode_len without the decoder panicking on buffer size mismatch.

Also adds `request_range()` to the SegmentSource trait with a default fallback implementation, efficient overrides in FileSegmentSource and BufferSegmentSource, a `RangeReadEnabled` session flag, and `pub const NAME` on all encoding structs for pattern matching in the planner.

Signed-off-by: jiaqizho <jiaqi.zhou@zilliz.com>

Author: jiaqizho

Cargo.lock

@@ -273,7 +273,8 @@ pub struct ALPArray {
 pub struct ALP;
 
 impl ALP {
-    pub const ID: ArrayId = ArrayId::new_ref("vortex.alp");
+    pub const NAME: &str = "vortex.alp";
+    pub const ID: ArrayId = ArrayId::new_ref(Self::NAME);
 }
 
 #[derive(Clone, prost::Message)]

encodings/alp/src/alp/array.rs

@@ -371,7 +371,8 @@ pub struct ALPRDArray {
 pub struct ALPRD;
 
 impl ALPRD {
-    pub const ID: ArrayId = ArrayId::new_ref("vortex.alprd");
+    pub const NAME: &str = "vortex.alprd";
+    pub const ID: ArrayId = ArrayId::new_ref(Self::NAME);
 }
 
 impl ALPRDArray {

encodings/alp/src/alp_rd/array.rs

@@ -213,7 +213,8 @@ pub struct ByteBoolArray {
 pub struct ByteBool;
 
 impl ByteBool {
-    pub const ID: ArrayId = ArrayId::new_ref("vortex.bytebool");
+    pub const NAME: &str = "vortex.bytebool";
+    pub const ID: ArrayId = ArrayId::new_ref(Self::NAME);
 }
 
 impl ByteBoolArray {

encodings/bytebool/src/array.rs

@@ -267,7 +267,8 @@ pub struct DateTimePartsArrayParts {
 pub struct DateTimeParts;
 
 impl DateTimeParts {
-    pub const ID: ArrayId = ArrayId::new_ref("vortex.datetimeparts");
+    pub const NAME: &str = "vortex.datetimeparts";
+    pub const ID: ArrayId = ArrayId::new_ref(Self::NAME);
 }
 
 impl DateTimePartsArray {

encodings/datetime-parts/src/array.rs

@@ -274,7 +274,8 @@ impl DecimalBytePartsArray {
 pub struct DecimalByteParts;
 
 impl DecimalByteParts {
-    pub const ID: ArrayId = ArrayId::new_ref("vortex.decimal_byte_parts");
+    pub const NAME: &str = "vortex.decimal_byte_parts";
+    pub const ID: ArrayId = ArrayId::new_ref(Self::NAME);
 }
 
 /// Converts a DecimalBytePartsArray to its canonical DecimalArray representation.

encodings/decimal-byte-parts/src/decimal_byte_parts/mod.rs

@@ -372,5 +372,6 @@ impl VTable for BitPacked {
 pub struct BitPacked;
 
 impl BitPacked {
-    pub const ID: ArrayId = ArrayId::new_ref("fastlanes.bitpacked");
+    pub const NAME: &str = "fastlanes.bitpacked";
+    pub const ID: ArrayId = ArrayId::new_ref(Self::NAME);
 }

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

@@ -25,7 +25,6 @@ use vortex_array::vtable::VTable;
 use vortex_array::vtable::ValidityVTableFromChildSliceHelper;
 use vortex_error::VortexResult;
 use vortex_error::vortex_ensure;
-use vortex_error::vortex_err;
 use vortex_error::vortex_panic;
 use vortex_session::VortexSession;
 
@@ -177,11 +176,12 @@ impl VTable for Delta {
         let ptype = PType::try_from(dtype)?;
         let lanes = match_each_unsigned_integer_ptype!(ptype, |T| { <T as FastLanes>::LANES });
 
-        // Compute the length of the bases array
-        let deltas_len = usize::try_from(metadata.0.deltas_len)
-            .map_err(|_| vortex_err!("deltas_len {} overflowed usize", metadata.0.deltas_len))?;
+        // Compute the length of the deltas array from len + offset rather than metadata.
+        // This allows range reads to work with sub-ranged children, where the buffer is
+        // shorter than the original metadata.deltas_len.
+        let deltas_len = len + metadata.0.offset as usize;
         let num_chunks = deltas_len / 1024;
-        let remainder_base_size = if deltas_len % 1024 > 0 { 1 } else { 0 };
+        let remainder_base_size = if !deltas_len.is_multiple_of(1024) { 1 } else { 0 };
         let bases_len = num_chunks * lanes + remainder_base_size;
 
         let bases = children.get(0, dtype, bases_len)?;
@@ -201,7 +201,8 @@ impl VTable for Delta {
 pub struct Delta;
 
 impl Delta {
-    pub const ID: ArrayId = ArrayId::new_ref("fastlanes.delta");
+    pub const NAME: &str = "fastlanes.delta";
+    pub const ID: ArrayId = ArrayId::new_ref(Self::NAME);
 }
 
 #[cfg(test)]

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

@@ -184,5 +184,6 @@ impl VTable for FoR {
 pub struct FoR;
 
 impl FoR {
-    pub const ID: ArrayId = ArrayId::new_ref("fastlanes.for");
+    pub const NAME: &str = "fastlanes.for";
+    pub const ID: ArrayId = ArrayId::new_ref(Self::NAME);
 }

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

@@ -242,7 +242,8 @@ impl VTable for RLE {
 pub struct RLE;
 
 impl RLE {
-    pub const ID: ArrayId = ArrayId::new_ref("fastlanes.rle");
+    pub const NAME: &str = "fastlanes.rle";
+    pub const ID: ArrayId = ArrayId::new_ref(Self::NAME);
 }
 
 #[cfg(test)]