fix: preserve nullable struct child offsets

Cargo.lock

@@ -5217,7 +5217,6 @@ dependencies = [
 [[package]]
 name = "lance-arrow"
 version = "13.0.0-beta.8"
-source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.8#3c20b7875f621c929fdcecabe74dbca507ec0f0d"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -5888,6 +5887,7 @@ dependencies = [
  "datafusion-common",
  "env_logger",
  "futures",
+ "lance-arrow",
  "lance-core",
  "lance-io",
  "lance-namespace",

Cargo.toml

@@ -1,5 +1,6 @@
 [workspace]
 members = ["rust/lancedb", "nodejs", "python"]
+exclude = ["python/vendor/lance-arrow"]
 resolver = "2"
 
 [workspace.package]
@@ -79,6 +80,10 @@ tonic = { version = "0.14", features = ["tls-native-roots", "tls-ring"] }
 uuid = { version = "1.7.0", features = ["v4", "v7"] }
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
 
+[patch."https://github.com/lance-format/lance.git"]
+# Carry the offset-safe validity fix until it is available in a Lance release.
+lance-arrow = { path = "python/vendor/lance-arrow" }
+
 [profile.ci]
 debug = "line-tables-only"
 inherits = "dev"

python/Cargo.toml

@@ -22,6 +22,9 @@ bytes.workspace = true
 lancedb.workspace = true
 datafusion-common.workspace = true
 lance-core.workspace = true
+# Make the patched Lance Arrow crate an explicit path dependency so Maturin
+# includes it at the path used by the workspace patch in source distributions.
+lance-arrow = { version = "=13.0.0-beta.8", path = "vendor/lance-arrow" }
 lance-namespace.workspace = true
 lance-namespace-impls.workspace = true
 lance-io.workspace = true

python/vendor/lance-arrow/Cargo.toml

@@ -0,0 +1,33 @@
+[package]
+name = "lance-arrow"
+version = "13.0.0-beta.8"
+edition = "2024"
+authors = ["Lance Devs <dev@lance.org>"]
+license = "Apache-2.0"
+repository = "https://github.com/lance-format/lance"
+readme = "README.md"
+description = "Arrow Extension for Lance"
+keywords = ["data-format", "data-science", "machine-learning", "apache-arrow", "data-analytics"]
+categories = ["database-implementations"]
+
+[lib]
+crate-type = ["cdylib", "rlib"]
+
+[dependencies]
+arrow-array = "58.0.0"
+arrow-buffer = "58.0.0"
+arrow-data = "58.0.0"
+arrow-ipc = { version = "58.0.0", features = ["zstd"] }
+arrow-ord = "58.0.0"
+arrow-schema = "58.0.0"
+arrow-select = "58.0.0"
+bytemuck = { version = "1", default-features = false, features = ["extern_crate_alloc"] }
+bytes = "1.11.1"
+futures = "0.3"
+half = { version = "2.1", default-features = false, features = ["num-traits", "std", "bytemuck"] }
+jsonb = { version = "0.5.3", default-features = false, features = ["databend"] }
+num-traits = "0.2"
+rand = { version = "0.9.1", features = ["small_rng"] }
+
+[target.'cfg(target_arch = "wasm32")'.dependencies]
+getrandom = { version = "0.2", features = ["js"] }

python/vendor/lance-arrow/README.lancedb.md

@@ -0,0 +1,7 @@
+This directory contains `rust/lance-arrow` from Lance tag `v13.0.0-beta.8`
+(commit `3c20b7875f621c929fdcecabe74dbca507ec0f0d`), licensed under Apache-2.0.
+
+LanceDB carries a local fix in `src/lib.rs` for sliced child validity bitmaps
+until a Lance release includes the fix. The patch is selected in the root
+`Cargo.toml`; the rest of Lance stays on the pinned release. An explicit unsafe
+block in `src/bfloat16.rs` lets this copy pass LanceDB's Rust 2024 warning gate.

python/vendor/lance-arrow/README.md

@@ -0,0 +1,6 @@
+# lance-arrow
+
+`lance-arrow` is an internal sub-crate, containing [Apache-Arrow](https://github.com/apache/arrow-rs)
+extensions used by [Lance](https://github.com/lance-format/lance).
+
+**Important Note**: This crate is **not intended for external usage**.

python/vendor/lance-arrow/src/bfloat16.rs

@@ -0,0 +1,405 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+
+//! bfloat16 support for Apache Arrow.
+
+use std::fmt::Formatter;
+use std::slice;
+
+use arrow_array::{Array, FixedSizeBinaryArray, builder::BooleanBufferBuilder};
+use arrow_buffer::{Buffer, MutableBuffer};
+use arrow_data::ArrayData;
+use arrow_schema::{ArrowError, DataType, Field as ArrowField};
+use half::bf16;
+
+use crate::{ARROW_EXT_NAME_KEY, FloatArray};
+
+/// The name of the bfloat16 extension in Arrow metadata
+pub const BFLOAT16_EXT_NAME: &str = "lance.bfloat16";
+
+/// Check whether the given field is a bfloat16 field
+///
+/// A field is a bfloat16 field if it has a data type of `FixedSizeBinary(2)` and the metadata
+/// contains the bfloat16 extension name.
+pub fn is_bfloat16_field(field: &ArrowField) -> bool {
+    field.data_type() == &DataType::FixedSizeBinary(2)
+        && field
+            .metadata()
+            .get(ARROW_EXT_NAME_KEY)
+            .map(|name| name == BFLOAT16_EXT_NAME)
+            .unwrap_or_default()
+}
+
+/// The bfloat16 data type
+///
+/// This implements the [`ArrowFloatType`](crate::floats::ArrowFloatType) trait for bfloat16 values.
+#[derive(Debug)]
+pub struct BFloat16Type {}
+
+/// An array of bfloat16 values
+///
+/// Note that bfloat16 is not the same thing as fp16 which is supported natively by arrow-rs.
+#[derive(Clone)]
+pub struct BFloat16Array {
+    inner: FixedSizeBinaryArray,
+}
+
+impl std::fmt::Debug for BFloat16Array {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "BFloat16Array\n[\n")?;
+        from_arrow::print_long_array(&self.inner, f, |array, i, f| {
+            if array.is_null(i) {
+                write!(f, "null")
+            } else {
+                let binary_values = array.value(i);
+                let value =
+                    bf16::from_bits(u16::from_le_bytes([binary_values[0], binary_values[1]]));
+                write!(f, "{:?}", value)
+            }
+        })?;
+        write!(f, "]")
+    }
+}
+
+impl BFloat16Array {
+    pub fn from_iter_values(iter: impl IntoIterator<Item = bf16>) -> Self {
+        let values: Vec<bf16> = iter.into_iter().collect();
+        values.into()
+    }
+
+    pub fn len(&self) -> usize {
+        self.inner.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.inner.is_empty()
+    }
+
+    pub fn is_null(&self, i: usize) -> bool {
+        self.inner.is_null(i)
+    }
+
+    pub fn null_count(&self) -> usize {
+        self.inner.null_count()
+    }
+
+    pub fn iter(&self) -> BFloat16Iter<'_> {
+        BFloat16Iter {
+            array: self,
+            index: 0,
+        }
+    }
+
+    pub fn value(&self, i: usize) -> bf16 {
+        assert!(
+            i < self.len(),
+            "Trying to access an element at index {} from a BFloat16Array of length {}",
+            i,
+            self.len()
+        );
+        // Safety:
+        // `i < self.len()
+        unsafe { self.value_unchecked(i) }
+    }
+
+    /// # Safety
+    /// Caller must ensure that `i < self.len()`
+    pub unsafe fn value_unchecked(&self, i: usize) -> bf16 {
+        // SAFETY: the caller guarantees that `i` is within this array's bounds.
+        let binary_value = unsafe { self.inner.value_unchecked(i) };
+        bf16::from_bits(u16::from_le_bytes([binary_value[0], binary_value[1]]))
+    }
+
+    pub fn into_inner(self) -> FixedSizeBinaryArray {
+        self.inner
+    }
+}
+
+impl FromIterator<Option<bf16>> for BFloat16Array {
+    fn from_iter<I: IntoIterator<Item = Option<bf16>>>(iter: I) -> Self {
+        let mut buffer = MutableBuffer::new(10);
+        // No null buffer builder :(
+        let mut nulls = BooleanBufferBuilder::new(10);
+        let mut len = 0;
+
+        for maybe_value in iter {
+            if let Some(value) = maybe_value {
+                let bytes = value.to_le_bytes();
+                buffer.extend(bytes);
+            } else {
+                buffer.extend([0u8, 0u8]);
+            }
+            nulls.append(maybe_value.is_some());
+            len += 1;
+        }
+
+        let null_buffer = nulls.finish();
+        let num_valid = null_buffer.count_set_bits();
+        let null_buffer = if num_valid == len {
+            None
+        } else {
+            Some(null_buffer.into_inner())
+        };
+
+        let array_data = ArrayData::builder(DataType::FixedSizeBinary(2))
+            .len(len)
+            .add_buffer(buffer.into())
+            .null_bit_buffer(null_buffer);
+        // SAFETY: the value buffer contains exactly `2 * len` bytes (two bytes
+        // pushed per iteration of the loop above, including the zero-fill for
+        // null slots), which matches the `FixedSizeBinary(2)` storage layout.
+        // The null bit buffer, when present, has `len` bits appended above, so
+        // its length covers the array's logical range.
+        let array_data = unsafe { array_data.build_unchecked() };
+        Self {
+            inner: FixedSizeBinaryArray::from(array_data),
+        }
+    }
+}
+
+impl FromIterator<bf16> for BFloat16Array {
+    fn from_iter<I: IntoIterator<Item = bf16>>(iter: I) -> Self {
+        Self::from_iter_values(iter)
+    }
+}
+
+impl From<Vec<bf16>> for BFloat16Array {
+    fn from(data: Vec<bf16>) -> Self {
+        let len = data.len();
+        // Zero-copy: `bf16` is `#[repr(transparent)]` over `u16` and derives
+        // `bytemuck::Pod`, so `cast_vec` reinterprets the allocation in place —
+        // no per-element copy or heap alloc. The crate-root `compile_error!`
+        // pins `target_endian = "little"`, so the resulting bytes match the
+        // `FixedSizeBinary(2)` on-disk order Lance writes elsewhere.
+        let raw: Vec<u16> = bytemuck::cast_vec(data);
+        let array_data = ArrayData::builder(DataType::FixedSizeBinary(2))
+            .len(len)
+            .add_buffer(Buffer::from_vec(raw));
+        // SAFETY: the value buffer contains exactly `2 * len` bytes — one
+        // `u16` per element after the layout-compatible cast — matching the
+        // `FixedSizeBinary(2)` storage layout. No null buffer is attached, so
+        // every element is logically valid.
+        let array_data = unsafe { array_data.build_unchecked() };
+        Self {
+            inner: FixedSizeBinaryArray::from(array_data),
+        }
+    }
+}
+
+impl TryFrom<FixedSizeBinaryArray> for BFloat16Array {
+    type Error = ArrowError;
+
+    fn try_from(value: FixedSizeBinaryArray) -> Result<Self, Self::Error> {
+        if value.value_length() == 2 {
+            Ok(Self { inner: value })
+        } else {
+            Err(ArrowError::InvalidArgumentError(
+                "FixedSizeBinaryArray must have a value length of 2".to_string(),
+            ))
+        }
+    }
+}
+
+impl PartialEq<Self> for BFloat16Array {
+    fn eq(&self, other: &Self) -> bool {
+        self.inner.eq(&other.inner)
+    }
+}
+
+pub struct BFloat16Iter<'a> {
+    array: &'a BFloat16Array,
+    index: usize,
+}
+
+impl<'a> Iterator for BFloat16Iter<'a> {
+    type Item = Option<bf16>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.index >= self.array.len() {
+            return None;
+        }
+        let i = self.index;
+        self.index += 1;
+        if self.array.is_null(i) {
+            Some(None)
+        } else {
+            Some(Some(self.array.value(i)))
+        }
+    }
+}
+
+/// Methods that are lifted from arrow-rs temporarily until they are made public.
+mod from_arrow {
+    use arrow_array::Array;
+
+    /// Helper function for printing potentially long arrays.
+    pub(super) fn print_long_array<A, F>(
+        array: &A,
+        f: &mut std::fmt::Formatter,
+        print_item: F,
+    ) -> std::fmt::Result
+    where
+        A: Array,
+        F: Fn(&A, usize, &mut std::fmt::Formatter) -> std::fmt::Result,
+    {
+        let head = std::cmp::min(10, array.len());
+
+        for i in 0..head {
+            if array.is_null(i) {
+                writeln!(f, "  null,")?;
+            } else {
+                write!(f, "  ")?;
+                print_item(array, i, f)?;
+                writeln!(f, ",")?;
+            }
+        }
+        if array.len() > 10 {
+            if array.len() > 20 {
+                writeln!(f, "  ...{} elements...,", array.len() - 20)?;
+            }
+
+            let tail = std::cmp::max(head, array.len() - 10);
+
+            for i in tail..array.len() {
+                if array.is_null(i) {
+                    writeln!(f, "  null,")?;
+                } else {
+                    write!(f, "  ")?;
+                    print_item(array, i, f)?;
+                    writeln!(f, ",")?;
+                }
+            }
+        }
+        Ok(())
+    }
+}
+
+impl FloatArray<BFloat16Type> for FixedSizeBinaryArray {
+    type FloatType = BFloat16Type;
+
+    /// Returns the underlying `bf16` values as a borrowed slice.
+    ///
+    /// # Preconditions
+    ///
+    /// - `value_length()` must be 2 (the `FixedSizeBinary(2)` storage shape
+    ///   used by [`BFloat16Array`]). Asserted at entry.
+    /// - The value buffer must be at least 2-byte aligned. Lance's in-tree
+    ///   constructors always satisfy this: value buffers are built either via
+    ///   `MutableBuffer` (aligned to arrow-buffer's `ALIGNMENT` constant, ≥32
+    ///   bytes) or via `Buffer::from_vec::<u16>` (aligned to `align_of::<u16>()`
+    ///   == 2); both meet `bf16`'s 2-byte requirement. Externally-built
+    ///   `FixedSizeBinaryArray`s arriving via FFI, IPC, or
+    ///   `Buffer::from_custom_allocation` are not required by arrow-rs to be
+    ///   aligned beyond a single byte; passing one to this method violates the
+    ///   precondition. A `debug_assert` below catches such inputs in debug and
+    ///   test builds.
+    ///
+    /// # Endianness
+    ///
+    /// `lance-arrow` is gated on `target_endian = "little"` at the crate root,
+    /// so this method always returns values in the same byte order Lance writes
+    /// (see [`BFloat16Array::value`] and the [`FromIterator`] impls).
+    fn as_slice(&self) -> &[bf16] {
+        assert_eq!(
+            self.value_length(),
+            2,
+            "BFloat16 arrays must use FixedSizeBinary(2) storage"
+        );
+        debug_assert_eq!(
+            (self.value_data().as_ptr() as usize) % std::mem::align_of::<bf16>(),
+            0,
+            "BFloat16 value buffer must be at least 2-byte aligned"
+        );
+        // SAFETY:
+        // - The assert above pins `value_size == 2`, so `value_data().len() / 2`
+        //   equals the array's logical element count.
+        //   `FixedSizeBinaryArray::From<ArrayData>` constructs its value buffer
+        //   as `buffers[0].slice_with_length(offset * 2, len * 2)` (arrow-array
+        //   `fixed_size_binary_array.rs`), so `value_data()` already returns
+        //   the offset-adjusted slice. Do not replace `value_data()` with an
+        //   accessor that returns the un-sliced backing buffer.
+        // - `bf16` is `#[repr(transparent)]` over `u16` (size 2, alignment 2);
+        //   every `u16` bit pattern is a valid `bf16`, so any byte content
+        //   yields a defined value — never UB.
+        // - Alignment is the caller's responsibility per the precondition
+        //   documented above. The `debug_assert_eq!` immediately preceding this
+        //   block catches violations in debug and test builds only — release
+        //   builds rely on callers honoring the precondition. arrow-rs
+        //   declares `FixedSizeBinary(n)`'s
+        //   `BufferSpec::FixedWidth { alignment: align_of::<u8>() == 1 }`
+        //   (arrow-data `data.rs`), so arrow-rs alone does not guarantee
+        //   2-byte alignment. Lance's in-tree construction paths build value
+        //   buffers via `MutableBuffer` (arrow-buffer `ALIGNMENT` constant,
+        //   ≥32 bytes) or `Buffer::from_vec::<u16>` (2-byte aligned), both of
+        //   which satisfy `bf16`'s 2-byte requirement.
+        // - The returned slice borrows from `self`; the underlying ref-counted,
+        //   immutable Arrow buffer cannot be mutated or freed for the slice's
+        //   lifetime.
+        unsafe {
+            slice::from_raw_parts(
+                self.value_data().as_ptr() as *const bf16,
+                self.value_data().len() / 2,
+            )
+        }
+    }
+
+    fn from_values(values: Vec<bf16>) -> Self {
+        BFloat16Array::from(values).into_inner()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_basics() {
+        let values: Vec<f32> = vec![1.0, 2.0, 3.0];
+        let values: Vec<bf16> = values.iter().map(|v| bf16::from_f32(*v)).collect();
+
+        let array = BFloat16Array::from_iter_values(values.clone());
+        let array2 = BFloat16Array::from(values.clone());
+        assert_eq!(array, array2);
+        assert_eq!(array.len(), 3);
+
+        // Pin the raw little-endian bytes emitted by `From<Vec<bf16>>` (rewritten to
+        // reinterpret the Vec via `bytemuck::cast_vec`), so a layout/byte-order
+        // regression is caught directly rather than only through Debug formatting.
+        // bf16 is the high 16 bits of the f32: 1.0->0x3F80, 2.0->0x4000, 3.0->0x4040.
+        let inner = array2.clone().into_inner();
+        let raw_bytes: Vec<u8> = (0..inner.len())
+            .flat_map(|i| inner.value(i).to_vec())
+            .collect();
+        assert_eq!(raw_bytes, vec![0x80, 0x3F, 0x00, 0x40, 0x40, 0x40]);
+
+        let expected_fmt = "BFloat16Array\n[\n  1.0,\n  2.0,\n  3.0,\n]";
+        assert_eq!(expected_fmt, format!("{:?}", array));
+
+        for (expected, value) in values.iter().zip(array.iter()) {
+            assert_eq!(Some(*expected), value);
+        }
+
+        for (expected, value) in values.as_slice().iter().zip(array2.iter()) {
+            assert_eq!(Some(*expected), value);
+        }
+
+        let arrow_array = array.into_inner();
+        assert_eq!(arrow_array.as_slice(), values.as_slice());
+    }
+
+    #[test]
+    fn test_nulls() {
+        let values: Vec<Option<bf16>> =
+            vec![Some(bf16::from_f32(1.0)), None, Some(bf16::from_f32(3.0))];
+        let array = BFloat16Array::from_iter(values.clone());
+        assert_eq!(array.len(), 3);
+        assert_eq!(array.null_count(), 1);
+
+        let expected_fmt = "BFloat16Array\n[\n  1.0,\n  null,\n  3.0,\n]";
+        assert_eq!(expected_fmt, format!("{:?}", array));
+
+        for (expected, value) in values.iter().zip(array.iter()) {
+            assert_eq!(*expected, value);
+        }
+    }
+}

python/vendor/lance-arrow/src/deepcopy.rs

@@ -0,0 +1,355 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+
+use std::sync::Arc;
+
+use arrow_array::{Array, RecordBatch, make_array};
+use arrow_buffer::{BooleanBuffer, Buffer, NullBuffer};
+use arrow_data::{ArrayData, ArrayDataBuilder, transform::MutableArrayData};
+use arrow_schema::DataType;
+
+pub fn deep_copy_buffer(buffer: &Buffer) -> Buffer {
+    Buffer::from(buffer.as_slice())
+}
+
+pub fn deep_copy_nulls(nulls: Option<&NullBuffer>) -> Option<NullBuffer> {
+    let nulls = nulls?;
+    let bit_buffer = deep_copy_buffer(nulls.inner().inner());
+    // SAFETY: `null_count` is taken from the source `NullBuffer`, which already
+    // upheld `NullBuffer::new_unchecked`'s invariant — the unset-bit count over
+    // the logical bit slice `[bit_offset, bit_offset + bit_len)`. `NullBuffer::slice`
+    // adjusts only `BooleanBuffer::bit_offset` / `bit_len` and never byte-advances
+    // the inner `Buffer`, so `deep_copy_buffer` (which copies the source `Buffer`'s
+    // `as_slice()` view from byte 0) reproduces the exact bit pattern at the same
+    // bit offsets; the unset-bit count is therefore preserved. `BooleanBuffer::new`
+    // panics (does not UB) if `bit_offset + bit_len > 8 * buffer.len()`, and the
+    // copy has the same length, so that check still passes.
+    Some(unsafe {
+        NullBuffer::new_unchecked(
+            BooleanBuffer::new(bit_buffer, nulls.offset(), nulls.len()),
+            nulls.null_count(),
+        )
+    })
+}
+
+pub fn deep_copy_array_data(data: &ArrayData) -> ArrayData {
+    let data_type = data.data_type().clone();
+    let len = data.len();
+    let nulls = deep_copy_nulls(data.nulls());
+    let offset = data.offset();
+    let buffers = data
+        .buffers()
+        .iter()
+        .map(deep_copy_buffer)
+        .collect::<Vec<_>>();
+    let child_data = data
+        .child_data()
+        .iter()
+        .map(deep_copy_array_data)
+        .collect::<Vec<_>>();
+    // SAFETY: `build_unchecked` inherits `ArrayData::new_unchecked`'s contract —
+    // `(data_type, len, offset, nulls, buffers, child_data)` must form a valid
+    // Arrow array. This call reproduces `data` structurally: `data_type`, `len`,
+    // and `offset` are forwarded unchanged; each buffer is replaced by a byte-
+    // identical copy of its offset-applied `as_slice()` view (the output buffer
+    // is `MutableBuffer`-allocated, at least as aligned as the source); `nulls`
+    // is deep-copied with the same bit offset/length and unset-bit count (see
+    // `deep_copy_nulls`); `child_data` is recursively cloned with the same
+    // guarantee. Every value-level invariant the source upheld — UTF-8 validity,
+    // monotonic offsets, in-bounds dictionary indices, run-end monotonicity,
+    // struct child-length matching — therefore transfers to the copy. If the
+    // source `ArrayData` was itself constructed via `new_unchecked` with an
+    // invalid payload, this function faithfully reproduces that invalidity.
+    unsafe {
+        ArrayDataBuilder::new(data_type)
+            .len(len)
+            .nulls(nulls)
+            .offset(offset)
+            .buffers(buffers)
+            .child_data(child_data)
+            .build_unchecked()
+    }
+}
+
+pub fn deep_copy_array(array: &dyn Array) -> Arc<dyn Array> {
+    let data = array.to_data();
+    let data = deep_copy_array_data(&data);
+    make_array(data)
+}
+
+pub fn deep_copy_batch(batch: &RecordBatch) -> crate::Result<RecordBatch> {
+    let arrays = batch
+        .columns()
+        .iter()
+        .map(|array| deep_copy_array(array))
+        .collect::<Vec<_>>();
+    RecordBatch::try_new(batch.schema(), arrays)
+}
+
+/// Deep copy array data, extracting only the sliced portion using MutableArrayData
+/// This is the most efficient and correct way to copy just the sliced data
+pub fn deep_copy_array_data_sliced(data: &ArrayData) -> ArrayData {
+    // Use MutableArrayData to efficiently copy just the slice
+    let mut mutable = MutableArrayData::new(vec![data], false, data.len());
+
+    // Which index this takes depends on the layout, because arrow's extenders
+    // disagree about who applies `ArrayData::offset()`. The bit-packed one adds
+    // it to the raw values buffer itself, so an offset-applied index there
+    // reads from twice the offset and runs off the end of the buffer. Every
+    // other layout either reads through an already-offset-applied view or
+    // forwards the index to children carrying their own offsets, and takes the
+    // absolute index it has always been given.
+    let start = match data.data_type() {
+        DataType::Boolean => 0,
+        _ => data.offset(),
+    };
+    mutable.extend(0, start, start + data.len());
+
+    // Freeze into immutable ArrayData
+    mutable.freeze()
+}
+
+/// Deep copy an array, extracting only the sliced portion using MutableArrayData
+pub fn deep_copy_array_sliced(array: &dyn Array) -> Arc<dyn Array> {
+    let data = array.to_data();
+    let data = deep_copy_array_data_sliced(&data);
+    make_array(data)
+}
+
+/// Deep copy a RecordBatch, extracting only the sliced portion using MutableArrayData
+pub fn deep_copy_batch_sliced(batch: &RecordBatch) -> crate::Result<RecordBatch> {
+    let arrays = batch
+        .columns()
+        .iter()
+        .map(|array| deep_copy_array_sliced(array))
+        .collect::<Vec<_>>();
+    RecordBatch::try_new(batch.schema(), arrays)
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use arrow_array::{Array, BooleanArray, Int32Array, RecordBatch, StringArray};
+    use arrow_data::ArrayDataBuilder;
+    use arrow_schema::{DataType, Field, Schema};
+
+    #[test]
+    fn raw_sliced_fixed_size_list_data_keeps_its_child_offset() {
+        // `ArrayData::slice` records the offset on the parent and leaves the
+        // child whole -- a shape `FixedSizeListArray::slice` never produces,
+        // but a legal one this public helper can be handed directly. Arrow's
+        // extender forwards `start * size` to that unsliced child and adds
+        // nothing, so the index it gets has to be the absolute one; this is
+        // here to stop the boolean fix from being generalised over it.
+        let child = Int32Array::from(vec![10, 11, 20, 21]).to_data();
+        let field = Arc::new(Field::new_list_field(DataType::Int32, false));
+        let data = ArrayDataBuilder::new(DataType::FixedSizeList(field, 2))
+            .len(2)
+            .add_child_data(child)
+            .build()
+            .unwrap();
+        let sliced = data.slice(1, 1);
+
+        let copied = super::deep_copy_array_data_sliced(&sliced);
+        let copied_child = Int32Array::from(copied.child_data()[0].clone());
+        assert_eq!(copied_child.values().as_ref(), &[20, 21]);
+    }
+
+    #[test]
+    fn raw_sliced_struct_of_fixed_size_list_reaches_the_right_values() {
+        // Slicing raw struct data pushes the window into the struct's children,
+        // and a fixed-size-list child takes it as its own parent offset with
+        // its values left whole. Two levels of offset, and the absolute index
+        // has to remain right through both of them.
+        let values = Int32Array::from(vec![10, 11, 20, 21, 30, 31]).to_data();
+        let item = Arc::new(Field::new_list_field(DataType::Int32, false));
+        let list = ArrayDataBuilder::new(DataType::FixedSizeList(item, 2))
+            .len(3)
+            .add_child_data(values)
+            .build()
+            .unwrap();
+        let field = Arc::new(Field::new("l", list.data_type().clone(), false));
+        let data = ArrayDataBuilder::new(DataType::Struct(vec![field].into()))
+            .len(3)
+            .add_child_data(list)
+            .build()
+            .unwrap();
+        let sliced = data.slice(2, 1);
+
+        let copied = super::deep_copy_array_data_sliced(&sliced);
+        let copied_values = Int32Array::from(copied.child_data()[0].child_data()[0].clone());
+        assert_eq!(copied_values.values().as_ref(), &[30, 31]);
+    }
+
+    #[test]
+    fn raw_parent_offset_struct_keeps_the_selected_child_row() {
+        // A struct whose children are whole and whose window is the parent
+        // offset: arrow's struct extender forwards the index to those children
+        // untouched, so it has to be the absolute one.
+        let field = Arc::new(Field::new("a", DataType::Int32, false));
+        let data = ArrayDataBuilder::new(DataType::Struct(vec![field].into()))
+            .len(1)
+            .offset(1)
+            .add_child_data(Int32Array::from(vec![10, 20]).to_data())
+            .build()
+            .unwrap();
+
+        let copied = super::deep_copy_array_data_sliced(&data);
+        let copied_child = Int32Array::from(copied.child_data()[0].clone());
+        assert_eq!(copied_child.values().as_ref(), &[20]);
+    }
+
+    #[test]
+    fn sliced_boolean_deep_copy_reads_from_the_slice() {
+        // A boolean slice keeps its `ArrayData::offset()` -- the buffer cannot
+        // advance by a fraction of a byte -- so a copy that adds the offset on
+        // top of what `MutableArrayData` already applies reads past the end of
+        // the values buffer and panics.
+        let array = BooleanArray::from(vec![true, false, true, false, true, false, true, false]);
+        for (offset, len) in [(0usize, 8usize), (1, 7), (3, 5), (7, 1)] {
+            let sliced = array.slice(offset, len);
+            let copied = super::deep_copy_array_sliced(&sliced);
+            let copied = copied.as_any().downcast_ref::<BooleanArray>().unwrap();
+            let expected: Vec<bool> = (0..len).map(|i| sliced.value(i)).collect();
+            let actual: Vec<bool> = (0..len).map(|i| copied.value(i)).collect();
+            assert_eq!(actual, expected, "offset={offset} len={len}");
+        }
+    }
+
+    #[test]
+    fn sliced_boolean_deep_copy_keeps_its_nulls() {
+        let array = BooleanArray::from(vec![
+            Some(true),
+            None,
+            Some(false),
+            Some(true),
+            None,
+            Some(false),
+        ]);
+        let sliced = array.slice(1, 4);
+        let copied = super::deep_copy_array_sliced(&sliced);
+        let copied = copied.as_any().downcast_ref::<BooleanArray>().unwrap();
+        let expected: Vec<Option<bool>> = (0..sliced.len())
+            .map(|i| (!sliced.is_null(i)).then(|| sliced.value(i)))
+            .collect();
+        let actual: Vec<Option<bool>> = (0..copied.len())
+            .map(|i| (!copied.is_null(i)).then(|| copied.value(i)))
+            .collect();
+        assert_eq!(actual, expected);
+    }
+
+    #[test]
+    fn test_deep_copy_sliced_array_with_nulls() {
+        let array = Arc::new(Int32Array::from(vec![
+            Some(1),
+            None,
+            Some(3),
+            None,
+            Some(5),
+        ]));
+        let sliced_array = array.slice(1, 3);
+        let copied_array = super::deep_copy_array(&sliced_array);
+        assert_eq!(sliced_array.len(), copied_array.len());
+        assert_eq!(sliced_array.nulls(), copied_array.nulls());
+    }
+
+    #[test]
+    fn test_deep_copy_array_data_sliced() {
+        let array = Int32Array::from((0..1000).collect::<Vec<i32>>());
+        let sliced = array.slice(100, 10);
+
+        let sliced_data = sliced.to_data();
+        let copied_data = super::deep_copy_array_data_sliced(&sliced_data);
+
+        assert_eq!(copied_data.len(), 10);
+        assert_eq!(copied_data.offset(), 0);
+
+        // Verify data correctness
+        let copied_array = Int32Array::from(copied_data);
+        for i in 0..10 {
+            assert_eq!(copied_array.value(i), 100 + i as i32);
+        }
+    }
+
+    #[test]
+    fn test_deep_copy_array_sliced() {
+        let array = Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5]));
+        let sliced = array.slice(1, 3);
+
+        let copied = super::deep_copy_array_sliced(&sliced);
+
+        assert_eq!(copied.len(), 3);
+        let copied_int = copied.as_any().downcast_ref::<Int32Array>().unwrap();
+        assert_eq!(copied_int.value(0), 2);
+        assert_eq!(copied_int.value(1), 3);
+        assert_eq!(copied_int.value(2), 4);
+    }
+
+    #[test]
+    fn test_deep_copy_batch_sliced() {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("id", DataType::Int32, false),
+            Field::new("name", DataType::Utf8, false),
+        ]));
+
+        let id_array = Arc::new(Int32Array::from((0..100).collect::<Vec<i32>>()));
+        let name_array = Arc::new(StringArray::from(
+            (0..100)
+                .map(|i| format!("name_{}", i))
+                .collect::<Vec<String>>(),
+        ));
+
+        let batch = RecordBatch::try_new(
+            schema,
+            vec![id_array as Arc<dyn Array>, name_array as Arc<dyn Array>],
+        )
+        .unwrap();
+
+        let sliced = batch.slice(10, 5);
+        let copied = super::deep_copy_batch_sliced(&sliced).unwrap();
+
+        assert_eq!(copied.num_rows(), 5);
+        assert_eq!(copied.num_columns(), 2);
+
+        // Verify data correctness
+        let id_col = copied
+            .column(0)
+            .as_any()
+            .downcast_ref::<Int32Array>()
+            .unwrap();
+        let name_col = copied
+            .column(1)
+            .as_any()
+            .downcast_ref::<StringArray>()
+            .unwrap();
+
+        for i in 0..5 {
+            assert_eq!(id_col.value(i), 10 + i as i32);
+            assert_eq!(name_col.value(i), format!("name_{}", 10 + i));
+        }
+    }
+
+    #[test]
+    fn test_deep_copy_array_sliced_with_nulls() {
+        let array = Arc::new(Int32Array::from(vec![
+            Some(1),
+            None,
+            Some(3),
+            None,
+            Some(5),
+        ]));
+        let sliced = array.slice(1, 3); // [None, Some(3), None]
+
+        let copied = super::deep_copy_array_sliced(&sliced);
+
+        assert_eq!(copied.len(), 3);
+        assert_eq!(copied.null_count(), 2); // Two nulls in the slice
+
+        let copied_int = copied.as_any().downcast_ref::<Int32Array>().unwrap();
+        assert!(!copied_int.is_valid(0)); // None
+        assert!(copied_int.is_valid(1)); // Some(3)
+        assert!(!copied_int.is_valid(2)); // None
+        assert_eq!(copied_int.value(1), 3);
+    }
+}

python/vendor/lance-arrow/src/floats.rs

@@ -0,0 +1,291 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+
+//! Floats Array
+
+use std::fmt::{Debug, Display};
+use std::iter::Sum;
+use std::sync::Arc;
+use std::{
+    fmt::Formatter,
+    ops::{AddAssign, DivAssign},
+};
+
+use arrow_array::{
+    Array, FixedSizeBinaryArray, Float16Array, Float32Array, Float64Array,
+    types::{Float16Type, Float32Type, Float64Type},
+};
+use arrow_schema::{DataType, Field};
+use half::{bf16, f16};
+use num_traits::{AsPrimitive, Bounded, Float, FromPrimitive};
+
+use super::bfloat16::{BFloat16Array, BFloat16Type};
+use crate::Result;
+use crate::bfloat16::is_bfloat16_field;
+
+/// Float data type.
+///
+/// This helps differentiate between the different float types,
+/// because bf16 is not officially supported [DataType] in arrow-rs.
+#[derive(Debug)]
+pub enum FloatType {
+    BFloat16,
+    Float16,
+    Float32,
+    Float64,
+}
+
+impl std::fmt::Display for FloatType {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::BFloat16 => write!(f, "bfloat16"),
+            Self::Float16 => write!(f, "float16"),
+            Self::Float32 => write!(f, "float32"),
+            Self::Float64 => write!(f, "float64"),
+        }
+    }
+}
+
+/// Try to convert a [DataType] to a [FloatType]. To support bfloat16, always
+/// prefer using the `TryFrom<&Field>` implementation.
+impl TryFrom<&DataType> for FloatType {
+    type Error = crate::ArrowError;
+
+    fn try_from(value: &DataType) -> Result<Self> {
+        match *value {
+            DataType::Float16 => Ok(Self::Float16),
+            DataType::Float32 => Ok(Self::Float32),
+            DataType::Float64 => Ok(Self::Float64),
+            _ => Err(crate::ArrowError::InvalidArgumentError(format!(
+                "{:?} is not a floating type",
+                value
+            ))),
+        }
+    }
+}
+
+impl TryFrom<&Field> for FloatType {
+    type Error = crate::ArrowError;
+
+    fn try_from(field: &Field) -> Result<Self> {
+        match field.data_type() {
+            DataType::FixedSizeBinary(2) if is_bfloat16_field(field) => Ok(Self::BFloat16),
+            _ => Self::try_from(field.data_type()),
+        }
+    }
+}
+
+/// Trait for float types used in Lance indexes
+///
+/// This mimics the utilities provided by [`arrow_array::ArrowPrimitiveType`]
+/// but applies to all float types (including bfloat16)
+pub trait ArrowFloatType: Debug {
+    type Native: FromPrimitive
+        + FloatToArrayType<ArrowType = Self>
+        + AsPrimitive<f32>
+        + Debug
+        + Display;
+
+    const FLOAT_TYPE: FloatType;
+    const MIN: Self::Native;
+    const MAX: Self::Native;
+
+    /// Arrow Float Array Type.
+    type ArrayType: FloatArray<Self>;
+
+    /// Returns empty array of this type.
+    fn empty_array() -> Self::ArrayType {
+        <Self::ArrayType as FloatArray<Self>>::from_values(Vec::new())
+    }
+}
+
+/// Trait to be implemented by native types that have a corresponding [`ArrowFloatType`]
+/// implementation.
+///
+/// This helps define what operations are supported by native floats and also helps convert
+/// from a native type back to the corresponding Arrow float type.
+pub trait FloatToArrayType:
+    Float
+    + Bounded
+    + Sum
+    + AddAssign<Self>
+    + AsPrimitive<f64>
+    + AsPrimitive<f32>
+    + DivAssign
+    + Send
+    + Sync
+    + Copy
+{
+    /// The corresponding [`ArrowFloatType`] implementation for this native type
+    type ArrowType: ArrowFloatType<Native = Self>;
+}
+
+impl FloatToArrayType for bf16 {
+    type ArrowType = BFloat16Type;
+}
+
+impl FloatToArrayType for f16 {
+    type ArrowType = Float16Type;
+}
+
+impl FloatToArrayType for f32 {
+    type ArrowType = Float32Type;
+}
+
+impl FloatToArrayType for f64 {
+    type ArrowType = Float64Type;
+}
+
+impl ArrowFloatType for BFloat16Type {
+    type Native = bf16;
+
+    const FLOAT_TYPE: FloatType = FloatType::BFloat16;
+    const MIN: Self::Native = bf16::MIN;
+    const MAX: Self::Native = bf16::MAX;
+
+    type ArrayType = FixedSizeBinaryArray;
+}
+
+impl ArrowFloatType for Float16Type {
+    type Native = f16;
+
+    const FLOAT_TYPE: FloatType = FloatType::Float16;
+    const MIN: Self::Native = f16::MIN;
+    const MAX: Self::Native = f16::MAX;
+
+    type ArrayType = Float16Array;
+}
+
+impl ArrowFloatType for Float32Type {
+    type Native = f32;
+
+    const FLOAT_TYPE: FloatType = FloatType::Float32;
+    const MIN: Self::Native = f32::MIN;
+    const MAX: Self::Native = f32::MAX;
+
+    type ArrayType = Float32Array;
+}
+
+impl ArrowFloatType for Float64Type {
+    type Native = f64;
+
+    const FLOAT_TYPE: FloatType = FloatType::Float64;
+    const MIN: Self::Native = f64::MIN;
+    const MAX: Self::Native = f64::MAX;
+
+    type ArrayType = Float64Array;
+}
+
+/// [FloatArray] is a trait that is implemented by all float type arrays
+///
+/// This is similar to [`arrow_array::PrimitiveArray`] but applies to all float types (including bfloat16)
+/// and is implemented as a trait and not a struct
+pub trait FloatArray<T: ArrowFloatType + ?Sized>: Array + Clone + 'static {
+    type FloatType: ArrowFloatType;
+
+    /// Returns a reference to the underlying data as a slice.
+    ///
+    /// # Panics
+    ///
+    /// Implementations may panic if the array's storage shape does not match
+    /// the expected element layout. In particular, the `bf16` impl panics if
+    /// `value_length() != 2` (the `FixedSizeBinary(2)` shape required by
+    /// `BFloat16Array`).
+    ///
+    /// # Preconditions
+    ///
+    /// Implementations may impose additional invariants on the underlying
+    /// buffer. The `bf16` impl requires the value buffer to be at least
+    /// 2-byte aligned — satisfied automatically by every in-tree Lance
+    /// constructor, but external callers passing externally-built arrays
+    /// (FFI, IPC, `Buffer::from_custom_allocation`) must ensure alignment.
+    /// See the impl's docstring for details.
+    fn as_slice(&self) -> &[T::Native];
+
+    /// Construct an array from a vector of values.
+    fn from_values(values: Vec<T::Native>) -> Self;
+
+    /// Construct an array from an iterator of values.
+    fn from_iter_values(values: impl IntoIterator<Item = T::Native>) -> Self
+    where
+        Self: Sized,
+    {
+        Self::from_values(values.into_iter().collect())
+    }
+}
+
+impl FloatArray<Float16Type> for Float16Array {
+    type FloatType = Float16Type;
+
+    fn as_slice(&self) -> &[<Float16Type as ArrowFloatType>::Native] {
+        self.values()
+    }
+
+    fn from_values(values: Vec<<Float16Type as ArrowFloatType>::Native>) -> Self {
+        Self::from(values)
+    }
+}
+
+impl FloatArray<Float32Type> for Float32Array {
+    type FloatType = Float32Type;
+
+    fn as_slice(&self) -> &[<Float32Type as ArrowFloatType>::Native] {
+        self.values()
+    }
+
+    fn from_values(values: Vec<<Float32Type as ArrowFloatType>::Native>) -> Self {
+        Self::from(values)
+    }
+}
+
+impl FloatArray<Float64Type> for Float64Array {
+    type FloatType = Float64Type;
+
+    fn as_slice(&self) -> &[<Float64Type as ArrowFloatType>::Native] {
+        self.values()
+    }
+
+    fn from_values(values: Vec<<Float64Type as ArrowFloatType>::Native>) -> Self {
+        Self::from(values)
+    }
+}
+
+/// Convert a float32 array to another float array
+///
+/// This is used during queries as query vectors are always provided as float32 arrays
+/// and need to be converted to the appropriate float type for the index.
+pub fn coerce_float_vector(input: &Float32Array, float_type: FloatType) -> Result<Arc<dyn Array>> {
+    match float_type {
+        FloatType::BFloat16 => Ok(Arc::new(
+            BFloat16Array::from_iter_values(input.values().iter().map(|v| bf16::from_f32(*v)))
+                .into_inner(),
+        )),
+        FloatType::Float16 => Ok(Arc::new(Float16Array::from_iter_values(
+            input.values().iter().map(|v| f16::from_f32(*v)),
+        ))),
+        FloatType::Float32 => Ok(Arc::new(input.clone())),
+        FloatType::Float64 => Ok(Arc::new(Float64Array::from_iter_values(
+            input.values().iter().map(|v| *v as f64),
+        ))),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_coerce_float_vector_bfloat16() {
+        let input = Float32Array::from(vec![1.0f32, 2.0, 3.0]);
+        let array = coerce_float_vector(&input, FloatType::BFloat16).unwrap();
+
+        assert_eq!(array.data_type(), &DataType::FixedSizeBinary(2));
+
+        let fixed = array
+            .as_any()
+            .downcast_ref::<FixedSizeBinaryArray>()
+            .unwrap();
+        let expected: Vec<bf16> = input.values().iter().map(|v| bf16::from_f32(*v)).collect();
+        assert_eq!(fixed.as_slice(), expected.as_slice());
+    }
+}

python/vendor/lance-arrow/src/ipc.rs

@@ -0,0 +1,620 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+
+//! Zero-copy Arrow IPC stream read/write utilities.
+//!
+//! Provides helpers for serializing and deserializing [`RecordBatch`]es as
+//! self-delimiting Arrow IPC streams using synchronous [`Read`]/[`Write`] I/O.
+//!
+//! These are designed for embedding IPC streams inside larger binary formats
+//! (e.g. a cache entry that contains multiple IPC sections). Each stream is
+//! self-delimiting (schema + batches + EOS marker) and can be read back
+//! independently.
+//!
+//! # Zero-copy reads
+//!
+//! [`read_ipc_stream`] and [`read_ipc_stream_single`] take `&Bytes` and use
+//! [`Bytes::slice`] to produce each message buffer. Because `Bytes::slice`
+//! increments a reference count rather than copying, the resulting
+//! [`Buffer`]s — and the array data decoded from them by [`FileDecoder`] —
+//! are all backed by the same allocation as the input.
+
+use std::io::{Read, Write};
+use std::sync::Arc;
+
+use arrow_array::RecordBatch;
+use arrow_buffer::Buffer;
+use arrow_ipc::convert::fb_to_schema;
+use arrow_ipc::reader::FileDecoder;
+use arrow_ipc::root_as_message;
+use arrow_ipc::writer::StreamWriter;
+use arrow_schema::ArrowError;
+use bytes::Bytes;
+
+// ---------------------------------------------------------------------------
+// Length-prefixed byte utilities
+// ---------------------------------------------------------------------------
+
+/// Write `data` prefixed by its length as a little-endian `u64`.
+///
+/// Paired with [`read_len_prefixed_bytes`].
+pub fn write_len_prefixed_bytes(writer: &mut dyn Write, data: &[u8]) -> Result<(), ArrowError> {
+    writer
+        .write_all(&(data.len() as u64).to_le_bytes())
+        .map_err(|e| ArrowError::IoError(e.to_string(), e))?;
+    writer
+        .write_all(data)
+        .map_err(|e| ArrowError::IoError(e.to_string(), e))
+}
+
+/// Read a byte slice written by [`write_len_prefixed_bytes`].
+///
+/// Reads an 8-byte little-endian length then exactly that many bytes.
+pub fn read_len_prefixed_bytes(reader: &mut dyn Read) -> Result<Vec<u8>, ArrowError> {
+    let mut len_buf = [0u8; 8];
+    reader
+        .read_exact(&mut len_buf)
+        .map_err(|e| ArrowError::IoError(e.to_string(), e))?;
+    let len = u64::from_le_bytes(len_buf) as usize;
+    let mut buf = vec![0u8; len];
+    reader
+        .read_exact(&mut buf)
+        .map_err(|e| ArrowError::IoError(e.to_string(), e))?;
+    Ok(buf)
+}
+
+// ---------------------------------------------------------------------------
+// IPC stream utilities
+// ---------------------------------------------------------------------------
+
+// 4-byte continuation marker used by modern Arrow IPC streams.
+const IPC_CONTINUATION: [u8; 4] = [0xff; 4];
+
+/// Write `batch` as a single-batch Arrow IPC stream to `writer`.
+pub fn write_ipc_stream(batch: &RecordBatch, writer: &mut dyn Write) -> Result<(), ArrowError> {
+    let mut sw = StreamWriter::try_new(&mut *writer, batch.schema_ref())?;
+    sw.write(batch)?;
+    sw.finish()
+}
+
+/// Write all batches from `iter` as a single Arrow IPC stream to `writer`.
+///
+/// `iter` must yield at least one batch; the schema is inferred from the first
+/// batch. Returns `ArrowError::InvalidArgumentError` if the iterator is empty.
+/// If you need to write an empty stream (schema only, no rows), construct a
+/// `StreamWriter` directly.
+pub fn write_ipc_stream_batches<I>(iter: I, writer: &mut dyn Write) -> Result<(), ArrowError>
+where
+    I: IntoIterator<Item = RecordBatch>,
+{
+    let mut iter = iter.into_iter();
+    let first = iter
+        .next()
+        .ok_or_else(|| ArrowError::InvalidArgumentError("no batches to serialize".into()))?;
+    let mut sw = StreamWriter::try_new(&mut *writer, first.schema_ref())?;
+    sw.write(&first)?;
+    for batch in iter {
+        sw.write(&batch)?;
+    }
+    sw.finish()
+}
+
+/// Read one complete Arrow IPC stream message from `data` as a zero-copy [`Buffer`].
+///
+/// Parses the first message starting at byte 0 of `data`. Returns `None` on
+/// EOS (size field == 0) or empty input. The returned [`Buffer`] is backed by
+/// `data`'s allocation — no bytes are copied.
+///
+/// The caller should advance its position by `buf.len()` after each call.
+fn read_one_ipc_message(data: &Bytes) -> Result<Option<Buffer>, ArrowError> {
+    let bytes = data.as_ref();
+
+    if bytes.is_empty() {
+        return Ok(None);
+    }
+    if bytes.len() < 4 {
+        return Err(ArrowError::IoError(
+            "IPC: truncated header".into(),
+            std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "truncated IPC header"),
+        ));
+    }
+
+    let has_continuation = bytes[..4] == IPC_CONTINUATION;
+    let (size_bytes, prefix_len): ([u8; 4], usize) = if has_continuation {
+        if bytes.len() < 8 {
+            return Err(ArrowError::IoError(
+                "IPC: truncated header after continuation".into(),
+                std::io::Error::new(
+                    std::io::ErrorKind::UnexpectedEof,
+                    "truncated after continuation",
+                ),
+            ));
+        }
+        (bytes[4..8].try_into().unwrap(), 8)
+    } else {
+        (bytes[..4].try_into().unwrap(), 4)
+    };
+
+    let meta_size = u32::from_le_bytes(size_bytes) as usize;
+    if meta_size == 0 {
+        return Ok(None); // EOS
+    }
+
+    let meta_end = prefix_len + meta_size;
+    if bytes.len() < meta_end {
+        return Err(ArrowError::IoError(
+            "IPC: truncated metadata".into(),
+            std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "truncated IPC metadata"),
+        ));
+    }
+
+    let msg = root_as_message(&bytes[prefix_len..meta_end])
+        .map_err(|e| ArrowError::ParseError(format!("IPC message parse error: {e}")))?;
+    let body_len = msg.bodyLength() as usize;
+
+    let total = meta_end + body_len;
+    if bytes.len() < total {
+        return Err(ArrowError::IoError(
+            "IPC: truncated body".into(),
+            std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "truncated IPC body"),
+        ));
+    }
+
+    // Zero-copy: Bytes::slice shares the backing allocation; Buffer::from
+    // wraps it without copying.
+    Ok(Some(Buffer::from(data.slice(0..total))))
+}
+
+/// Read a length-prefixed byte slice at `offset` in `data`, advancing `offset`.
+///
+/// Reads an 8-byte little-endian length, then slices exactly that many bytes
+/// from `data`. The returned [`Bytes`] is zero-copy (shares `data`'s allocation).
+pub fn read_len_prefixed_bytes_at(data: &Bytes, offset: &mut usize) -> Result<Bytes, ArrowError> {
+    let bytes = data.as_ref();
+    let len_end = offset
+        .checked_add(8)
+        .filter(|&e| e <= bytes.len())
+        .ok_or_else(|| {
+            ArrowError::IoError(
+                "length-prefixed bytes: truncated length field".into(),
+                std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "truncated length"),
+            )
+        })?;
+    let len = u64::from_le_bytes(bytes[*offset..len_end].try_into().unwrap()) as usize;
+    *offset = len_end;
+    let data_end = offset
+        .checked_add(len)
+        .filter(|&e| e <= bytes.len())
+        .ok_or_else(|| {
+            ArrowError::IoError(
+                "length-prefixed bytes: truncated data".into(),
+                std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "truncated data"),
+            )
+        })?;
+    let result = data.slice(*offset..data_end);
+    *offset = data_end;
+    Ok(result)
+}
+
+/// Read all [`RecordBatch`]es from one Arrow IPC stream starting at `offset`,
+/// advancing `offset` past the stream (including the EOS marker).
+///
+/// Zero-copy: array buffers borrow from `data`'s allocation.
+pub fn read_ipc_stream_at(
+    data: &Bytes,
+    offset: &mut usize,
+) -> Result<Vec<RecordBatch>, ArrowError> {
+    let batches = read_ipc_stream(&data.slice(*offset..))?;
+
+    // Recompute how many bytes were consumed by re-parsing message sizes.
+    // We can't get this from read_ipc_stream directly, so we re-walk the
+    // message headers (metadata only, no body re-read) to sum up lengths.
+    let slice = &data.as_ref()[*offset..];
+    let mut consumed = 0usize;
+    loop {
+        let rem = &slice[consumed..];
+        if rem.is_empty() {
+            break;
+        }
+        let has_cont = rem.len() >= 4 && rem[..4] == IPC_CONTINUATION;
+        let (size_bytes, prefix_len): ([u8; 4], usize) = if has_cont {
+            if rem.len() < 8 {
+                break;
+            }
+            (rem[4..8].try_into().unwrap(), 8)
+        } else {
+            if rem.len() < 4 {
+                break;
+            }
+            (rem[..4].try_into().unwrap(), 4)
+        };
+        let meta_size = u32::from_le_bytes(size_bytes) as usize;
+        if meta_size == 0 {
+            // EOS — consume it and stop.
+            consumed += prefix_len;
+            break;
+        }
+        let meta_end = prefix_len + meta_size;
+        if rem.len() < meta_end {
+            break;
+        }
+        let msg = root_as_message(&rem[prefix_len..meta_end])
+            .map_err(|e| ArrowError::ParseError(format!("IPC message parse error: {e}")))?;
+        let body_len = msg.bodyLength() as usize;
+        consumed += meta_end + body_len;
+    }
+    *offset += consumed;
+
+    Ok(batches)
+}
+
+/// Read exactly one [`RecordBatch`] from one Arrow IPC stream starting at `offset`,
+/// advancing `offset` past the stream (including the EOS marker).
+///
+/// Zero-copy: array buffers borrow from `data`'s allocation.
+pub fn read_ipc_stream_single_at(
+    data: &Bytes,
+    offset: &mut usize,
+) -> Result<RecordBatch, ArrowError> {
+    let mut batches = read_ipc_stream_at(data, offset)?;
+    match batches.len() {
+        1 => Ok(batches.remove(0)),
+        n => Err(ArrowError::ParseError(format!(
+            "expected exactly 1 IPC record batch, got {n}"
+        ))),
+    }
+}
+
+/// Extract the prefix length and metadata size from a raw IPC message buffer.
+///
+/// Modern IPC streams have an 8-byte prefix `[continuation: 4][size: 4]`.
+/// Legacy streams have a 4-byte prefix `[size: 4]`. Returns `(prefix_len, meta_size)`.
+fn parse_ipc_message_prefix(buf: &Buffer) -> Result<(usize, usize), ArrowError> {
+    let has_continuation = buf.len() >= 4 && buf[..4] == IPC_CONTINUATION;
+    if has_continuation {
+        if buf.len() < 8 {
+            return Err(ArrowError::ParseError(
+                "IPC message buffer too short".into(),
+            ));
+        }
+        let meta_size = u32::from_le_bytes(buf[4..8].try_into().unwrap()) as usize;
+        Ok((8, meta_size))
+    } else {
+        if buf.len() < 4 {
+            return Err(ArrowError::ParseError(
+                "IPC message buffer too short".into(),
+            ));
+        }
+        let meta_size = u32::from_le_bytes(buf[..4].try_into().unwrap()) as usize;
+        Ok((4, meta_size))
+    }
+}
+
+/// Read all [`RecordBatch`]es from one Arrow IPC stream.
+///
+/// Zero-copy: each batch's array data buffers are borrowed from the input
+/// message buffer(s) and not copied during decoding.
+///
+/// Uses [`FileDecoder`] directly (rather than `StreamDecoder`) to avoid a
+/// known edge case where `StreamDecoder` does not produce a batch for messages
+/// with a zero-length body when the message exactly fills the decode buffer.
+pub fn read_ipc_stream(data: &Bytes) -> Result<Vec<RecordBatch>, ArrowError> {
+    let mut offset = 0usize;
+
+    let schema_buf = read_one_ipc_message(&data.slice(offset..))?.ok_or_else(|| {
+        ArrowError::ParseError("IPC stream: expected schema message, got EOS".into())
+    })?;
+    offset += schema_buf.len();
+
+    let (prefix_len, meta_size) = parse_ipc_message_prefix(&schema_buf)?;
+    let schema_msg = root_as_message(&schema_buf[prefix_len..prefix_len + meta_size])
+        .map_err(|e| ArrowError::ParseError(format!("IPC schema parse error: {e}")))?;
+    let schema = Arc::new(fb_to_schema(schema_msg.header_as_schema().ok_or_else(
+        || ArrowError::ParseError("IPC stream: first message is not a schema".into()),
+    )?));
+    let mut decoder = FileDecoder::new(schema, schema_msg.version());
+
+    let mut batches = Vec::new();
+
+    loop {
+        let Some(buf) = read_one_ipc_message(&data.slice(offset..))? else {
+            break;
+        };
+        offset += buf.len();
+
+        let (prefix_len, meta_size) = parse_ipc_message_prefix(&buf)?;
+        let msg = root_as_message(&buf[prefix_len..prefix_len + meta_size])
+            .map_err(|e| ArrowError::ParseError(format!("IPC message parse error: {e}")))?;
+        let body_len = msg.bodyLength() as usize;
+
+        // Block offset = 0 since the buffer starts at the message boundary.
+        // metaDataLength = prefix_len + meta_size (prefix + flatbuf + padding).
+        let block = arrow_ipc::Block::new(0, (prefix_len + meta_size) as i32, body_len as i64);
+
+        match msg.header_type() {
+            arrow_ipc::MessageHeader::RecordBatch => {
+                if let Some(batch) = decoder.read_record_batch(&block, &buf)? {
+                    batches.push(batch);
+                }
+            }
+            arrow_ipc::MessageHeader::DictionaryBatch => {
+                decoder.read_dictionary(&block, &buf)?;
+            }
+            _ => break,
+        }
+    }
+
+    Ok(batches)
+}
+
+/// Read exactly one [`RecordBatch`] from one Arrow IPC stream.
+pub fn read_ipc_stream_single(data: &Bytes) -> Result<RecordBatch, ArrowError> {
+    let mut batches = read_ipc_stream(data)?;
+    match batches.len() {
+        1 => Ok(batches.remove(0)),
+        n => Err(ArrowError::ParseError(format!(
+            "expected exactly 1 IPC record batch, got {n}"
+        ))),
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Aligned IPC sections
+// ---------------------------------------------------------------------------
+
+/// Byte alignment that each IPC section's stream start is padded to.
+///
+/// When several IPC streams are concatenated into one larger blob (e.g. a
+/// cache entry), a section that starts at an arbitrary offset would leave its
+/// array data misaligned. [`FileDecoder`] with `require_alignment = false`
+/// then silently copies each buffer into a freshly aligned allocation on
+/// every read, defeating zero-copy. Padding each section start to a 64-byte
+/// boundary keeps the decoded buffers borrowed directly from the input.
+pub const IPC_SECTION_ALIGNMENT: usize = 64;
+
+/// Number of zero-padding bytes needed to advance `pos` to the next
+/// [`IPC_SECTION_ALIGNMENT`] boundary.
+fn section_padding(pos: usize) -> usize {
+    (IPC_SECTION_ALIGNMENT - (pos % IPC_SECTION_ALIGNMENT)) % IPC_SECTION_ALIGNMENT
+}
+
+/// A [`Write`] adapter that counts the bytes written through it.
+struct CountingWriter<'a> {
+    inner: &'a mut dyn Write,
+    count: usize,
+}
+
+impl Write for CountingWriter<'_> {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        let n = self.inner.write(buf)?;
+        self.count += n;
+        Ok(n)
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        self.inner.flush()
+    }
+}
+
+/// Write zero padding so the next byte lands on an [`IPC_SECTION_ALIGNMENT`]
+/// boundary, advancing `pos` past it.
+fn write_section_padding(writer: &mut dyn Write, pos: &mut usize) -> Result<(), ArrowError> {
+    let pad = section_padding(*pos);
+    if pad > 0 {
+        const ZEROS: [u8; IPC_SECTION_ALIGNMENT] = [0u8; IPC_SECTION_ALIGNMENT];
+        writer
+            .write_all(&ZEROS[..pad])
+            .map_err(|e| ArrowError::IoError(e.to_string(), e))?;
+        *pos += pad;
+    }
+    Ok(())
+}
+
+/// Write `batch` as a 64-byte-aligned single-batch Arrow IPC section.
+///
+/// `pos` is the absolute byte offset of `writer` within the enclosing blob.
+/// Zero padding is written first so the IPC stream begins on an
+/// [`IPC_SECTION_ALIGNMENT`] boundary, then the stream itself. `pos` is
+/// advanced past both the padding and the stream so the caller can write
+/// further aligned sections.
+///
+/// Paired with [`read_ipc_section_at`]. For the decoded buffers to be borrowed
+/// zero-copy, the blob must ultimately be read back from a buffer whose base
+/// address is at least 64-byte aligned.
+pub fn write_ipc_section(
+    writer: &mut dyn Write,
+    pos: &mut usize,
+    batch: &RecordBatch,
+) -> Result<(), ArrowError> {
+    write_section_padding(writer, pos)?;
+
+    let mut counting = CountingWriter {
+        inner: writer,
+        count: 0,
+    };
+    write_ipc_stream(batch, &mut counting)?;
+    *pos += counting.count;
+    Ok(())
+}
+
+/// Read a single [`RecordBatch`] from an aligned IPC section at `offset`.
+///
+/// Skips the alignment padding written by [`write_ipc_section`], then reads
+/// the stream, advancing `offset` past the section (padding + stream + EOS).
+///
+/// Zero-copy: array buffers borrow from `data`'s allocation when `data`'s base
+/// address is at least 64-byte aligned (see [`write_ipc_section`]).
+pub fn read_ipc_section_at(data: &Bytes, offset: &mut usize) -> Result<RecordBatch, ArrowError> {
+    *offset += section_padding(*offset);
+    read_ipc_stream_single_at(data, offset)
+}
+
+/// Write `batches` as a single 64-byte-aligned multi-batch Arrow IPC section.
+///
+/// Like [`write_ipc_section`] but emits every batch from `iter` into one IPC
+/// stream (schema + N batches + EOS). `iter` must yield at least one batch.
+/// Paired with [`read_ipc_section_batches_at`].
+pub fn write_ipc_section_batches<I>(
+    writer: &mut dyn Write,
+    pos: &mut usize,
+    iter: I,
+) -> Result<(), ArrowError>
+where
+    I: IntoIterator<Item = RecordBatch>,
+{
+    write_section_padding(writer, pos)?;
+
+    let mut counting = CountingWriter {
+        inner: writer,
+        count: 0,
+    };
+    write_ipc_stream_batches(iter, &mut counting)?;
+    *pos += counting.count;
+    Ok(())
+}
+
+/// Read all [`RecordBatch`]es from an aligned multi-batch IPC section at
+/// `offset`, advancing `offset` past the section (padding + stream + EOS).
+///
+/// Zero-copy: array buffers borrow from `data`'s allocation when `data`'s base
+/// address is at least 64-byte aligned (see [`write_ipc_section_batches`]).
+pub fn read_ipc_section_batches_at(
+    data: &Bytes,
+    offset: &mut usize,
+) -> Result<Vec<RecordBatch>, ArrowError> {
+    *offset += section_padding(*offset);
+    read_ipc_stream_at(data, offset)
+}
+
+#[cfg(test)]
+mod tests {
+    use arrow_array::{ArrayRef, record_batch};
+
+    use super::*;
+
+    #[test]
+    fn test_ipc_roundtrip() {
+        let batch1 = record_batch!(
+            ("int", Int32, [1, 2, 3]),
+            ("str", Utf8, ["foo", "bar", "baz"])
+        )
+        .unwrap();
+        let batch2 = record_batch!(("int", Int32, [4, 5]), ("str", Utf8, ["qux", "quux"])).unwrap();
+        let batches = vec![batch1.clone(), batch2.clone()];
+
+        let mut buf = Vec::new();
+        write_ipc_stream_batches(batches, &mut buf).unwrap();
+
+        let data = Bytes::from(buf);
+
+        let batches = read_ipc_stream(&data).unwrap();
+        assert_eq!(batches.len(), 2);
+        assert_eq!(batches[0], batch1);
+        assert_eq!(batches[1], batch2);
+
+        let data_base = data.as_ptr() as usize;
+        let data_end = data_base + data.len();
+        let assert_col_zero_copy = |array: &ArrayRef| {
+            for buffer in array.to_data().buffers() {
+                let ptr = buffer.as_ptr() as usize;
+                assert!(
+                    ptr >= data_base && ptr < data_end,
+                    "buffer at {ptr:#x} is not backed by the input Bytes allocation \
+                     [{data_base:#x}..{data_end:#x})"
+                );
+            }
+        };
+
+        for batch in &batches {
+            assert_eq!(batch.schema(), batch1.schema());
+            assert_col_zero_copy(batch.column(0));
+            assert_col_zero_copy(batch.column(1));
+        }
+    }
+
+    /// Allocate a [`Bytes`] whose base address is 64-byte aligned, modelling a
+    /// backend that reads cache entries into an aligned buffer. A plain
+    /// `Bytes::from(vec)` only guarantees the allocator's alignment for `u8`.
+    fn aligned_bytes(payload: &[u8]) -> Bytes {
+        let mut v = vec![0u8; payload.len() + IPC_SECTION_ALIGNMENT];
+        let pad = section_padding(v.as_ptr() as usize);
+        v[pad..pad + payload.len()].copy_from_slice(payload);
+        Bytes::from(v).slice(pad..pad + payload.len())
+    }
+
+    #[test]
+    fn test_aligned_ipc_sections_are_zero_copy() {
+        // A LargeBinary column exercises the i64-offset buffer whose 8-byte
+        // alignment requirement triggers a realigning memcpy when misaligned.
+        let blocks = arrow_array::LargeBinaryArray::from_vec(vec![&b"hello"[..], b"world"]);
+        let section_a = RecordBatch::try_from_iter([("a", Arc::new(blocks) as ArrayRef)]).unwrap();
+        let section_b = record_batch!(("b", Int64, [10i64, 20, 30, 40, 50])).unwrap();
+
+        let mut buf = Vec::new();
+        // Arbitrary, deliberately non-64-aligned preamble so the first section
+        // must be padded rather than landing at offset 0 by luck.
+        buf.extend_from_slice(&[0xABu8; 7]);
+        let mut pos = buf.len();
+        // The first section's stream begins after padding the 7-byte preamble
+        // up to the next 64-byte boundary.
+        assert_eq!(7 + section_padding(7), IPC_SECTION_ALIGNMENT);
+        write_ipc_section(&mut buf, &mut pos, &section_a).unwrap();
+        write_ipc_section(&mut buf, &mut pos, &section_b).unwrap();
+
+        let data = aligned_bytes(&buf);
+        assert_eq!(
+            section_padding(data.as_ptr() as usize),
+            0,
+            "base not aligned"
+        );
+
+        let mut offset = 7;
+        let read_a = read_ipc_section_at(&data, &mut offset).unwrap();
+        let read_b = read_ipc_section_at(&data, &mut offset).unwrap();
+        assert_eq!(read_a, section_a);
+        assert_eq!(read_b, section_b);
+
+        let data_base = data.as_ptr() as usize;
+        let data_end = data_base + data.len();
+        for batch in [&read_a, &read_b] {
+            for buffer in batch.column(0).to_data().buffers() {
+                let ptr = buffer.as_ptr() as usize;
+                assert!(
+                    ptr >= data_base && ptr < data_end,
+                    "section buffer at {ptr:#x} was realigned out of the input \
+                     [{data_base:#x}..{data_end:#x}) — misaligned section",
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn test_aligned_multi_batch_section_roundtrip_zero_copy() {
+        // A multi-batch section (e.g. IVF SQ storage chunks) must round-trip
+        // every batch and decode the first batch's buffers zero-copy.
+        let b1 = record_batch!(("v", Int64, [1i64, 2, 3])).unwrap();
+        let b2 = record_batch!(("v", Int64, [4i64, 5])).unwrap();
+        let b3 = record_batch!(("v", Int64, [6i64])).unwrap();
+
+        let mut buf = vec![0xCDu8; 5];
+        let mut pos = buf.len();
+        write_ipc_section_batches(&mut buf, &mut pos, [b1.clone(), b2.clone(), b3.clone()])
+            .unwrap();
+
+        let data = aligned_bytes(&buf);
+        let mut offset = 5;
+        let read = read_ipc_section_batches_at(&data, &mut offset).unwrap();
+        assert_eq!(read, vec![b1, b2, b3]);
+        assert_eq!(offset, buf.len(), "offset should land at section end");
+
+        let data_base = data.as_ptr() as usize;
+        let data_end = data_base + data.len();
+        for buffer in read[0].column(0).to_data().buffers() {
+            let ptr = buffer.as_ptr() as usize;
+            assert!(
+                ptr >= data_base && ptr < data_end,
+                "first batch buffer at {ptr:#x} was realigned out of the input",
+            );
+        }
+    }
+}

python/vendor/lance-arrow/src/list.rs

@@ -0,0 +1,176 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+
+use std::sync::Arc;
+
+use arrow_array::{Array, BooleanArray, GenericListArray, OffsetSizeTrait};
+use arrow_buffer::{BooleanBufferBuilder, OffsetBuffer, ScalarBuffer};
+use arrow_schema::Field;
+
+pub trait ListArrayExt {
+    /// Filters out masked null items from the list array
+    ///
+    /// It is legal for a list array to have a null entry with a non-zero length.  The
+    /// values inside the entry are "garbage" and should be ignored.  This function
+    /// filters the values array to remove the garbage values.
+    ///
+    /// The output list will always have zero-length nulls.
+    fn filter_garbage_nulls(&self) -> Self;
+    /// Returns a copy of the list's values array that has been sliced to size
+    ///
+    /// It is legal for a list array's offsets to not start with zero.  It's also legal
+    /// for a list array's offsets to not extend to the entire values array.  This function
+    /// behaves similarly to `values()` except it slices the array so that it starts at
+    /// the first list offset and ends at the last list offset.
+    fn trimmed_values(&self) -> Arc<dyn Array>;
+    /// The offset type of the underlying list array.
+    type Offset: OffsetSizeTrait;
+    /// Returns offsets shifted so the first offset is zero, matching
+    /// [`Self::trimmed_values`].
+    ///
+    /// Sliced list arrays (e.g. a filtered batch) keep offsets that reference the
+    /// original values buffer, so combining them with trimmed values produces
+    /// offsets that exceed the values length. Use this together with
+    /// `trimmed_values` when constructing a new list array.
+    fn trimmed_offsets(&self) -> OffsetBuffer<Self::Offset>;
+}
+
+impl<OffsetSize: OffsetSizeTrait> ListArrayExt for GenericListArray<OffsetSize> {
+    fn filter_garbage_nulls(&self) -> Self {
+        if self.is_empty() {
+            return self.clone();
+        }
+        let Some(validity) = self.nulls().cloned() else {
+            return self.clone();
+        };
+
+        let mut should_keep = BooleanBufferBuilder::new(self.values().len());
+
+        // Handle case where offsets do not start at 0
+        let preamble_len = self.offsets().first().unwrap().to_usize().unwrap();
+        should_keep.append_n(preamble_len, false);
+
+        let mut new_offsets: Vec<OffsetSize> = Vec::with_capacity(self.len() + 1);
+        new_offsets.push(OffsetSize::zero());
+        let mut cur_len = OffsetSize::zero();
+        for (offset, is_valid) in self.offsets().windows(2).zip(validity.iter()) {
+            let len = offset[1] - offset[0];
+            if is_valid {
+                cur_len += len;
+                should_keep.append_n(len.to_usize().unwrap(), true);
+                new_offsets.push(cur_len);
+            } else {
+                should_keep.append_n(len.to_usize().unwrap(), false);
+                new_offsets.push(cur_len);
+            }
+        }
+
+        // Offsets may not reference entire values buffer
+        let trailer = self.values().len() - should_keep.len();
+        should_keep.append_n(trailer, false);
+
+        let should_keep = should_keep.finish();
+        let should_keep = BooleanArray::new(should_keep, None);
+        let new_values = arrow_select::filter::filter(self.values(), &should_keep).unwrap();
+        let new_offsets = ScalarBuffer::from(new_offsets);
+        let new_offsets = OffsetBuffer::new(new_offsets);
+
+        Self::new(
+            Arc::new(Field::new(
+                "item",
+                self.value_type(),
+                self.values().is_nullable(),
+            )),
+            new_offsets,
+            new_values,
+            Some(validity),
+        )
+    }
+
+    fn trimmed_values(&self) -> Arc<dyn Array> {
+        let first_value = self
+            .offsets()
+            .first()
+            .map(|v| v.to_usize().unwrap())
+            .unwrap_or(0);
+        let last_value = self
+            .offsets()
+            .last()
+            .map(|v| v.to_usize().unwrap())
+            .unwrap_or(0);
+        self.values().slice(first_value, last_value - first_value)
+    }
+
+    type Offset = OffsetSize;
+
+    fn trimmed_offsets(&self) -> OffsetBuffer<OffsetSize> {
+        let offsets = self.offsets();
+        let Some(&first) = offsets.first() else {
+            return offsets.clone();
+        };
+        if first == OffsetSize::zero() {
+            return offsets.clone();
+        }
+        let shifted: Vec<OffsetSize> = offsets.iter().map(|&o| o - first).collect();
+        OffsetBuffer::new(ScalarBuffer::from(shifted))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use arrow_array::{ListArray, UInt64Array};
+    use arrow_buffer::{BooleanBuffer, NullBuffer, OffsetBuffer, ScalarBuffer};
+    use arrow_schema::{DataType, Field};
+
+    use super::ListArrayExt;
+
+    #[test]
+    fn test_filter_garbage_nulls() {
+        let items = UInt64Array::from(vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
+        let offsets = ScalarBuffer::<i32>::from(vec![2, 5, 8, 9]);
+        let offsets = OffsetBuffer::new(offsets);
+        let list_validity = NullBuffer::new(BooleanBuffer::from(vec![true, false, true]));
+        let list_arr = ListArray::new(
+            Arc::new(Field::new("item", DataType::UInt64, true)),
+            offsets,
+            Arc::new(items),
+            Some(list_validity.clone()),
+        );
+
+        let filtered = list_arr.filter_garbage_nulls();
+
+        let expected_items = UInt64Array::from(vec![2, 3, 4, 8]);
+        let offsets = ScalarBuffer::<i32>::from(vec![0, 3, 3, 4]);
+        let expected = ListArray::new(
+            Arc::new(Field::new("item", DataType::UInt64, false)),
+            OffsetBuffer::new(offsets),
+            Arc::new(expected_items),
+            Some(list_validity),
+        );
+
+        assert_eq!(filtered, expected);
+    }
+
+    #[test]
+    fn test_trim_values() {
+        let items = UInt64Array::from(vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
+        let offsets = ScalarBuffer::<i32>::from(vec![2, 5, 6, 8, 9]);
+        let offsets = OffsetBuffer::new(offsets);
+        let list_arr = ListArray::new(
+            Arc::new(Field::new("item", DataType::UInt64, true)),
+            offsets,
+            Arc::new(items),
+            None,
+        );
+        let list_arr = list_arr.slice(1, 2);
+
+        let trimmed = list_arr.trimmed_values();
+
+        let expected_items = UInt64Array::from(vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
+        let expected_items = expected_items.slice(5, 3);
+
+        assert_eq!(trimmed.as_ref(), &expected_items);
+    }
+}

python/vendor/lance-arrow/src/memory.rs

@@ -0,0 +1,91 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+
+use std::collections::HashSet;
+
+use arrow_array::{Array, RecordBatch};
+use arrow_data::ArrayData;
+
+/// Counts memory used by buffers of Arrow arrays and RecordBatches.
+///
+/// This is meant to capture how much memory is being used by the Arrow data
+/// structures as they are. It does not represent the memory used if the data
+/// were to be serialized and then deserialized. In particular:
+///
+/// * This does not double count memory used by buffers shared by multiple
+///   arrays or batches. Round-tripped data may use more memory because of this.
+/// * This counts the **total** size of the buffers, even if the array is a slice.
+///   Round-tripped data may use less memory because of this.
+#[derive(Default)]
+pub struct MemoryAccumulator {
+    seen: HashSet<usize>,
+    total: usize,
+}
+
+impl MemoryAccumulator {
+    pub fn record_array(&mut self, array: &dyn Array) {
+        let data = array.to_data();
+        self.record_array_data(&data);
+    }
+
+    fn record_array_data(&mut self, data: &ArrayData) {
+        for buffer in data.buffers() {
+            let ptr = buffer.as_ptr();
+            if self.seen.insert(ptr as usize) {
+                self.total += buffer.capacity();
+            }
+        }
+
+        if let Some(nulls) = data.nulls() {
+            let null_buf = nulls.inner().inner();
+            let ptr = null_buf.as_ptr();
+            if self.seen.insert(ptr as usize) {
+                self.total += null_buf.capacity();
+            }
+        }
+
+        for child in data.child_data() {
+            self.record_array_data(child);
+        }
+    }
+
+    pub fn record_batch(&mut self, batch: &RecordBatch) {
+        for array in batch.columns() {
+            self.record_array(array);
+        }
+    }
+
+    pub fn total(&self) -> usize {
+        self.total
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use arrow_array::Int32Array;
+    use arrow_schema::{DataType, Field, Schema};
+
+    use super::*;
+
+    #[test]
+    fn test_memory_accumulator() {
+        let batch = RecordBatch::try_new(
+            Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)])),
+            vec![Arc::new(Int32Array::from(vec![1, 2, 3]))],
+        )
+        .unwrap();
+        let slice = batch.slice(1, 2);
+
+        let mut acc = MemoryAccumulator::default();
+
+        // Should record whole buffer, not just slice
+        acc.record_batch(&slice);
+        assert_eq!(acc.total(), 3 * std::mem::size_of::<i32>());
+
+        // Should not double count
+        acc.record_batch(&slice);
+        assert_eq!(acc.total(), 3 * std::mem::size_of::<i32>());
+    }
+}

python/vendor/lance-arrow/src/scalar.rs

@@ -0,0 +1,301 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+
+use arrow_array::{ArrayRef, UInt64Array, make_array};
+use arrow_buffer::Buffer;
+use arrow_data::ArrayDataBuilder;
+use arrow_schema::{ArrowError, DataType};
+use arrow_select::take::take;
+
+use crate::DataTypeExt;
+
+type Result<T> = std::result::Result<T, ArrowError>;
+
+pub const INLINE_VALUE_MAX_BYTES: usize = 32;
+
+pub fn extract_scalar_value(array: &ArrayRef, idx: usize) -> Result<ArrayRef> {
+    if idx >= array.len() {
+        return Err(ArrowError::InvalidArgumentError(
+            "Scalar index out of bounds".to_string(),
+        ));
+    }
+
+    take(array.as_ref(), &UInt64Array::from(vec![idx as u64]), None)
+}
+
+fn read_u32(buf: &[u8], offset: &mut usize) -> Result<u32> {
+    if *offset + 4 > buf.len() {
+        return Err(ArrowError::InvalidArgumentError(
+            "Invalid scalar value buffer: unexpected EOF".to_string(),
+        ));
+    }
+    let bytes = [
+        buf[*offset],
+        buf[*offset + 1],
+        buf[*offset + 2],
+        buf[*offset + 3],
+    ];
+    *offset += 4;
+    Ok(u32::from_le_bytes(bytes))
+}
+
+fn read_bytes<'a>(buf: &'a [u8], offset: &mut usize, len: usize) -> Result<&'a [u8]> {
+    if *offset + len > buf.len() {
+        return Err(ArrowError::InvalidArgumentError(
+            "Invalid scalar value buffer: unexpected EOF".to_string(),
+        ));
+    }
+    let slice = &buf[*offset..*offset + len];
+    *offset += len;
+    Ok(slice)
+}
+
+fn write_u32(out: &mut Vec<u8>, v: u32) {
+    out.extend_from_slice(&v.to_le_bytes());
+}
+
+fn write_bytes(out: &mut Vec<u8>, bytes: &[u8]) {
+    out.extend_from_slice(bytes);
+}
+
+pub fn encode_scalar_value_buffer(scalar: &ArrayRef) -> Result<Vec<u8>> {
+    if scalar.len() != 1 || scalar.null_count() != 0 {
+        return Err(ArrowError::InvalidArgumentError(
+            "Scalar value buffer must be a single non-null value".to_string(),
+        ));
+    }
+    let data = scalar.to_data();
+    if data.offset() != 0 {
+        return Err(ArrowError::InvalidArgumentError(
+            "Scalar value buffer must have offset=0".to_string(),
+        ));
+    }
+    if !data.child_data().is_empty() {
+        return Err(ArrowError::InvalidArgumentError(
+            "Scalar value buffer does not support nested types".to_string(),
+        ));
+    }
+
+    // Minimal format (RFC): store the Arrow value buffers for a length-1 array.
+    // Null bitmap and child data are intentionally not supported here.
+    //
+    // | u32 num_buffers |
+    // | u32 buffer_0_len | ... | u32 buffer_{n-1}_len |
+    // | buffer_0 bytes | ... | buffer_{n-1} bytes |
+    let mut out = Vec::with_capacity(128);
+    let buffers = data.buffers();
+    write_u32(&mut out, buffers.len() as u32);
+    for b in buffers {
+        write_u32(&mut out, b.len() as u32);
+    }
+    for b in buffers {
+        write_bytes(&mut out, b.as_slice());
+    }
+    Ok(out)
+}
+
+pub fn decode_scalar_from_value_buffer(
+    data_type: &DataType,
+    value_buffer: &[u8],
+) -> Result<ArrayRef> {
+    if matches!(
+        data_type,
+        DataType::Struct(_) | DataType::FixedSizeList(_, _)
+    ) {
+        return Err(ArrowError::InvalidArgumentError(format!(
+            "Scalar value buffer does not support nested data type {:?}",
+            data_type
+        )));
+    }
+
+    let mut offset = 0;
+    let num_buffers = read_u32(value_buffer, &mut offset)? as usize;
+    let buffer_lens = (0..num_buffers)
+        .map(|_| read_u32(value_buffer, &mut offset).map(|l| l as usize))
+        .collect::<Result<Vec<_>>>()?;
+
+    let mut buffers = Vec::with_capacity(num_buffers);
+    for len in buffer_lens {
+        let bytes = read_bytes(value_buffer, &mut offset, len)?;
+        buffers.push(Buffer::from_vec(bytes.to_vec()));
+    }
+
+    if offset != value_buffer.len() {
+        return Err(ArrowError::InvalidArgumentError(
+            "Invalid scalar value buffer: trailing bytes".to_string(),
+        ));
+    }
+
+    let mut builder = ArrayDataBuilder::new(data_type.clone())
+        .len(1)
+        .null_count(0);
+    for b in buffers {
+        builder = builder.add_buffer(b);
+    }
+    Ok(make_array(builder.build()?))
+}
+
+pub fn decode_scalar_from_inline_value(
+    data_type: &DataType,
+    inline_value: &[u8],
+) -> Result<ArrayRef> {
+    // I expect our input to be safe here, but I added some debug_assert_eq statements just in case.
+    // If they are triggered, we may need to change them to return actual errors.
+    //
+    // Boolean values are bit-packed in Arrow and therefore are not "fixed-stride" in bytes.
+    // As a result, `byte_width_opt()` returns `None` for `DataType::Boolean`, even though a
+    // length-1 scalar can be represented inline using a single byte (matching `try_inline_value`).
+    if matches!(data_type, DataType::Boolean) {
+        debug_assert_eq!(
+            inline_value.len(),
+            1,
+            "Invalid boolean inline scalar length (expected 1 byte, got {})",
+            inline_value.len()
+        );
+    } else if let Some(byte_width) = data_type.byte_width_opt() {
+        debug_assert_eq!(
+            inline_value.len(),
+            byte_width,
+            "Inline constant length mismatch for {:?}: expected {} bytes but got {}",
+            data_type,
+            byte_width,
+            inline_value.len()
+        );
+    }
+
+    let data = ArrayDataBuilder::new(data_type.clone())
+        .len(1)
+        .null_count(0)
+        .add_buffer(Buffer::from_vec(inline_value.to_vec()))
+        .build()?;
+    Ok(make_array(data))
+}
+
+pub fn try_inline_value(scalar: &ArrayRef) -> Option<Vec<u8>> {
+    if scalar.null_count() != 0 || scalar.len() != 1 {
+        return None;
+    }
+    let data = scalar.to_data();
+    if !data.child_data().is_empty() {
+        return None;
+    }
+    if data.buffers().len() != 1 {
+        return None;
+    }
+    let bytes = data.buffers()[0].as_slice();
+    if bytes.len() > INLINE_VALUE_MAX_BYTES {
+        return None;
+    }
+    Some(bytes.to_vec())
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use arrow_array::{
+        BooleanArray, DictionaryArray, FixedSizeBinaryArray, Int8Array, Int32Array, StringArray,
+        cast::AsArray, types::Int8Type,
+    };
+
+    use super::*;
+
+    #[test]
+    fn test_extract_scalar_value() {
+        let array: ArrayRef = Arc::new(Int32Array::from(vec![Some(1), None, Some(3)]));
+        let scalar = extract_scalar_value(&array, 2).unwrap();
+        assert_eq!(scalar.len(), 1);
+        assert_eq!(
+            scalar
+                .as_primitive::<arrow_array::types::Int32Type>()
+                .value(0),
+            3
+        );
+    }
+
+    #[test]
+    fn test_extract_scalar_value_from_full_dictionary() {
+        let values = Arc::new(StringArray::from(
+            (0..=i8::MAX)
+                .map(|value| format!("value-{value}"))
+                .collect::<Vec<_>>(),
+        ));
+        let keys = Int8Array::from((0..=i8::MAX).collect::<Vec<_>>());
+        let array: ArrayRef = Arc::new(DictionaryArray::<Int8Type>::new(keys, values));
+
+        let scalar = extract_scalar_value(&array, i8::MAX as usize).unwrap();
+
+        let scalar = scalar.as_dictionary::<Int8Type>();
+        assert_eq!(scalar.len(), 1);
+        assert_eq!(scalar.key(0), Some(i8::MAX as usize));
+        assert_eq!(scalar.values().len(), i8::MAX as usize + 1);
+    }
+
+    #[test]
+    fn test_scalar_value_buffer_utf8_round_trip() {
+        let scalar: ArrayRef = Arc::new(StringArray::from(vec!["hello"]));
+        let buf = encode_scalar_value_buffer(&scalar).unwrap();
+        let decoded = decode_scalar_from_value_buffer(&DataType::Utf8, &buf).unwrap();
+        assert_eq!(decoded.len(), 1);
+        assert_eq!(decoded.null_count(), 0);
+        assert_eq!(decoded.as_string::<i32>().value(0), "hello");
+    }
+
+    #[test]
+    fn test_scalar_value_buffer_fixed_size_binary_round_trip() {
+        let val = vec![0xABu8; 33];
+        let scalar: ArrayRef = Arc::new(
+            FixedSizeBinaryArray::try_from_sparse_iter_with_size(
+                std::iter::once(Some(val.as_slice())),
+                33,
+            )
+            .unwrap(),
+        );
+        let buf = encode_scalar_value_buffer(&scalar).unwrap();
+        let decoded =
+            decode_scalar_from_value_buffer(&DataType::FixedSizeBinary(33), &buf).unwrap();
+        assert_eq!(decoded.len(), 1);
+        assert_eq!(decoded.as_fixed_size_binary().value(0), val.as_slice());
+    }
+
+    #[test]
+    fn test_inline_value_boolean_round_trip() {
+        let scalar: ArrayRef = Arc::new(BooleanArray::from_iter([Some(true)]));
+        let inline = try_inline_value(&scalar).unwrap();
+        let decoded = decode_scalar_from_inline_value(&DataType::Boolean, &inline).unwrap();
+        assert_eq!(decoded.len(), 1);
+        assert_eq!(decoded.null_count(), 0);
+        assert!(decoded.as_boolean().value(0));
+    }
+
+    #[test]
+    fn test_scalar_value_buffer_rejects_nested_type() {
+        let field = Arc::new(arrow_schema::Field::new("item", DataType::Int32, false));
+        let list: ArrayRef = Arc::new(arrow_array::FixedSizeListArray::new(
+            field,
+            2,
+            Arc::new(Int32Array::from(vec![1, 2])),
+            None,
+        ));
+        let scalar = list.slice(0, 1);
+        assert!(encode_scalar_value_buffer(&scalar).is_err());
+    }
+
+    #[test]
+    fn test_decode_scalar_from_value_buffer_rejects_nested_type() {
+        let buf = Vec::<u8>::new();
+        let res =
+            decode_scalar_from_value_buffer(&DataType::Struct(arrow_schema::Fields::empty()), &buf);
+        assert!(res.is_err());
+    }
+
+    #[test]
+    fn test_decode_scalar_from_value_buffer_trailing_bytes() {
+        // num_buffers = 0, plus an extra byte
+        let mut bytes = Vec::new();
+        bytes.extend_from_slice(&0u32.to_le_bytes());
+        bytes.push(1);
+        let res = decode_scalar_from_value_buffer(&DataType::Int32, &bytes);
+        assert!(res.is_err());
+    }
+}

python/vendor/lance-arrow/src/schema.rs

@@ -0,0 +1,207 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+
+//! Extension to arrow schema
+
+use arrow_schema::{ArrowError, DataType, Field, FieldRef, Schema};
+
+use crate::{ARROW_EXT_NAME_KEY, BLOB_META_KEY, BLOB_V2_EXT_NAME};
+
+pub enum Indentation {
+    OneLine,
+    MultiLine(u8),
+}
+
+impl Indentation {
+    fn value(&self) -> String {
+        match self {
+            Self::OneLine => "".to_string(),
+            Self::MultiLine(spaces) => " ".repeat(*spaces as usize),
+        }
+    }
+
+    fn deepen(&self) -> Self {
+        match self {
+            Self::OneLine => Self::OneLine,
+            Self::MultiLine(spaces) => Self::MultiLine(spaces + 2),
+        }
+    }
+}
+
+/// Extends the functionality of [arrow_schema::Field].
+pub trait FieldExt {
+    /// Create a compact string representation of the field
+    ///
+    /// This is intended for display purposes and not for serialization
+    fn to_compact_string(&self, indent: Indentation) -> String;
+
+    /// Check if the field is marked as a packed struct
+    fn is_packed_struct(&self) -> bool;
+
+    /// Check if the field is marked as a blob
+    fn is_blob(&self) -> bool;
+
+    /// Check if the field is marked as a blob
+    fn is_blob_v2(&self) -> bool;
+}
+
+impl FieldExt for Field {
+    fn to_compact_string(&self, indent: Indentation) -> String {
+        let mut result = format!("{}: ", self.name().clone());
+        match self.data_type() {
+            DataType::Struct(fields) => {
+                result += "{";
+                result += &indent.value();
+                for (field_idx, field) in fields.iter().enumerate() {
+                    result += field.to_compact_string(indent.deepen()).as_str();
+                    if field_idx < fields.len() - 1 {
+                        result += ",";
+                    }
+                    result += indent.value().as_str();
+                }
+                result += "}";
+            }
+            DataType::List(field)
+            | DataType::LargeList(field)
+            | DataType::ListView(field)
+            | DataType::LargeListView(field) => {
+                result += "[";
+                result += field.to_compact_string(indent.deepen()).as_str();
+                result += "]";
+            }
+            DataType::FixedSizeList(child, dimension) => {
+                result += &format!(
+                    "[{}; {}]",
+                    child.to_compact_string(indent.deepen()),
+                    dimension
+                );
+            }
+            DataType::Dictionary(key_type, value_type) => {
+                result += &value_type.to_string();
+                result += "@";
+                result += &key_type.to_string();
+            }
+            _ => {
+                result += &self.data_type().to_string();
+            }
+        }
+        if self.is_nullable() {
+            result += "?";
+        }
+        result
+    }
+
+    // Check if field has metadata `packed` set to true, this check is case insensitive.
+    fn is_packed_struct(&self) -> bool {
+        let field_metadata = self.metadata();
+        const PACKED_KEYS: [&str; 2] = ["packed", "lance-encoding:packed"];
+        PACKED_KEYS.iter().any(|key| {
+            field_metadata
+                .get(*key)
+                .map(|value| value.eq_ignore_ascii_case("true"))
+                .unwrap_or(false)
+        })
+    }
+
+    fn is_blob(&self) -> bool {
+        let field_metadata = self.metadata();
+        field_metadata.get(BLOB_META_KEY).is_some()
+            || field_metadata
+                .get(ARROW_EXT_NAME_KEY)
+                .map(|value| value == BLOB_V2_EXT_NAME)
+                .unwrap_or(false)
+    }
+
+    fn is_blob_v2(&self) -> bool {
+        let field_metadata = self.metadata();
+        field_metadata
+            .get(ARROW_EXT_NAME_KEY)
+            .map(|value| value == BLOB_V2_EXT_NAME)
+            .unwrap_or(false)
+    }
+}
+
+/// Extends the functionality of [arrow_schema::Schema].
+pub trait SchemaExt {
+    /// Create a new [`Schema`] with one extra field.
+    fn try_with_column(&self, field: Field) -> std::result::Result<Schema, ArrowError>;
+
+    fn try_with_column_at(
+        &self,
+        index: usize,
+        field: Field,
+    ) -> std::result::Result<Schema, ArrowError>;
+
+    fn field_names(&self) -> Vec<&String>;
+
+    fn without_column(&self, column_name: &str) -> Schema;
+
+    /// Create a compact string representation of the schema
+    ///
+    /// This is intended for display purposes and not for serialization
+    fn to_compact_string(&self, indent: Indentation) -> String;
+}
+
+impl SchemaExt for Schema {
+    fn try_with_column(&self, field: Field) -> std::result::Result<Schema, ArrowError> {
+        if self.column_with_name(field.name()).is_some() {
+            return Err(ArrowError::SchemaError(format!(
+                "Can not append column {} on schema: {:?}",
+                field.name(),
+                self
+            )));
+        };
+        let mut fields: Vec<FieldRef> = self.fields().iter().cloned().collect();
+        fields.push(FieldRef::new(field));
+        Ok(Self::new_with_metadata(fields, self.metadata.clone()))
+    }
+
+    fn try_with_column_at(
+        &self,
+        index: usize,
+        field: Field,
+    ) -> std::result::Result<Schema, ArrowError> {
+        if self.column_with_name(field.name()).is_some() {
+            return Err(ArrowError::SchemaError(format!(
+                "Failed to modify schema: Inserting column {} would create a duplicate column in schema: {:?}",
+                field.name(),
+                self
+            )));
+        };
+        let mut fields: Vec<FieldRef> = self.fields().iter().cloned().collect();
+        fields.insert(index, FieldRef::new(field));
+        Ok(Self::new_with_metadata(fields, self.metadata.clone()))
+    }
+
+    /// Project the schema to remove the given column.
+    ///
+    /// This only works on top-level fields right now. If a field does not exist,
+    /// the schema will be returned as is.
+    fn without_column(&self, column_name: &str) -> Schema {
+        let fields: Vec<FieldRef> = self
+            .fields()
+            .iter()
+            .filter(|f| f.name() != column_name)
+            .cloned()
+            .collect();
+        Self::new_with_metadata(fields, self.metadata.clone())
+    }
+
+    fn field_names(&self) -> Vec<&String> {
+        self.fields().iter().map(|f| f.name()).collect()
+    }
+
+    fn to_compact_string(&self, indent: Indentation) -> String {
+        let mut result = "{".to_string();
+        result += &indent.value();
+        for (field_idx, field) in self.fields.iter().enumerate() {
+            result += field.to_compact_string(indent.deepen()).as_str();
+            if field_idx < self.fields.len() - 1 {
+                result += ",";
+            }
+            result += indent.value().as_str();
+        }
+        result += "}";
+        result
+    }
+}

python/vendor/lance-arrow/src/stream.rs

@@ -0,0 +1,553 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+
+//! Utilities for working with streams of [`RecordBatch`].
+
+use arrow_array::RecordBatch;
+use arrow_schema::{ArrowError, SchemaRef};
+use futures::stream::{self, Stream, StreamExt};
+use std::pin::Pin;
+
+use crate::deepcopy::deep_copy_batch_sliced;
+
+/// Rechunks a stream of [`RecordBatch`] so that each output batch has
+/// approximately `target_bytes` of array data.
+///
+/// Small input batches are accumulated (by concatenation) until at least
+/// `min_bytes` of data has been collected. If the resulting batch exceeds
+/// `max_bytes`, it is sliced into roughly equal pieces of ~`max_bytes`
+/// (assuming uniform row sizes).
+pub fn rechunk_stream_by_size<S, E>(
+    input: S,
+    input_schema: SchemaRef,
+    min_bytes: usize,
+    max_bytes: usize,
+) -> impl Stream<Item = Result<RecordBatch, E>>
+where
+    S: Stream<Item = Result<RecordBatch, E>>,
+    E: From<ArrowError>,
+{
+    rechunk_stream_by_size_inner(input, input_schema, min_bytes, max_bytes, false)
+}
+
+/// Like [`rechunk_stream_by_size`] but deep-copies slices so that
+/// `get_array_memory_size` reflects the true size of each output batch.
+///
+/// After a normal `RecordBatch::slice`, the backing buffers are shared with
+/// the original batch, so `get_array_memory_size` still reports the full
+/// parent size.  This variant deep-copies every slice produced during the
+/// splitting phase, which allows the stream to detect and re-split slices
+/// that still exceed `max_bytes` (e.g. because a single row is much larger
+/// than average).
+///
+/// The deep copy is a last resort and potentially expensive for large
+/// batches.  However, it is only performed when a batch actually needs to be
+/// sliced — batches that are already within the target range pass through at
+/// zero cost.  Use this only when the hard cap on `max_bytes` is a
+/// correctness requirement, not merely a performance hint.
+pub fn rechunk_stream_by_size_deep_copy<S, E>(
+    input: S,
+    input_schema: SchemaRef,
+    min_bytes: usize,
+    max_bytes: usize,
+) -> impl Stream<Item = Result<RecordBatch, E>>
+where
+    S: Stream<Item = Result<RecordBatch, E>>,
+    E: From<ArrowError>,
+{
+    rechunk_stream_by_size_inner(input, input_schema, min_bytes, max_bytes, true)
+}
+
+fn rechunk_stream_by_size_inner<S, E>(
+    input: S,
+    input_schema: SchemaRef,
+    min_bytes: usize,
+    max_bytes: usize,
+    deep_copy: bool,
+) -> impl Stream<Item = Result<RecordBatch, E>>
+where
+    S: Stream<Item = Result<RecordBatch, E>>,
+    E: From<ArrowError>,
+{
+    stream::try_unfold(
+        RechunkState {
+            input: Box::pin(input),
+            accumulated: Vec::new(),
+            acc_bytes: 0,
+            done: false,
+            input_schema,
+            min_bytes,
+            max_bytes,
+            deep_copy,
+        },
+        |mut state| async move {
+            if state.done && state.accumulated.is_empty() {
+                return Ok(None);
+            }
+
+            // Pull batches until we reach the byte target or exhaust input.
+            // Always pull at least one batch so that min_bytes=0 works.
+            while !state.done && (state.accumulated.is_empty() || state.acc_bytes < state.min_bytes)
+            {
+                match state.input.next().await {
+                    Some(Ok(batch)) => {
+                        state.acc_bytes += batch.get_array_memory_size();
+                        state.accumulated.push(batch);
+                    }
+                    Some(Err(e)) => return Err(e),
+                    None => {
+                        state.done = true;
+                    }
+                }
+            }
+
+            if state.accumulated.is_empty() {
+                return Ok(None);
+            }
+
+            // Fast path: if the first accumulated batch already meets the
+            // byte threshold, deliver it directly instead of concatenating
+            // everything together (which would just get sliced back apart).
+            if state.accumulated.len() > 1
+                && state.accumulated[0].get_array_memory_size() >= state.min_bytes
+            {
+                let b = state.accumulated.remove(0);
+                state.acc_bytes -= b.get_array_memory_size();
+                return Ok(Some((b, state)));
+            }
+
+            let batch = if state.accumulated.len() == 1 {
+                state.accumulated.pop().unwrap()
+            } else {
+                let b =
+                    arrow_select::concat::concat_batches(&state.input_schema, &state.accumulated)
+                        .map_err(E::from)?;
+                state.accumulated.clear();
+                b
+            };
+            state.acc_bytes = 0;
+
+            // Slice the batch into ~max_bytes pieces assuming uniform row sizes.
+            let mut slices =
+                slice_batch(batch, state.max_bytes, state.deep_copy).map_err(E::from)?;
+
+            if slices.len() == 1 {
+                Ok(Some((slices.pop().unwrap(), state)))
+            } else {
+                let first = slices.remove(0);
+
+                // Stash leftover slices for subsequent iterations.
+                for a in &slices {
+                    state.acc_bytes += a.get_array_memory_size();
+                }
+                state.accumulated = slices;
+
+                Ok(Some((first, state)))
+            }
+        },
+    )
+}
+
+/// Slice a batch into pieces of at most `max_bytes`.
+///
+/// When `deep_copy` is false, slices share buffers with the original batch
+/// and `get_array_memory_size` will still report the parent buffer size.
+/// This is fine when the caller only needs approximate sizing.
+///
+/// When `deep_copy` is true, each slice is deep-copied so that
+/// `get_array_memory_size` reflects the true size.  If a deep-copied slice
+/// still exceeds `max_bytes` (due to non-uniform row sizes), it is
+/// recursively split until every piece is within budget or contains only a
+/// single row.
+fn slice_batch(
+    batch: RecordBatch,
+    max_bytes: usize,
+    deep_copy: bool,
+) -> Result<Vec<RecordBatch>, ArrowError> {
+    let batch_bytes = batch.get_array_memory_size();
+    let num_rows = batch.num_rows();
+
+    if batch_bytes <= max_bytes {
+        return Ok(vec![batch]);
+    }
+
+    if num_rows <= 1 {
+        // A single row cannot be split further, but the size just measured may
+        // belong to its source buffer rather than to the row:
+        // `get_array_memory_size` reports whole buffers, and a slice shares
+        // them. Returning it uncopied hands the caller that inflated figure --
+        // a 100 KB row inside a 200 MB buffer measures as 200 MB, and a caller
+        // budgeting on it concludes the row cannot fit anywhere.
+        if deep_copy {
+            return Ok(vec![deep_copy_batch_sliced(&batch)?]);
+        }
+        return Ok(vec![batch]);
+    }
+
+    let rows_per_chunk = (max_bytes as u64 * num_rows as u64 / batch_bytes as u64).max(1) as usize;
+
+    let mut result = Vec::new();
+    let mut offset = 0;
+    while offset < num_rows {
+        let len = rows_per_chunk.min(num_rows - offset);
+        let slice = batch.slice(offset, len);
+        if deep_copy {
+            let copied = deep_copy_batch_sliced(&slice)?;
+            // Recurse: the deep-copied slice has accurate sizes, so if it
+            // still exceeds max_bytes we can split further.
+            result.extend(slice_batch(copied, max_bytes, true)?);
+        } else {
+            result.push(slice);
+        }
+        offset += len;
+    }
+
+    Ok(result)
+}
+
+/// Internal state for [`rechunk_stream`].
+///
+/// Kept as a named struct so the `try_unfold` closure stays readable.
+struct RechunkState<S> {
+    input: Pin<Box<S>>,
+    accumulated: Vec<RecordBatch>,
+    acc_bytes: usize,
+    done: bool,
+    input_schema: SchemaRef,
+    min_bytes: usize,
+    max_bytes: usize,
+    deep_copy: bool,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use std::sync::Arc;
+
+    use arrow_array::Int32Array;
+    use arrow_schema::{DataType, Field, Schema};
+    use futures::executor::block_on;
+
+    /// A single row that shares a large source buffer must report its own
+    /// size, not the buffer's. Callers budget on this figure, and an inflated
+    /// one makes a small row look impossible to place.
+    #[test]
+    fn a_single_row_slice_reports_its_own_size() {
+        use arrow_array::LargeStringArray;
+
+        let row = "x".repeat(100 * 1024);
+        let values: Vec<&str> = (0..2000).map(|_| row.as_str()).collect();
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            "text",
+            DataType::LargeUtf8,
+            false,
+        )]));
+        let batch =
+            RecordBatch::try_new(schema, vec![Arc::new(LargeStringArray::from(values))]).unwrap();
+        let whole = batch.get_array_memory_size();
+
+        let one_row = batch.slice(7, 1);
+        assert_eq!(
+            one_row.get_array_memory_size(),
+            whole,
+            "a slice shares its source buffer, which is the reason this test exists"
+        );
+
+        // Budget far below the row so the split recurses to the single-row floor.
+        let chunks = slice_batch(one_row, 1024, true).unwrap();
+        assert_eq!(chunks.len(), 1);
+        let measured = chunks[0].get_array_memory_size();
+        assert_eq!(chunks[0].num_rows(), 1);
+        assert!(
+            measured < whole / 100,
+            "single row still measured as {measured} bytes against a {whole} byte source"
+        );
+    }
+
+    fn make_batch(num_rows: usize) -> RecordBatch {
+        let schema = test_schema();
+        let values: Vec<i32> = (0..num_rows as i32).collect();
+        RecordBatch::try_new(schema, vec![Arc::new(Int32Array::from(values))]).unwrap()
+    }
+
+    fn test_schema() -> SchemaRef {
+        Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]))
+    }
+
+    fn collect_rechunked(
+        batches: Vec<RecordBatch>,
+        min_bytes: usize,
+        max_bytes: usize,
+    ) -> Vec<RecordBatch> {
+        let input = stream::iter(batches.into_iter().map(Ok::<_, ArrowError>));
+        let rechunked = rechunk_stream_by_size(input, test_schema(), min_bytes, max_bytes);
+        block_on(rechunked.collect::<Vec<_>>())
+            .into_iter()
+            .map(|r| r.unwrap())
+            .collect()
+    }
+
+    fn total_rows(batches: &[RecordBatch]) -> usize {
+        batches.iter().map(|b| b.num_rows()).sum()
+    }
+
+    #[test]
+    fn test_empty_stream() {
+        let result = collect_rechunked(vec![], 100, 200);
+        assert!(result.is_empty());
+    }
+
+    #[test]
+    fn test_single_batch_passthrough() {
+        let batch = make_batch(100);
+        let bytes = batch.get_array_memory_size();
+        // Batch is between min and max — should pass through as-is.
+        let result = collect_rechunked(vec![batch], bytes / 2, bytes * 2);
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[0].num_rows(), 100);
+    }
+
+    #[test]
+    fn test_small_batches_concatenated() {
+        let one_batch_bytes = make_batch(10).get_array_memory_size();
+        let batches: Vec<_> = (0..8).map(|_| make_batch(10)).collect();
+        // min = 5 batches worth, max = 10 batches worth.
+        let result = collect_rechunked(batches, one_batch_bytes * 5, one_batch_bytes * 10);
+        assert_eq!(total_rows(&result), 80);
+        // Should have been concatenated into fewer batches than the 8 inputs.
+        assert!(
+            result.len() < 8,
+            "expected fewer output batches, got {}",
+            result.len()
+        );
+    }
+
+    #[test]
+    fn test_large_batch_sliced() {
+        let batch = make_batch(1000);
+        let bytes = batch.get_array_memory_size();
+        let result = collect_rechunked(vec![batch], bytes / 8, bytes / 4);
+        assert_eq!(total_rows(&result), 1000);
+        assert!(
+            result.len() >= 4,
+            "expected at least 4 slices, got {}",
+            result.len()
+        );
+    }
+
+    #[test]
+    fn test_sliced_leftovers_are_not_recombined() {
+        // Key test for the fast-path optimisation. When a large batch is
+        // sliced, leftover slices should be delivered one-at-a-time without
+        // being concatenated back together.  We verify this by checking that
+        // every output buffer pointer falls inside the original batch's
+        // allocation (i.e. they are all zero-copy slices, not fresh copies).
+        let batch = make_batch(1000);
+        let bytes = batch.get_array_memory_size();
+        let orig_data = batch.column(0).to_data();
+        let orig_buf = &orig_data.buffers()[0];
+        let orig_start = orig_buf.as_ptr() as usize;
+        let orig_end = orig_start + orig_buf.len();
+
+        let result = collect_rechunked(vec![batch], bytes / 8, bytes / 4);
+
+        assert_eq!(total_rows(&result), 1000);
+        assert!(result.len() >= 4);
+
+        for (i, b) in result.iter().enumerate() {
+            let ptr = b.column(0).to_data().buffers()[0].as_ptr() as usize;
+            assert!(
+                ptr >= orig_start && ptr < orig_end,
+                "slice {i} buffer at {ptr:#x} is outside the original allocation \
+                 [{orig_start:#x}, {orig_end:#x}) — it was re-concatenated"
+            );
+        }
+    }
+
+    #[test]
+    fn test_flush_remainder_on_stream_end() {
+        // Data below min_bytes should still be flushed when the stream ends.
+        let batch = make_batch(10);
+        let bytes = batch.get_array_memory_size();
+        let result = collect_rechunked(vec![batch], bytes * 100, bytes * 200);
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[0].num_rows(), 10);
+    }
+
+    #[test]
+    fn test_large_then_small_batches() {
+        // After a large batch is fully drained, subsequent small batches
+        // should be accumulated normally.
+        let large = make_batch(1000);
+        let small_bytes = make_batch(10).get_array_memory_size();
+        let batches = vec![
+            large,
+            make_batch(10),
+            make_batch(10),
+            make_batch(10),
+            make_batch(10),
+            make_batch(10),
+        ];
+        let result = collect_rechunked(batches, small_bytes * 3, small_bytes * 100);
+        assert_eq!(total_rows(&result), 1050);
+        // The large batch should appear (possibly sliced) followed by
+        // concatenated small batches, so we should have fewer output batches
+        // than the 6 inputs.
+        assert!(result.len() < 6);
+    }
+
+    #[test]
+    fn test_row_preservation_across_slicing() {
+        // Verify that every input row appears exactly once in the output
+        // and in the correct order after slicing.
+        let batch = make_batch(237); // odd count to exercise remainder slice
+        let bytes = batch.get_array_memory_size();
+        let result = collect_rechunked(vec![batch], bytes / 8, bytes / 5);
+
+        assert_eq!(total_rows(&result), 237);
+
+        let values: Vec<i32> = result
+            .iter()
+            .flat_map(|b| {
+                b.column(0)
+                    .as_any()
+                    .downcast_ref::<Int32Array>()
+                    .unwrap()
+                    .values()
+                    .iter()
+                    .copied()
+            })
+            .collect();
+        let expected: Vec<i32> = (0..237).collect();
+        assert_eq!(values, expected);
+    }
+
+    #[test]
+    fn test_min_bytes_zero_still_yields_all_rows() {
+        // When min_bytes=0, the stream should still yield every batch.
+        // This is the "chop only, don't coalesce" use case.
+        let batches: Vec<_> = (0..5).map(|_| make_batch(100)).collect();
+        let batch_bytes = batches[0].get_array_memory_size();
+        let result = collect_rechunked(batches, 0, batch_bytes * 2);
+        assert_eq!(total_rows(&result), 500);
+    }
+
+    #[test]
+    fn test_min_bytes_zero_slices_oversized() {
+        // min_bytes=0 with a small max_bytes should still slice large batches.
+        let batch = make_batch(1000);
+        let bytes = batch.get_array_memory_size();
+        let result = collect_rechunked(vec![batch], 0, bytes / 4);
+        assert_eq!(total_rows(&result), 1000);
+        assert!(
+            result.len() >= 4,
+            "expected at least 4 slices, got {}",
+            result.len()
+        );
+    }
+
+    /// Build a batch with one variable-length string column.
+    /// Every row is `small_size` bytes except the row at index `big_row_idx`
+    /// which is `big_size` bytes.
+    fn make_variable_batch(
+        num_rows: usize,
+        small_size: usize,
+        big_row_idx: usize,
+        big_size: usize,
+    ) -> RecordBatch {
+        let schema = Arc::new(Schema::new(vec![Field::new("s", DataType::Utf8, false)]));
+        let values: Vec<String> = (0..num_rows)
+            .map(|i| {
+                if i == big_row_idx {
+                    "X".repeat(big_size)
+                } else {
+                    "x".repeat(small_size)
+                }
+            })
+            .collect();
+        let array = arrow_array::StringArray::from(values);
+        RecordBatch::try_new(schema, vec![Arc::new(array)]).unwrap()
+    }
+
+    fn variable_schema() -> SchemaRef {
+        Arc::new(Schema::new(vec![Field::new("s", DataType::Utf8, false)]))
+    }
+
+    fn collect_rechunked_variable(
+        batches: Vec<RecordBatch>,
+        min_bytes: usize,
+        max_bytes: usize,
+    ) -> Vec<RecordBatch> {
+        let input = stream::iter(batches.into_iter().map(Ok::<_, ArrowError>));
+        let rechunked =
+            rechunk_stream_by_size_deep_copy(input, variable_schema(), min_bytes, max_bytes);
+        block_on(rechunked.collect::<Vec<_>>())
+            .into_iter()
+            .map(|r| r.unwrap())
+            .collect()
+    }
+
+    #[test]
+    fn test_oversized_row_at_end() {
+        // 100 rows: 99 small (64 bytes each) + 1 large (100KiB) at the end.
+        let batch = make_variable_batch(100, 64, 99, 100 * 1024);
+        let max_bytes = 64 * 1024;
+        let result = collect_rechunked_variable(vec![batch], 0, max_bytes);
+        assert_eq!(total_rows(&result), 100);
+        for (i, b) in result.iter().enumerate() {
+            let size = b.get_array_memory_size();
+            assert!(
+                size <= max_bytes || b.num_rows() == 1,
+                "batch {i} has {size} bytes (max {max_bytes}) and {} rows",
+                b.num_rows()
+            );
+        }
+    }
+
+    #[test]
+    fn test_oversized_row_at_start() {
+        // 100 rows: 1 large (100KiB) at the start + 99 small (64 bytes each).
+        let batch = make_variable_batch(100, 64, 0, 100 * 1024);
+        let max_bytes = 64 * 1024;
+        let result = collect_rechunked_variable(vec![batch], 0, max_bytes);
+        assert_eq!(total_rows(&result), 100);
+        for (i, b) in result.iter().enumerate() {
+            let size = b.get_array_memory_size();
+            assert!(
+                size <= max_bytes || b.num_rows() == 1,
+                "batch {i} has {size} bytes (max {max_bytes}) and {} rows",
+                b.num_rows()
+            );
+        }
+    }
+
+    #[test]
+    fn test_oversized_row_in_middle() {
+        // 100 rows: 1 large (100KiB) in the middle + 99 small (64 bytes each).
+        let batch = make_variable_batch(100, 64, 50, 100 * 1024);
+        let max_bytes = 64 * 1024;
+        let result = collect_rechunked_variable(vec![batch], 0, max_bytes);
+        assert_eq!(total_rows(&result), 100);
+        for (i, b) in result.iter().enumerate() {
+            let size = b.get_array_memory_size();
+            assert!(
+                size <= max_bytes || b.num_rows() == 1,
+                "batch {i} has {size} bytes (max {max_bytes}) and {} rows",
+                b.num_rows()
+            );
+        }
+    }
+
+    #[test]
+    fn test_error_propagation() {
+        let input = stream::iter(vec![
+            Ok(make_batch(10)),
+            Err(ArrowError::ComputeError("boom".into())),
+            Ok(make_batch(10)),
+        ]);
+        let rechunked = rechunk_stream_by_size(input, test_schema(), 1, usize::MAX);
+        let results: Vec<Result<RecordBatch, ArrowError>> = block_on(rechunked.collect());
+        assert!(results.iter().any(|r| r.is_err()));
+    }
+}

python/vendor/lance-arrow/src/struct.rs

@@ -0,0 +1,191 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+
+//! Extension to arrow struct arrays
+
+use arrow_array::{Array, StructArray, cast::AsArray, make_array};
+use arrow_buffer::NullBuffer;
+use arrow_data::{ArrayData, ArrayDataBuilder};
+use arrow_schema::ArrowError;
+
+pub trait StructArrayExt {
+    /// Removes the offset / length of the struct array by pushing it into the children
+    ///
+    /// In arrow-rs when slice is called it recursively slices the children.
+    /// In arrow-cpp when slice is called it just sets the offset/length of
+    ///   the struct array and leaves the children as-is
+    ///
+    /// Both are legal approaches (╥﹏╥)
+    ///
+    /// This method helps reduce complexity by folding into the arrow-rs approach
+    fn normalize_slicing(&self) -> Result<Self, ArrowError>
+    where
+        Self: Sized;
+
+    /// Structs are allowed to mask valid items.  For example, a struct array might be:
+    ///
+    /// [ {"items": [1, 2, 3]}, NULL, {"items": NULL}]
+    ///
+    /// However, the underlying items array might be: [[1, 2, 3], [4, 5], NULL]
+    ///
+    /// The [4, 5] list is masked out because the struct array is null.
+    ///
+    /// The struct validity would be [true, false, true] and the list validity would be [true, true, false]
+    ///
+    /// This method pushes nulls down into all children.  In the above example the list validity would become
+    /// [true, false, false].
+    ///
+    /// This method is not recursive.  If a child is a struct array it will not push that child's nulls down.
+    ///
+    /// This method does not remove garbage lists.  It only updates the validity so a future call to
+    /// [crate::list::ListArrayExt::filter_garbage_nulls] will remove the garbage lists (without
+    /// this pushdown it would not)
+    fn pushdown_nulls(&self) -> Result<Self, ArrowError>
+    where
+        Self: Sized;
+}
+
+fn normalized_struct_array_data(data: ArrayData) -> Result<ArrayData, ArrowError> {
+    let parent_offset = data.offset();
+    let parent_len = data.len();
+    let modified_children = data
+        .child_data()
+        .iter()
+        .map(|d| {
+            let d = normalized_struct_array_data(d.clone())?;
+            let offset = d.offset();
+            let len = d.len();
+            if len < parent_len + parent_offset {
+                return Err(ArrowError::InvalidArgumentError(format!(
+                    "Child array {} has length {} which is less than the parent length {} plus the parent offset {}",
+                    d.data_type(),
+                    len,
+                    parent_len,
+                    parent_offset
+                )));
+            }
+            let new_offset = offset + parent_offset;
+            d.into_builder().offset(new_offset)
+                .len(parent_len)
+                .build()
+        })
+        .collect::<Result<Vec<_>, _>>()?;
+    ArrayDataBuilder::new(data.data_type().clone())
+        .len(parent_len)
+        .offset(0)
+        .buffers(data.buffers().to_vec())
+        .child_data(modified_children)
+        .build()
+}
+
+impl StructArrayExt for StructArray {
+    fn normalize_slicing(&self) -> Result<Self, ArrowError>
+    where
+        Self: Sized,
+    {
+        if self.offset() == 0 && self.columns().iter().all(|c| c.len() == self.len()) {
+            return Ok(self.clone());
+        }
+
+        let data = normalized_struct_array_data(self.to_data())?;
+        Ok(Self::from(data))
+    }
+
+    fn pushdown_nulls(&self) -> Result<Self, ArrowError>
+    where
+        Self: Sized,
+    {
+        let Some(validity) = self.nulls() else {
+            return Ok(self.clone());
+        };
+        let data = self.to_data();
+        let children = data
+            .child_data()
+            .iter()
+            .map(|c| {
+                if let Some(child_validity) = c.nulls() {
+                    let new_validity = child_validity.inner() & validity.inner();
+                    c.clone()
+                        .into_builder()
+                        .nulls(Some(NullBuffer::from(new_validity)))
+                        .build()
+                } else {
+                    Ok(c.clone()
+                        .into_builder()
+                        .nulls(Some(validity.clone()))
+                        .build()?)
+                }
+            })
+            .collect::<Result<Vec<_>, _>>()?;
+        let arr = make_array(data.into_builder().child_data(children).build()?);
+        Ok(arr.as_struct().clone())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use arrow_array::{Array, Int32Array, StructArray, cast::AsArray, make_array};
+    use arrow_schema::{DataType, Field, Fields};
+    use std::sync::Arc;
+
+    use crate::r#struct::StructArrayExt;
+
+    #[test]
+    fn test_normalize_slicing_no_offset() {
+        let x = Int32Array::from(vec![1, 2, 3]);
+        let y = Int32Array::from(vec![4, 5, 6]);
+        let struct_array = StructArray::new(
+            Fields::from(vec![
+                Field::new("x", DataType::Int32, true),
+                Field::new("y", DataType::Int32, true),
+            ]),
+            vec![Arc::new(x), Arc::new(y)],
+            None,
+        );
+
+        let normalized = struct_array.normalize_slicing().unwrap();
+        assert_eq!(normalized, struct_array);
+    }
+
+    #[test]
+    fn test_arrow_rs_slicing() {
+        let x = Int32Array::from(vec![1, 2, 3, 4]);
+        let y = Int32Array::from(vec![5, 6, 7, 8]);
+        let struct_array = StructArray::new(
+            Fields::from(vec![
+                Field::new("x", DataType::Int32, true),
+                Field::new("y", DataType::Int32, true),
+            ]),
+            vec![Arc::new(x), Arc::new(y)],
+            None,
+        );
+
+        // Slicing with arrow-rs propagates the slicing to the children so there should
+        // be no change needed to the struct array
+        let sliced = struct_array.slice(1, 2);
+        let normalized = sliced.normalize_slicing().unwrap();
+
+        assert_eq!(normalized, sliced);
+    }
+
+    #[test]
+    fn test_arrow_cpp_slicing() {
+        let x = Int32Array::from(vec![1, 2, 3, 4]);
+        let y = Int32Array::from(vec![5, 6, 7, 8]);
+        let struct_array = StructArray::new(
+            Fields::from(vec![
+                Field::new("x", DataType::Int32, true),
+                Field::new("y", DataType::Int32, true),
+            ]),
+            vec![Arc::new(x), Arc::new(y)],
+            None,
+        );
+
+        let data = struct_array.to_data();
+        let sliced = data.into_builder().offset(1).len(2).build().unwrap();
+        let sliced = make_array(sliced);
+        let normalized = sliced.as_struct().clone().normalize_slicing().unwrap();
+
+        assert_eq!(normalized, struct_array.slice(1, 2));
+    }
+}

rust/lancedb/tests/nullable_struct_merge.rs

@@ -0,0 +1,64 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+use std::sync::Arc;
+
+use arrow_array::{Array, ArrayRef, BooleanArray, Int32Array, RecordBatch, StructArray};
+use arrow_buffer::NullBuffer;
+use arrow_schema::{DataType, Field, Schema};
+use lance_arrow::RecordBatchExt;
+
+#[test]
+fn merge_nullable_struct_with_sliced_boolean_child() {
+    let left_struct = StructArray::new(
+        vec![Field::new("id", DataType::Int32, false)].into(),
+        vec![Arc::new(Int32Array::from(vec![1, 2, 3, 4])) as ArrayRef],
+        None,
+    );
+
+    // BooleanArray retains its offset when sliced. The merged child bitmap must
+    // cover that offset as well as the four visible rows.
+    let boolean_child: ArrayRef = Arc::new(BooleanArray::from_iter((0..1030).map(|index| {
+        if index == 1027 {
+            None
+        } else {
+            Some(index % 2 == 1)
+        }
+    })));
+    let boolean_child = boolean_child.slice(1025, 4);
+    assert_eq!(boolean_child.offset(), 1025);
+    let right_struct = StructArray::new(
+        vec![Field::new("flag", DataType::Boolean, true)].into(),
+        vec![boolean_child],
+        Some(NullBuffer::from(vec![true, false, true, true])),
+    );
+
+    let left_batch = RecordBatch::try_new(
+        Arc::new(Schema::new(vec![Field::new(
+            "nested",
+            left_struct.data_type().clone(),
+            true,
+        )])),
+        vec![Arc::new(left_struct)],
+    )
+    .unwrap();
+    let right_batch = RecordBatch::try_new(
+        Arc::new(Schema::new(vec![Field::new(
+            "nested",
+            right_struct.data_type().clone(),
+            true,
+        )])),
+        vec![Arc::new(right_struct)],
+    )
+    .unwrap();
+
+    let merged = left_batch.merge(&right_batch).unwrap();
+    let nested = merged.column_by_name("nested").unwrap();
+    let nested = nested.as_any().downcast_ref::<StructArray>().unwrap();
+    let flag = nested.column_by_name("flag").unwrap();
+    let flag = flag.as_any().downcast_ref::<BooleanArray>().unwrap();
+    assert_eq!(
+        flag.iter().collect::<Vec<_>>(),
+        vec![Some(true), None, None, Some(false)]
+    );
+}