Let multivector runs straddle chunk boundaries (#10283)

* Read vector runs that straddle a chunk boundary Resolve a run into per-chunk parts instead of a single range, borrowing when it lands in one chunk and copying when it spans two. The read pipeline schedules one range per read, so a straddling run is read outside it. No writer produces such a run yet, so this changes nothing on its own. It is what a reader needs before one does — including edge and live-reload readers, which read files a different version wrote. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * Place multivector runs without regard to chunk boundaries Writers appended a multivector's inner vectors at the end of the row space unless the run would cross a chunk boundary, in which case they skipped the chunk tail — the batch writers padding the skipped rows with explicit zero rows. That made chunk geometry part of the interface every multivector storage had to reuse. Runs now go at the end unconditionally and the chunked storage splits the write across chunks, as it already did for a batch of single vectors. What is left of the geometry is a size cap: a multivector may not exceed one chunk. It is fill-independent, so it constrains nothing about placement, and it is what the volatile storage needs anyway — that one returns a plain slice and so cannot serve a straddling run. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * Split a run at chunk boundaries in one place Reading, writing in place and appending each derived the split from `remaining_chunk_capacity`, so every one of them had to know that a run does not necessarily fit where it starts. `split_run` hands out the parts instead: one per chunk the run covers, each carrying where it goes and how much of the run it takes. Nothing asks how much room is left any more, and `get_chunk_offset` goes with it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * Keep straddling runs on the read pipeline Reading a straddling run outside the pipeline blocked the scheduling loop on one read, which costs a round trip on a backend that fetches remotely and drops the batch back to sequential. A run is now scheduled as one read per chunk it covers. Parts complete in any order, so each run holds what has landed until the last part does, then hands the callback the stitched vectors. Runs taking a single read carry the caller's data in the tag and never touch that table. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * Stop capping a multivector at one chunk The cap outlived its reason on disk, but the volatile storage still needed it: its `get_many` handed out a slice of one chunk, so a run that crossed a boundary had nowhere to come from. And since a volatile storage is a target of the batched copy that builds a segment, dropping the cap only on disk would have turned a rejected write into a failed merge. So the volatile storage splits and stitches too. Both are a few lines each, and placing a run no longer skips a chunk tail, so `extend` is now `insert_many` at the end of the storage. Nothing user-facing moves: `MAX_MULTIVECTOR_FLATTENED_LEN` caps a multivector at 1M elements, far inside a 32 MiB chunk, so the storages only ever rejected what reached them unvalidated. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * Schedule a single-read run without the queue The scheduling loop resolved every run into the queue and then took it straight back out, so the overwhelmingly common run — one that fits a chunk — paid a push and a pop for nothing. It now goes to the pipeline directly, and the queue holds only what a straddling run leaves behind. Worth ~10% on the multivector read benchmark, and it collapses the "top up, then take" pair into one decision. Extracting that bookkeeping into helpers instead was measured and is much worse: the mmap pipeline alternates one schedule with one wait, so the loop body is a few dozen nanoseconds, and a helper carrying the cold map and stitching paths is too big for the compiler to inline back into it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * Repoint the multivector WAL-replay test at a live rejection The test upserted a multivector too large for a storage chunk, which no longer fails: the storages stopped capping one at a chunk. Nothing else covered a multivector operation that only the apply path rejects. A raw blob that is not a whole number of quantized records still does, so the test now uses that, alongside its dense and sparse siblings. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * Test reading multivectors with legacy chunk-tail padding Locks the compatibility contract that pre-straddle files — runs that skip a chunk's leftover slots — still reopen as single-chunk borrows. * chore: retrigger CI after flaky test-consensus-compose * Move ReadTag into for_each_vector, its only user Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AN4Hgbd65gDhesthJk5bUY --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com> Co-authored-by: qdrant-cloud-bot <111755117+qdrant-cloud-bot@users.noreply.github.com>
2026-09-21 13:37:46 -05:00 · 2026-09-03 12:45:53 +02:00
parent 374506c2c9
commit 1dad4d9f9f
15 changed files with 631 additions and 405 deletions
@@ -1705,16 +1705,12 @@ async fn test_malformed_sparse_raw_upsert_is_skipped_on_wal_replay() {
    assert_bad_op_skipped_on_wal_replay(config, valid_upsert, malformed_raw_upsert).await;
 }

-/// A multivector exceeding whole-chunk capacity in TurboQuant storage is only
-/// rejected during apply, after the WAL write. Public APIs can't produce one
-/// (`MAX_MULTIVECTOR_FLATTENED_LEN`); the internal API applies operations
-/// without that validation.
+/// Multivector counterpart of [`test_malformed_raw_upsert_is_skipped_on_wal_replay`]:
+/// a raw blob that is not a whole number of TurboQuant records. Only the apply
+/// path rejects it, after the WAL write.
 #[tokio::test(flavor = "multi_thread")]
-async fn test_over_capacity_multivector_upsert_is_skipped_on_wal_replay() {
-    // Test builds use a 512 KiB storage chunk; a Turbo4 record of a dim-128
-    // subvector is at least 64 bytes, so one chunk holds at most 8192 subvectors.
+async fn test_malformed_multivector_raw_upsert_is_skipped_on_wal_replay() {
    const DIM: usize = 128;
-    const OVER_CAPACITY_COUNT: usize = 9000;

    // Single multivector with TurboQuant (Turbo4) storage.
    let mut vector_params = VectorParamsBuilder::new(DIM as u64, Distance::Dot)
@@ -1727,23 +1723,28 @@ async fn test_over_capacity_multivector_upsert_is_skipped_on_wal_replay() {
        vectors: VectorsConfig::Single(vector_params),
        ..CollectionParams::empty()
    };
-    // The over-capacity operation (~4.6 MB of f32 input) must fit a WAL segment.
-    config.wal_config.wal_capacity_mb = 16;

-    let multi_upsert = |id: u64, count: usize| {
-        CollectionUpdateOperations::PointOperation(PointOperations::UpsertPoints(
-            PointInsertOperationsInternal::from(vec![PointStructPersisted {
-                id: id.into(),
-                vector: VectorStructPersisted::MultiDense(vec![vec![1.0; DIM]; count]),
+    // A valid multivector point via the plain upsert path.
+    let valid_upsert = CollectionUpdateOperations::PointOperation(PointOperations::UpsertPoints(
+        PointInsertOperationsInternal::from(vec![PointStructPersisted {
+            id: 1.into(),
+            vector: VectorStructPersisted::MultiDense(vec![vec![1.0; DIM]; 2]),
+            payload: None,
+        }]),
+    ));
+
+    // A quantized record of a dim-128 subvector is at least 64 bytes, so three
+    // bytes are never a whole number of them.
+    let malformed_raw_upsert =
+        CollectionUpdateOperations::PointOperation(PointOperations::UpsertPointsRaw(vec![
+            PointStructRawPersisted {
+                id: 2.into(),
+                vectors: std::iter::once((DEFAULT_VECTOR_NAME.to_owned(), vec![0_u8, 1, 2]))
+                    .collect(),
                payload: None,
-            }]),
-        ))
-    };
+                payload_raw: None,
+            },
+        ]));

-    assert_bad_op_skipped_on_wal_replay(
-        config,
-        multi_upsert(1, 2),
-        multi_upsert(2, OVER_CAPACITY_COUNT),
-    )
-    .await;
+    assert_bad_op_skipped_on_wal_replay(config, valid_upsert, malformed_raw_upsert).await;
 }
@@ -35,22 +35,74 @@ pub(super) struct ChunkedVectorsConfig {
    pub(super) populate: Option<bool>,
 }

+/// One chunk's share of a run of vectors.
+pub(super) struct RunPart {
+    pub chunk_idx: usize,
+    /// Where the part starts within its chunk, in elements.
+    pub element_offset: usize,
+    /// Vectors in the part.
+    pub count: usize,
+}
+
 impl ChunkedVectorsConfig {
    pub fn get_chunk_index(&self, key: usize) -> usize {
        key / self.chunk_size_vectors
    }

-    pub fn get_chunk_offset(&self, key: usize) -> usize {
-        let chunk_vector_idx = key % self.chunk_size_vectors;
-        chunk_vector_idx * self.dim
-    }
+    /// Split the run of `count` vectors at `key` into one part per chunk it
+    /// covers, in order.
+    ///
+    /// Always yields at least one part, so an empty run still resolves to a
+    /// position.
+    pub fn split_run(&self, key: usize, count: usize) -> RunParts<'_> {
+        let parts = match count {
+            0 => 1,
+            count => self.get_chunk_index(key + count - 1) - self.get_chunk_index(key) + 1,
+        };

-    /// How many vectors still fit in the chunk holding `key`, starting at it.
-    pub fn remaining_chunk_capacity(&self, key: usize) -> usize {
-        self.chunk_size_vectors - key % self.chunk_size_vectors
+        RunParts {
+            config: self,
+            key,
+            left: count,
+            parts,
+        }
    }
 }

+/// Iterator of [`ChunkedVectorsConfig::split_run`].
+pub(super) struct RunParts<'a> {
+    config: &'a ChunkedVectorsConfig,
+    key: usize,
+    left: usize,
+    parts: usize,
+}
+
+impl Iterator for RunParts<'_> {
+    type Item = RunPart;
+
+    fn next(&mut self) -> Option<RunPart> {
+        self.parts = self.parts.checked_sub(1)?;
+
+        let in_chunk = self.key % self.config.chunk_size_vectors;
+        let part = RunPart {
+            chunk_idx: self.key / self.config.chunk_size_vectors,
+            element_offset: in_chunk * self.config.dim,
+            count: self.left.min(self.config.chunk_size_vectors - in_chunk),
+        };
+
+        self.key += part.count;
+        self.left -= part.count;
+
+        Some(part)
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (self.parts, Some(self.parts))
+    }
+}
+
+impl ExactSizeIterator for RunParts<'_> {}
+
 /// Load the stored config, or create it (and its file) on first open.
 pub(super) fn ensure_config<T, Fs>(
    fs: &Fs,
@@ -1,7 +1,8 @@
 //! Fixed-dimension vectors stored flattened across a directory of chunk files.
 //!
 //! The directory holds a config file, a status file carrying the vector count,
-//! and `chunk_<n>.mmap` files. A vector never straddles a chunk boundary.
+//! and `chunk_<n>.mmap` files. A single vector never straddles a chunk
+//! boundary, a run of them may.
 //!
 //! Three types share that layout:
 //!
@@ -63,10 +64,13 @@ where

 #[cfg(test)]
 mod tests {
+    use std::borrow::Cow;
    use std::iter::zip;

    use common::counter::hardware_counter::HardwareCounterCell;
+    use common::generic_consts::Random;
    use common::mmap::AdviceSetting;
+    use common::types::PointOffsetType;
    use common::universal_io::{MmapFile, MmapFs, Populate};
    use rand::SeedableRng;
    use rand::prelude::StdRng;
@@ -151,4 +155,110 @@ mod tests {
            chunked_mmap.flusher()().unwrap();
        }
    }
+
+    /// A run of vectors crossing a chunk boundary is written to both chunks and
+    /// read back as one copied slice.
+    #[test]
+    fn run_across_chunk_boundary_round_trips() {
+        let dir = Builder::new().prefix("storage_dir").tempdir().unwrap();
+        let dim = 500;
+        let hw_counter = HardwareCounterCell::new();
+        let mut rng = StdRng::seed_from_u64(42);
+
+        let mut chunked_mmap: ChunkedVectors<VectorElementType, MmapFile> = ChunkedVectors::open(
+            MmapFs,
+            dir.path(),
+            dim,
+            AdviceSetting::Global,
+            Populate::Blocking,
+        )
+        .unwrap();
+
+        // Start the run two vectors before the boundary, so it spans both chunks
+        let per_chunk = chunked_mmap.config.chunk_size_vectors;
+        let start = per_chunk - 2;
+        let count = 5;
+        let run: Vec<VectorElementType> = (0..count)
+            .flat_map(|_| random_vector(&mut rng, dim))
+            .collect();
+
+        chunked_mmap
+            .insert_many(start, &run, count, &hw_counter)
+            .unwrap();
+
+        assert_eq!(chunked_mmap.chunks.len(), 2);
+        assert_eq!(chunked_mmap.len(), start + count);
+
+        let read = chunked_mmap.get_many::<Random>(start, count).unwrap();
+        assert!(matches!(read, Cow::Owned(_)), "straddling read must copy");
+        assert_eq!(read.as_ref(), run.as_slice());
+
+        // The parts are readable on their own too, borrowed from their chunk
+        for (i, vector) in run.chunks_exact(dim).enumerate() {
+            let one = chunked_mmap.get::<Random>(start + i).unwrap();
+            assert!(matches!(one, Cow::Borrowed(_)));
+            assert_eq!(one.as_ref(), vector);
+        }
+    }
+
+    /// The batched path schedules one read per chunk a run covers and stitches
+    /// them once they land, alongside runs that take a single read.
+    #[test]
+    fn for_each_vector_stitches_straddling_runs() {
+        let dir = Builder::new().prefix("storage_dir").tempdir().unwrap();
+        let dim = 500;
+        let hw_counter = HardwareCounterCell::new();
+        let mut rng = StdRng::seed_from_u64(42);
+
+        let mut chunked_mmap: ChunkedVectors<VectorElementType, MmapFile> = ChunkedVectors::open(
+            MmapFs,
+            dir.path(),
+            dim,
+            AdviceSetting::Global,
+            Populate::Blocking,
+        )
+        .unwrap();
+
+        let per_chunk = chunked_mmap.config.chunk_size_vectors;
+        let straddle_start = per_chunk - 2;
+        let straddle_count = 5;
+
+        // A run before the boundary, one across it, one after it
+        let runs = [
+            (0, 1),
+            (straddle_start, straddle_count),
+            (straddle_start + straddle_count, 3),
+        ];
+        let expected: Vec<Vec<VectorElementType>> = runs
+            .iter()
+            .map(|&(_, count)| {
+                (0..count)
+                    .flat_map(|_| random_vector(&mut rng, dim))
+                    .collect()
+            })
+            .collect();
+
+        for (&(start, count), vectors) in zip(&runs, &expected) {
+            chunked_mmap
+                .insert_many(start, vectors, count, &hw_counter)
+                .unwrap();
+        }
+
+        let mut read = vec![None; runs.len()];
+        chunked_mmap
+            .for_each_vector::<Random, _>(
+                runs.iter()
+                    .enumerate()
+                    .map(|(i, &(start, count))| (i, start as PointOffsetType, count as u32)),
+                |i, vectors| {
+                    read[i] = Some(vectors.to_vec());
+                    Ok(())
+                },
+            )
+            .unwrap();
+
+        for (i, expected) in expected.iter().enumerate() {
+            assert_eq!(read[i].as_ref(), Some(expected), "run {i}");
+        }
+    }
 }
@@ -1,18 +1,47 @@
 use std::borrow::Cow;
+use std::collections::VecDeque;
 use std::mem::MaybeUninit;

+use ahash::AHashMap;
 use common::generic_consts::{AccessPattern, Random, Sequential};
 use common::maybe_uninit::maybe_uninit_fill_from;
 use common::types::PointOffsetType;
 use common::universal_io::{ReadPipeline, ReadRange, TypedStorage, UniversalRead, UserData};
-use num_traits::AsPrimitive;

 use super::ReadOnlyChunkedVectors;
 use crate::common::operation_error::{OperationError, OperationResult};
+use crate::vector_storage::chunked_vectors::config::RunPart;
 use crate::vector_storage::common::{PAGE_SIZE_BYTES, VECTOR_READ_BATCH_SIZE};
 use crate::vector_storage::query_scorer::is_read_with_prefetch_efficient;
 use crate::vector_storage::{VectorOffset, VectorOffsetType};

+/// A run whose parts are still arriving from the read pipeline.
+struct SplitRun<'a, U, T: Clone> {
+    user_data: U,
+    landed: Vec<Option<Cow<'a, [T]>>>,
+    missing: usize,
+}
+
+impl<'a, U, T: Clone> SplitRun<'a, U, T> {
+    fn new(user_data: U, parts: usize) -> Self {
+        Self {
+            user_data,
+            landed: (0..parts).map(|_| None).collect(),
+            missing: parts,
+        }
+    }
+
+    /// The whole run, in order. Every part must have landed.
+    fn stitch(self) -> (U, Cow<'a, [T]>) {
+        let mut stitched = Vec::new();
+        for part in self.landed {
+            stitched.extend_from_slice(&part.expect("every part landed"));
+        }
+
+        (self.user_data, Cow::Owned(stitched))
+    }
+}
+
 impl<T: bytemuck::Pod + Send, S: UniversalRead> ReadOnlyChunkedVectors<T, S> {
    #[inline]
    pub fn max_vector_size_bytes(&self) -> usize {
@@ -29,51 +58,41 @@ impl<T: bytemuck::Pod + Send, S: UniversalRead> ReadOnlyChunkedVectors<T, S> {
        self.config.dim
    }

-    // returns how many vectors can be inserted starting from key
-    pub fn get_remaining_chunk_keys(&self, start_key: VectorOffsetType) -> usize {
-        self.config.remaining_chunk_capacity(start_key.as_())
-    }
-
-    #[inline]
-    fn read_range(&self, offset: VectorOffsetType, count: usize) -> Option<(usize, ReadRange)> {
+    /// Per-chunk parts of `offset..offset + count`, in order.
+    ///
+    /// More than one part when the run straddles a chunk boundary.
+    fn read_ranges(
+        &self,
+        offset: VectorOffsetType,
+        count: usize,
+    ) -> Option<impl ExactSizeIterator<Item = (usize, ReadRange)> + '_> {
        if offset.checked_add(count)? > self.len {
            return None;
        }

-        let chunk_idx = self.config.get_chunk_index(offset);
-        if chunk_idx >= self.chunks.len() {
-            return None;
-        }
+        Some(self.config.split_run(offset, count).map(|part| {
+            let RunPart {
+                chunk_idx,
+                element_offset,
+                count,
+            } = part;

-        let element_offset = self.config.get_chunk_offset(offset);
-        let elements_length = count * self.config.dim;
-        if element_offset + elements_length > self.config.chunk_size_vectors * self.config.dim {
-            return None;
-        }
+            let range = ReadRange {
+                byte_offset: (element_offset * size_of::<T>()) as u64,
+                length: (count * self.config.dim) as u64,
+            };

-        let range = ReadRange {
-            byte_offset: (element_offset * size_of::<T>()) as u64,
-            length: elements_length as u64,
-        };
-
-        Some((chunk_idx, range))
+            (chunk_idx, range)
+        }))
    }

-    /// Returns `count` flattened vectors starting from `starting_key`.
-    ///
-    /// Returns `None` when:
-    /// - chunk boundary is crossed
-    /// - any section of `start_key..start_key + count` is out of bounds
-    #[inline]
-    fn get_many_impl(
+    fn read_part(
        &self,
-        start_key: VectorOffsetType,
-        count: usize,
+        chunk_idx: usize,
+        range: ReadRange,
        force_sequential: bool,
    ) -> Option<Cow<'_, [T]>> {
-        let (chunk_idx, range) = self.read_range(start_key, count)?;
-
-        let chunk = &self.chunks[chunk_idx];
+        let chunk = self.chunks.get(chunk_idx)?;

        let use_sequential =
            force_sequential || range.length as usize * size_of::<T>() > PAGE_SIZE_BYTES * 4;
@@ -85,6 +104,31 @@ impl<T: bytemuck::Pod + Send, S: UniversalRead> ReadOnlyChunkedVectors<T, S> {
        }
    }

+    /// Returns `count` flattened vectors starting from `starting_key`.
+    ///
+    /// Borrows a single chunk, or copies when the run straddles a boundary.
+    /// Returns `None` when any section of `start_key..start_key + count` is
+    /// out of bounds.
+    #[inline]
+    fn get_many_impl(
+        &self,
+        start_key: VectorOffsetType,
+        count: usize,
+        force_sequential: bool,
+    ) -> Option<Cow<'_, [T]>> {
+        let mut parts = self.read_ranges(start_key, count)?;
+
+        let (chunk_idx, range) = parts.next()?;
+        let mut vectors = self.read_part(chunk_idx, range, force_sequential)?;
+
+        for (chunk_idx, range) in parts {
+            let part = self.read_part(chunk_idx, range, force_sequential)?;
+            vectors.to_mut().extend_from_slice(&part);
+        }
+
+        Some(vectors)
+    }
+
    #[inline]
    pub fn get<P: AccessPattern>(&self, key: VectorOffsetType) -> Option<Cow<'_, [T]>> {
        self.get_many_impl(key, 1, P::IS_SEQUENTIAL)
@@ -149,7 +193,8 @@ impl<T: bytemuck::Pod + Send, S: UniversalRead> ReadOnlyChunkedVectors<T, S> {
    /// Invoke `callback` for each flattened multi-vector at the given offsets.
    ///
    /// Drives the read pipeline directly across chunk files: refills it from the
-    /// offsets, then drains completed reads.
+    /// offsets, then drains completed reads. A run spanning several chunks is
+    /// one scheduled read per chunk, stitched once the last one lands.
    pub fn for_each_vector<P, U>(
        &self,
        mut offsets: impl Iterator<Item = (U, PointOffsetType, u32)>,
@@ -159,29 +204,93 @@ impl<T: bytemuck::Pod + Send, S: UniversalRead> ReadOnlyChunkedVectors<T, S> {
        P: AccessPattern,
        U: UserData,
    {
+        /// What a scheduled read carries back, so its completion knows what it is.
+        #[derive(Debug)]
+        enum ReadTag<U> {
+            /// The run takes this one read, and the caller's data rides along
+            /// with it
+            Whole(U),
+            /// One part of a run taking several reads, filed under `run` until
+            /// the others land
+            Part { run: u32, index: u32 },
+        }
+
+        let out_of_bounds = || OperationError::service_error("vector offset out of bounds");
+
        // access pattern does not matter for io_uring
-        let mut pipeline = S::ReadPipeline::<'_, U>::new()?;
+        let mut pipeline = S::ReadPipeline::<'_, ReadTag<U>>::new()?;
+
+        // A run resolves to as many reads as it covers chunks, which can be more
+        // than the pipeline has room for, so they wait here
+        let mut queued: VecDeque<(ReadTag<U>, usize, ReadRange)> = VecDeque::new();
+        // Stays empty unless a run straddles a chunk boundary
+        let mut split_runs: AHashMap<u32, SplitRun<'_, U, T>> = AHashMap::new();
+        let mut next_run: u32 = 0;

        loop {
-            while pipeline.can_schedule()
-                && let Some((user_data, offset, count)) = offsets.next()
-            {
-                let (chunk_idx, range) = self
-                    .read_range(offset as _, count as _)
-                    .ok_or_else(|| OperationError::service_error("vector offset out of bounds"))?;
-                let range = range.into_byte_range::<T>();
+            while pipeline.can_schedule() {
+                // Parts of a split run that had no room last time go first
+                let (tag, chunk_idx, range) = match queued.pop_front() {
+                    Some(read) => read,
+                    None => {
+                        let Some((user_data, offset, count)) = offsets.next() else {
+                            break;
+                        };
+
+                        let mut ranges = self
+                            .read_ranges(offset as _, count as _)
+                            .ok_or_else(out_of_bounds)?;
+
+                        // A run across chunks is queued whole and taken from the
+                        // top on the next turns
+                        if ranges.len() > 1 {
+                            let run = next_run;
+                            next_run = run.wrapping_add(1);
+                            split_runs.insert(run, SplitRun::new(user_data, ranges.len()));
+
+                            queued.extend(ranges.enumerate().map(|(index, (chunk_idx, range))| {
+                                let index = index as u32;
+                                (ReadTag::Part { run, index }, chunk_idx, range)
+                            }));
+                            continue;
+                        }
+
+                        let (chunk_idx, range) = ranges.next().ok_or_else(out_of_bounds)?;
+                        (ReadTag::Whole(user_data), chunk_idx, range)
+                    }
+                };
+
+                let chunk = self.chunks.get(chunk_idx).ok_or_else(out_of_bounds)?;
                pipeline.schedule::<P>(
-                    user_data,
-                    &self.chunks[chunk_idx].inner,
-                    range,
+                    tag,
+                    &chunk.inner,
+                    range.into_byte_range::<T>(),
                    align_of::<T>(),
                )?;
            }

-            let Some((user_data, vector)) = pipeline.wait_bytemuck::<T>()? else {
+            let Some((tag, vectors)) = pipeline.wait_bytemuck::<T>()? else {
+                debug_assert!(queued.is_empty(), "scheduling left reads behind");
+                debug_assert!(split_runs.is_empty(), "a run never got all its parts");
                break;
            };
-            callback(user_data, vector)?;
+
+            let (user_data, vectors) = match tag {
+                ReadTag::Whole(user_data) => (user_data, vectors),
+                ReadTag::Part { run, index } => {
+                    let split = split_runs.get_mut(&run).expect("part of an in-flight run");
+                    split.landed[index as usize] = Some(vectors);
+                    split.missing -= 1;
+
+                    if split.missing > 0 {
+                        continue;
+                    }
+
+                    split_runs.remove(&run).expect("just filed").stitch()
+                }
+            };
+
+            callback(user_data, vectors)?;
        }

        Ok(())
@@ -75,14 +75,6 @@ where
        Ok(read_status_len(&self.fs, &status_file(&self.directory))?)
    }

-    /// How many more vectors fit in the chunk that `key` falls in.
-    ///
-    /// A vector never straddles a chunk, so a caller placing a run of rows has
-    /// to skip to the next chunk when the run does not fit in this one.
-    pub fn remaining_chunk_keys(&self, key: usize) -> usize {
-        self.config.remaining_chunk_capacity(key)
-    }
-
    /// Replace the stored vector count.
    fn save_len(&self, len: usize) -> OperationResult<()> {
        self.fs.atomic_save(
@@ -176,42 +168,47 @@ where
    /// match the argument
    // Takes &mut self to enforce the single-writer contract the appends rest on
    #[allow(clippy::needless_pass_by_ref_mut)]
-    pub fn append_many<'a>(
+    pub fn append_many<'a, I>(
        &mut self,
        start_key: VectorOffsetType,
-        vectors: impl IntoIterator<Item = &'a [T]>,
+        vectors: I,
        hw_counter: &HardwareCounterCell,
-    ) -> OperationResult<()> {
+    ) -> OperationResult<()>
+    where
+        I: IntoIterator<Item = &'a [T]>,
+        I::IntoIter: ExactSizeIterator,
+    {
        self.ensure_chunk_lengths(start_key)?;

-        let mut vectors = vectors.into_iter().peekable();
-        let mut len = start_key;
+        let mut vectors = vectors.into_iter();
+        let count = vectors.len();

-        while vectors.peek().is_some() {
-            let chunk_idx = self.config.get_chunk_index(len);
-            let chunk_offset = self.config.get_chunk_offset(len);
-            let capacity = self.config.remaining_chunk_capacity(len);
+        for part in self.config.split_run(start_key, count) {
+            // The part an empty run resolves to: no batch to append, and
+            // opening its chunk would create the file for nothing
+            if part.count == 0 {
+                continue;
+            }

-            let batch: Vec<&[T]> = vectors.by_ref().take(capacity).collect();
+            let batch: Vec<&[T]> = vectors.by_ref().take(part.count).collect();
            for vector in &batch {
                assert_eq!(vector.len(), self.config.dim, "Vector size mismatch");
            }
            let batch_bytes = batch.len() * self.config.dim * size_of::<T>();

-            let mut chunk = self.open_chunk_for_append(chunk_idx, chunk_offset == 0)?;
+            let mut chunk = self.open_chunk_for_append(part.chunk_idx, part.element_offset == 0)?;
            chunk.append_batch(
-                (chunk_offset * size_of::<T>()) as u64,
+                (part.element_offset * size_of::<T>()) as u64,
                batch.iter().copied(),
            )?;
            // Flush in case of local backends.
            chunk.flusher()()?;

            hw_counter.vector_io_write_counter().incr_delta(batch_bytes);
-            len += batch.len();
        }

        // Persist the watermark only after the data landed
-        self.save_len(len)?;
+        self.save_len(start_key + count)?;

        Ok(())
    }
@@ -6,7 +6,7 @@ use num_traits::AsPrimitive;

 use super::ChunkedVectors;
 use super::chunks::create_chunk;
-use crate::common::operation_error::{OperationError, OperationResult};
+use crate::common::operation_error::OperationResult;
 use crate::vector_storage::VectorOffsetType;

 impl<T, S> ChunkedVectors<T, S>
@@ -50,26 +50,21 @@ where
        );

        let start_key = start_key.as_();
-        let chunk_idx = self.inner.config.get_chunk_index(start_key);
-        let chunk_offset = self.inner.config.get_chunk_offset(start_key);

-        // check if the vectors fit in the chunk
-        if chunk_offset + vectors.len()
-            > self.inner.config.dim * self.inner.config.chunk_size_vectors
-        {
-            return Err(OperationError::service_error(format!(
-                "Vectors do not fit in the chunk. Chunk idx {chunk_idx}, chunk offset {chunk_offset}, vectors count {count}",
-            )));
-        }
-
-        // Ensure capacity
-        while chunk_idx >= self.inner.chunks.len() {
+        // Ensure capacity for the whole run up front, so the write loop below
+        // does not need `&mut self`
+        let last_key = start_key + count.saturating_sub(1);
+        while self.inner.config.get_chunk_index(last_key) >= self.inner.chunks.len() {
            self.add_chunk()?;
        }

-        let chunk = &mut self.inner.chunks[chunk_idx];
-
-        chunk.write((chunk_offset * size_of::<T>()) as u64, vectors)?;
+        let mut rest = vectors;
+        for part in self.inner.config.split_run(start_key, count) {
+            let (elements, tail) = rest.split_at(part.count * self.inner.config.dim);
+            self.inner.chunks[part.chunk_idx]
+                .write((part.element_offset * size_of::<T>()) as u64, elements)?;
+            rest = tail;
+        }

        hw_counter
            .vector_io_write_counter()
@@ -15,7 +15,7 @@ use super::buffered_offsets::BufferedOffsets;
 use crate::common::Flusher;
 use crate::common::flags::FlagsMode;
 use crate::common::flags::bitvec_flags::BitvecFlags;
-use crate::common::operation_error::{OperationError, OperationResult, check_process_stopped};
+use crate::common::operation_error::{OperationResult, check_process_stopped};
 use crate::data_types::named_vectors::{CowMultiVector, CowVector};
 use crate::data_types::primitive::PrimitiveVectorElement;
 use crate::data_types::vectors::{
@@ -141,14 +141,6 @@ impl<T: PrimitiveVectorElement> AppendableMmapMultiDenseVectorStorage<T> {
        hw_counter: &HardwareCounterCell,
    ) -> OperationResult<()> {
        assert_eq!(multi_vector.dim, self.vectors.dim());
-        let multivector_size_in_bytes = std::mem::size_of_val(multi_vector.flattened_vectors);
-        let max_vector_size_bytes = self.vectors.max_vector_size_bytes();
-        if multivector_size_in_bytes >= max_vector_size_bytes {
-            return Err(OperationError::service_error(format!(
-                "Cannot insert multi vector of size {multivector_size_in_bytes} to the mmap vector storage.\
-                 It's too large, maximum size is {max_vector_size_bytes}."
-            )));
-        }

        let mut offset = self
            .offsets
@@ -157,11 +149,7 @@ impl<T: PrimitiveVectorElement> AppendableMmapMultiDenseVectorStorage<T> {

        if multi_vector.vectors_count() > offset.capacity as usize {
            // append vector to the end
-            let mut new_key = self.vectors.len();
-            let chunk_left_keys = self.vectors.get_remaining_chunk_keys(new_key);
-            if multi_vector.vectors_count() > chunk_left_keys {
-                new_key += chunk_left_keys;
-            }
+            let new_key = self.vectors.len();

            offset = MultivectorMmapOffset {
                offset: new_key as PointOffsetType,
@@ -795,4 +783,91 @@ mod tests {
            "point 1 must be intact — the unbuffered skew would clobber its head rows",
        );
    }
+
+    /// Pre-straddle writers skipped a chunk's leftover slots so each run stayed
+    /// inside one chunk. Fabricate that layout, flush, reopen, and confirm the
+    /// new reader still recovers the point as a single-chunk borrow.
+    #[test]
+    fn reads_legacy_layout_with_chunk_tail_padding() {
+        use crate::vector_storage::common::CHUNK_SIZE;
+
+        const DIM: usize = 128;
+        let dir = Builder::new().prefix("legacy_pad").tempdir().unwrap();
+        let hw = HardwareCounterCell::disposable();
+        let padded = multivec(3, 7.0, DIM);
+        let per_chunk = CHUNK_SIZE / (DIM * std::mem::size_of::<VectorElementType>());
+
+        {
+            let mut storage =
+                open_appendable_memmap_multi_vector_storage_impl::<VectorElementType>(
+                    dir.path(),
+                    DIM,
+                    Distance::Dot,
+                    MultiVectorConfig::default(),
+                    AdviceSetting::Global,
+                    false,
+                )
+                .unwrap();
+
+            let filler = multivec(per_chunk - 1, 1.0, DIM);
+            storage
+                .insert_vector(0, VectorRef::from(&filler), &hw)
+                .unwrap();
+            assert_eq!(storage.vectors.len(), per_chunk - 1);
+
+            // Old place(): jump to the next chunk, leaving the tail slot unused.
+            let start = per_chunk;
+            storage
+                .vectors
+                .insert_many(start as VectorOffsetType, &padded.flattened_vectors, 3, &hw)
+                .unwrap();
+            storage.offsets.set(
+                1,
+                MultivectorMmapOffset {
+                    offset: start as PointOffsetType,
+                    count: 3,
+                    capacity: 3,
+                },
+            );
+            assert_eq!(storage.vectors.len(), per_chunk + 3);
+
+            let read = storage.get_multi::<Random>(1);
+            assert!(matches!(read, CowMultiVector::Borrowed(_)));
+            assert_eq!(
+                read.as_vec_ref().flattened_vectors,
+                padded.flattened_vectors.as_slice(),
+            );
+
+            // Batched path must agree with the single-key getter.
+            let mut seen = None;
+            storage
+                .for_each_flat_multi::<Random, _>([((), 1)], |(), _, flat| {
+                    seen = Some(flat.into_owned());
+                })
+                .unwrap();
+            assert_eq!(seen.unwrap(), padded.flattened_vectors);
+
+            storage.flusher()().unwrap();
+        }
+
+        let storage = open_appendable_memmap_multi_vector_storage_impl::<VectorElementType>(
+            dir.path(),
+            DIM,
+            Distance::Dot,
+            MultiVectorConfig::default(),
+            AdviceSetting::Global,
+            false,
+        )
+        .unwrap();
+        let offset = storage.offsets.get::<Random>(1).unwrap();
+        assert_eq!(offset.offset as usize, per_chunk);
+        assert_eq!(offset.count, 3);
+
+        let read = storage.get_multi::<Random>(1);
+        assert!(matches!(read, CowMultiVector::Borrowed(_)));
+        assert_eq!(
+            read.as_vec_ref().flattened_vectors,
+            padded.flattened_vectors.as_slice(),
+        );
+    }
 }
@@ -23,8 +23,7 @@ use crate::vector_storage::update_only::VectorToStore;
 /// vectors flat, the per-point row ranges into them, and the deleted flags.
 ///
 /// Unlike the single-vector storages, rows are not indexed by point slot — a
-/// point owns a run of them — so this writer tracks where the row space ends
-/// and places each point's run itself.
+/// point owns a run of them — so this writer tracks where the row space ends.
 ///
 /// [`AppendableMmapMultiDenseVectorStorage`]: super::appendable_mmap_multi_dense_vector_storage::AppendableMmapMultiDenseVectorStorage
 pub struct UpdateOnlyMultiDenseVectorStorage<
@@ -71,10 +70,6 @@ impl<T: PrimitiveVectorElement, S: UniversalAppend + 'static>
        vectors: impl IntoIterator<Item = VectorToStore<'a>>,
        hw_counter: &HardwareCounterCell,
    ) -> OperationResult<()> {
-        // Rows are placed as we go, and a run that would straddle a chunk
-        // skips to the next one — the rows of a batch are therefore not
-        // gapless. The gaps become explicit zero rows, so a single append
-        // lands every run exactly where its offset entry points.
        let batch_start = self.next_row;
        let mut rows: Vec<T> = Vec::new();
        let mut offsets = Vec::new();
@@ -94,15 +89,13 @@ impl<T: PrimitiveVectorElement, S: UniversalAppend + 'static>
            };

            let count = flattened.len() / self.dim;
-            let row = self.place(count)?;
-            rows.resize(rows.len() + (row - self.next_row) * self.dim, T::default());
            rows.extend_from_slice(&flattened);
            offsets.push(MultivectorMmapOffset {
-                offset: row as PointOffsetType,
+                offset: self.next_row as PointOffsetType,
                count: count as PointOffsetType,
                capacity: count as PointOffsetType,
            });
-            self.next_row = row + count;
+            self.next_row += count;
        }

        self.vectors.append_many(
@@ -124,22 +117,6 @@ impl<T: PrimitiveVectorElement, S: UniversalAppend + 'static>
        self.deleted.flush(hw_counter)
    }

-    /// Where a run of `count` rows goes: at the end, or at the start of the next
-    /// chunk when it would otherwise straddle a chunk boundary.
-    fn place(&self, count: usize) -> OperationResult<usize> {
-        let remaining = self.vectors.remaining_chunk_keys(self.next_row);
-        if count > remaining {
-            let max = self.vectors.remaining_chunk_keys(0);
-            if count > max {
-                return Err(OperationError::service_error(format!(
-                    "Cannot insert a multi vector of {count} inner vectors, a chunk holds {max}",
-                )));
-            }
-            return Ok(self.next_row + remaining);
-        }
-        Ok(self.next_row)
-    }
-
    fn flatten_decoded(&self, vector: VectorRef<'_>) -> OperationResult<Vec<T>> {
        let multi = TypedMultiDenseVectorRef::<VectorElementType>::try_from(vector)?;
        if multi.dim != self.dim {
@@ -10,12 +10,12 @@ use common::types::PointOffsetType;
 use common::universal_io::UserData;

 use crate::common::Flusher;
-use crate::common::operation_error::{OperationError, OperationResult, check_process_stopped};
+use crate::common::operation_error::{OperationResult, check_process_stopped};
 use crate::data_types::named_vectors::{CowMultiVector, CowVector};
 use crate::data_types::primitive::PrimitiveVectorElement;
 use crate::data_types::vectors::{TypedMultiDenseVectorRef, VectorElementType, VectorRef};
 use crate::types::{Distance, MultiVectorConfig, VectorStorageDatatype};
-use crate::vector_storage::common::CHUNK_SIZE;
+use crate::vector_storage::multi_dense::appendable_mmap_multi_dense_vector_storage::flattened_to_multi_vector;
 use crate::vector_storage::volatile_chunked_vectors::VolatileChunkedVectors;
 use crate::vector_storage::{
    MultiVectorStorage, MultiVectorStorageRead, VectorOffsetType, VectorStorage, VectorStorageEnum,
@@ -156,12 +156,6 @@ impl<T: PrimitiveVectorElement> VolatileMultiDenseVectorStorage<T> {
        _hw_counter: &HardwareCounterCell,
    ) -> OperationResult<()> {
        assert_eq!(multi_vector.dim, self.dim);
-        let multivector_size_in_bytes = std::mem::size_of_val(multi_vector.flattened_vectors);
-        if multivector_size_in_bytes >= CHUNK_SIZE {
-            return Err(OperationError::service_error(format!(
-                "Cannot insert multi vector of size {multivector_size_in_bytes} to the vector storage. It's too large, maximum size is {CHUNK_SIZE}.",
-            )));
-        }

        let key_usize = key as usize;
        if key_usize >= self.vectors_metadata.len() {
@@ -173,37 +167,29 @@ impl<T: PrimitiveVectorElement> VolatileMultiDenseVectorStorage<T> {
        metadata.inner_vectors_count = multi_vector.vectors_count();

        if multi_vector.vectors_count() > metadata.inner_vector_capacity {
+            // Does not fit its old place, so it is appended to the end
            metadata.inner_vector_capacity = metadata.inner_vectors_count;
            metadata.start = self.vectors.len();
-            let left_keys = self.vectors.get_chunk_left_keys(metadata.start);
-            if multi_vector.vectors_count() > left_keys {
-                metadata.start += left_keys;
-            }
-            self.vectors.insert_many(
-                metadata.start,
-                multi_vector.flattened_vectors,
-                multi_vector.vectors_count(),
-            )?;
-        } else {
-            self.vectors.insert_many(
-                metadata.start,
-                multi_vector.flattened_vectors,
-                multi_vector.vectors_count(),
-            )?;
        }

+        self.vectors.insert_many(
+            metadata.start,
+            multi_vector.flattened_vectors,
+            multi_vector.vectors_count(),
+        )?;
+
        self.set_deleted(key, is_deleted);
        Ok(())
    }

-    fn get_multi_impl(&self, key: PointOffsetType) -> Option<TypedMultiDenseVectorRef<'_, T>> {
+    fn get_multi_impl(&self, key: PointOffsetType) -> Option<CowMultiVector<'_, T>> {
        let &MultiVectorMetadata {
            start,
            inner_vectors_count,
            ..
        } = self.vectors_metadata.get(key as usize)?;
        let vectors = self.vectors.get_many(start, inner_vectors_count)?;
-        Some(TypedMultiDenseVectorRef::new(vectors, self.dim))
+        Some(flattened_to_multi_vector(vectors, self.dim))
    }
 }

@@ -223,7 +209,7 @@ impl<T: PrimitiveVectorElement> MultiVectorStorageRead<T> for VolatileMultiDense
        key: PointOffsetType,
    ) -> Option<CowMultiVector<'_, T>> {
        // No sequential optimizations available for in memory storage.
-        self.get_multi_impl(key).map(CowMultiVector::Borrowed)
+        self.get_multi_impl(key)
    }

    fn for_each_in_batch_multi<F>(&self, keys: &[PointOffsetType], mut callback: F)
@@ -232,7 +218,7 @@ impl<T: PrimitiveVectorElement> MultiVectorStorageRead<T> for VolatileMultiDense
    {
        for (idx, &key) in keys.iter().enumerate() {
            let vector = self.get_multi_impl(key).expect("multi vector exists");
-            callback(idx, vector);
+            callback(idx, vector.as_ref());
        }
    }

@@ -57,9 +57,9 @@ impl<S: UniversalRead> QuantizedChunkedStorageRead<S> {
        self.data.clear_cache()
    }

-    /// Concatenated records `[index, index + count)` in one contiguous read.
-    /// `None` if out of bounds or the range straddles a chunk boundary
-    /// (read-only counterpart of [`super::QuantizedChunkedStorage::get_many`]).
+    /// Concatenated records `[index, index + count)`, copied when the range
+    /// straddles a chunk boundary. `None` if out of bounds (read-only
+    /// counterpart of [`super::QuantizedChunkedStorage::get_many`]).
    pub fn get_many<P: AccessPattern>(
        &self,
        index: PointOffsetType,
@@ -70,8 +70,7 @@ impl<S: UniversalRead> QuantizedChunkedStorageRead<S> {

    /// Batched counterpart of [`Self::get_many`]: invoke `callback` with the
    /// concatenated records of each `(user_data, start, count)` range, batching
-    /// the underlying reads. Like [`Self::get_many`], a range must not straddle
-    /// a chunk boundary.
+    /// the underlying reads.
    pub fn for_each_many<P: AccessPattern, U: UserData>(
        &self,
        ranges: impl Iterator<Item = (U, PointOffsetType, u32)>,
@@ -50,15 +50,9 @@ impl<S: UniversalWrite + Send + 'static> QuantizedChunkedStorage<S> {
        data.clear_cache()
    }

-    /// Record slots left in the chunk containing `start`.
-    pub fn get_remaining_chunk_keys(&self, start: PointOffsetType) -> usize {
-        self.data
-            .get_remaining_chunk_keys(start as VectorOffsetType)
-    }
-
    /// Returns multiple continuous vectors given a start `index` and a `count` of vectors to return.
    ///
-    /// This function returns `None` if the vector is out of bounds or is located across multiple chunks.
+    /// This function returns `None` if the vector is out of bounds.
    pub fn get_many<P>(&self, index: PointOffsetType, count: usize) -> Option<Cow<'_, [u8]>>
    where
        P: AccessPattern,
@@ -68,8 +62,7 @@ impl<S: UniversalWrite + Send + 'static> QuantizedChunkedStorage<S> {

    /// Batched counterpart of [`Self::get_many`]: invoke `callback` with the
    /// concatenated records of each `(user_data, start, count)` range, batching
-    /// the underlying reads. Like [`Self::get_many`], a range must not
-    /// straddle a chunk boundary.
+    /// the underlying reads.
    pub fn for_each_many<P, U>(
        &self,
        ranges: impl Iterator<Item = (U, PointOffsetType, u32)>,
@@ -4,7 +4,6 @@ use common::counter::hardware_counter::HardwareCounterCell;
 use common::generic_consts::Random;
 use common::mmap::AdviceSetting;
 use common::types::PointOffsetType;
-use common::validation::MAX_MULTIVECTOR_FLATTENED_LEN;
 use rstest::rstest;
 use tempfile::Builder;

@@ -351,29 +350,29 @@ fn test_update_from_delete_points_multi_dense_vector_storage(
    }
 }

+/// An inner vector nearly fills a chunk, so the multivector spans several.
 #[rstest]
 #[case(MultiDenseStorageType::AppendableMmapFloat)]
 fn test_large_multi_dense_vector_storage(#[case] storage_type: MultiDenseStorageType) {
-    assert!(MAX_MULTIVECTOR_FLATTENED_LEN * std::mem::size_of::<VectorElementType>() < CHUNK_SIZE);
-
    let vec_dim = 100_000;
-    let vec_count = 100;
+    let vec_count = 10;
+    assert!(vec_dim * vec_count * std::mem::size_of::<VectorElementType>() > CHUNK_SIZE);
+
    let dir = Builder::new().prefix("storage_dir").tempdir().unwrap();
    let mut storage = create_vector_storage(storage_type, vec_dim, dir.path());

-    let vectors = vec![vec![0.0; vec_dim]; vec_count];
+    let vectors: Vec<Vec<VectorElementType>> = (0..vec_count)
+        .map(|i| vec![i as VectorElementType; vec_dim])
+        .collect();
    let multivec = MultiDenseVectorInternal::try_from(vectors).unwrap();

    let hw_counter = HardwareCounterCell::new();
-    let result = storage.insert_vector(0, VectorRef::from(&multivec), &hw_counter);
-    match result {
-        Ok(_) => {
-            panic!("Inserting vector should fail");
-        }
-        Err(e) => {
-            assert!(e.to_string().contains("too large"));
-        }
-    }
+    storage
+        .insert_vector(0, VectorRef::from(&multivec), &hw_counter)
+        .unwrap();
+
+    let stored = storage.get_vector::<Random>(0);
+    assert_eq!(stored.as_vec_ref(), VectorRef::from(&multivec));
 }

 #[test]
@@ -418,29 +417,29 @@ fn test_update_from_delete_points_volatile_multi_dense_vector_storage() {
    }
 }

+/// An inner vector nearly fills a chunk, so the multivector spans several.
 #[test]
 fn test_large_volatile_multi_dense_vector_storage() {
-    assert!(MAX_MULTIVECTOR_FLATTENED_LEN * std::mem::size_of::<VectorElementType>() < CHUNK_SIZE);
-
    let vec_dim = 100_000;
-    let vec_count = 100;
+    let vec_count = 10;
+    assert!(vec_dim * vec_count * std::mem::size_of::<VectorElementType>() > CHUNK_SIZE);
+
    let mut storage = new_volatile_multi_dense_vector_storage(
        vec_dim,
        Distance::Dot,
        MultiVectorConfig::default(),
    );

-    let vectors = vec![vec![0.0; vec_dim]; vec_count];
+    let vectors: Vec<Vec<VectorElementType>> = (0..vec_count)
+        .map(|i| vec![i as VectorElementType; vec_dim])
+        .collect();
    let multivec = MultiDenseVectorInternal::try_from(vectors).unwrap();

    let hw_counter = HardwareCounterCell::new();
-    let result = storage.insert_vector(0, VectorRef::from(&multivec), &hw_counter);
-    match result {
-        Ok(_) => {
-            panic!("Inserting vector should fail");
-        }
-        Err(e) => {
-            assert!(e.to_string().contains("too large"));
-        }
-    }
+    storage
+        .insert_vector(0, VectorRef::from(&multivec), &hw_counter)
+        .unwrap();
+
+    let stored = storage.get_vector::<Random>(0);
+    assert_eq!(stored.as_vec_ref(), VectorRef::from(&multivec));
 }
@@ -158,14 +158,6 @@ pub fn open_appendable_turbo_multi_vector_storage(
    })
 }

-/// Rejection message for a multivector that cannot fit a whole storage chunk
-/// ([`AppendableMmapMultiTurboVectorStorage::fresh_range_start`] returning `None`). Single
-/// source of truth for both write paths; only the error class differs by call
-/// site (user error on ingest, service error on the optimizer merge).
-fn exceeds_chunk_capacity_message(count: PointOffsetType) -> String {
-    format!("Multivector of {count} subvectors exceeds the chunk capacity")
-}
-
 impl AppendableMmapMultiTurboVectorStorage {
    /// Offset record for `key`, if the point exists.
    fn get_offset<P: AccessPattern>(&self, key: PointOffsetType) -> Option<MultivectorMmapOffset> {
@@ -247,7 +239,6 @@ impl AppendableMmapMultiTurboVectorStorage {
        let records = self
            .storage
            .get_many::<Random>(offset.offset, offset.count as usize)
-            // SAFETY: `fresh_range_start` guarantees ranges never straddle across a boundary.
            .expect("Multivector not found");
        self.dequantize_records(&records)
    }
@@ -293,24 +284,6 @@ impl AppendableMmapMultiTurboVectorStorage {
            })
    }

-    /// Start of a fresh range for `count` records, never straddling a chunk
-    /// boundary: skips the tail when the range wouldn't fit it, `None` when
-    /// even a whole chunk can't hold it. Callers classify the `None` case via
-    /// [`exceeds_chunk_capacity_message`]: user error on the ingest path,
-    /// service error on the optimizer merge.
-    fn fresh_range_start(&self, count: PointOffsetType) -> Option<PointOffsetType> {
-        let start = self.storage.vectors_count() as PointOffsetType;
-        let left = self.storage.get_remaining_chunk_keys(start);
-        if count as usize <= left {
-            return Some(start);
-        }
-        let next_chunk = start + left as PointOffsetType;
-        if count as usize > self.storage.get_remaining_chunk_keys(next_chunk) {
-            return None;
-        }
-        Some(next_chunk)
-    }
-
    /// Record range for upserting `count` records at `key`: reuse the existing
    /// range in place when the new count fits its capacity, else allocate a
    /// fresh range at the end.
@@ -318,7 +291,7 @@ impl AppendableMmapMultiTurboVectorStorage {
        &self,
        key: PointOffsetType,
        count: PointOffsetType,
-    ) -> OperationResult<MultivectorMmapOffset> {
+    ) -> MultivectorMmapOffset {
        let mut offset = self
            .offsets
            .get::<Random>(key as VectorOffsetType)
@@ -326,20 +299,15 @@ impl AppendableMmapMultiTurboVectorStorage {
            .unwrap_or_default();

        if count > offset.capacity {
-            // User error so WAL replay skips the op instead of crash-looping.
-            // Reachable only internally (no `MAX_MULTIVECTOR_FLATTENED_LEN` check).
-            let fresh_start = self.fresh_range_start(count).ok_or_else(|| {
-                OperationError::malformed_vector_blob(exceeds_chunk_capacity_message(count))
-            })?;
            offset = MultivectorMmapOffset {
-                offset: fresh_start,
+                offset: self.storage.vectors_count() as PointOffsetType,
                count,
                capacity: count,
            };
        } else {
            offset.count = count;
        }
-        Ok(offset)
+        offset
    }

    /// Encode and upsert one multivector at `key`.
@@ -352,7 +320,7 @@ impl AppendableMmapMultiTurboVectorStorage {
        assert_eq!(multi_vector.dim, self.dim);

        let count = multi_vector.vectors_count() as PointOffsetType;
-        let offset = self.record_range_for_upsert(key, count)?;
+        let offset = self.record_range_for_upsert(key, count);

        for (i, inner) in multi_vector
            .flattened_vectors
@@ -394,7 +362,7 @@ impl AppendableMmapMultiTurboVectorStorage {
        }

        let count = (bytes.len() / record_size) as PointOffsetType;
-        let offset = self.record_range_for_upsert(key, count)?;
+        let offset = self.record_range_for_upsert(key, count);

        for (i, encoded) in bytes.chunks_exact(record_size).enumerate() {
            self.storage.upsert_vector(
@@ -731,8 +699,6 @@ impl MultiTQVectorStorageRead for AppendableMmapMultiTurboVectorStorage {
                self.storage
                    .get_many::<P>(offset.offset, offset.count as usize)
            })
-            // `get_many` is also `None` for a range across multiple chunks, but
-            // `fresh_range_start` guarantees ranges never straddle a boundary.
            .expect("Multivector not found")
    }
 }
@@ -758,11 +724,7 @@ impl MultiTQVectorStorage for AppendableMmapMultiTurboVectorStorage {

            let count = (blob.len() / record_size) as u32;
            let key = self.offsets.len() as PointOffsetType;
-            // Optimizer merge of already-stored data: over-capacity here is
-            // genuine corruption, kept as a service error.
-            let inner_start = self.fresh_range_start(count).ok_or_else(|| {
-                OperationError::service_error(exceeds_chunk_capacity_message(count))
-            })?;
+            let inner_start = self.storage.vectors_count() as PointOffsetType;
            for (i, record) in blob.chunks_exact(record_size).enumerate() {
                self.storage.upsert_vector(
                    inner_start + i as PointOffsetType,
@@ -1286,11 +1248,10 @@ mod tests {
        assert_blob(&storage, &grown);
    }

-    /// Both write paths skip the chunk tail when a fresh range would straddle
-    /// the boundary (mirroring the multi-dense storage), so a multivector never
-    /// spans two chunks and reads stay borrowable.
+    /// Both write paths place fresh ranges at the end of the record space, so a
+    /// multivector may span two chunks; those reads copy instead of borrowing.
    #[test]
-    fn fresh_ranges_skip_chunk_tail_so_reads_borrow() {
+    fn fresh_ranges_straddle_chunks_and_reads_stitch() {
        const DIM: usize = 128;
        let distance = Distance::Dot;
        let dir = Builder::new()
@@ -1320,8 +1281,8 @@ mod tests {
        storage.update_from(&mut it, &stopped).unwrap();
        assert_eq!(storage.storage.vectors_count(), records_per_chunk - 1);

-        // count=3 does not fit the chunk's single remaining slot: the tail is
-        // skipped and the range starts at the next chunk boundary.
+        // count=3 does not fit the chunk's single remaining slot, so the run
+        // continues into the next chunk. No slot is skipped.
        let multi = multi_of(DIM, 3, 7);
        storage
            .insert_vector(
@@ -1331,48 +1292,44 @@ mod tests {
            )
            .unwrap();
        let offset = storage.get_offset::<Random>(1).unwrap();
-        assert_eq!(offset.offset as usize, records_per_chunk);
+        assert_eq!(offset.offset as usize, records_per_chunk - 1);
        assert_eq!(offset.count, 3);
-        // One padding slot at the chunk tail, never referenced by any offset.
-        assert_eq!(storage.storage.vectors_count(), records_per_chunk + 3);
+        assert_eq!(storage.storage.vectors_count(), records_per_chunk + 2);

-        // Within one chunk the getter borrows, and the bytes are exact.
+        // Straddling reads copy, and the bytes are exact.
        let blob = storage.get_multi_tq::<Random>(1);
-        assert!(matches!(blob, Cow::Borrowed(_)));
+        assert!(matches!(blob, Cow::Owned(_)));
        assert_eq!(
            blob.as_ref(),
            oracle.encode_multi(&multi).concat().as_slice(),
        );

-        // `update_from` skips too: pad to one slot before the second chunk
-        // boundary, then append a 3-record point that lands at the boundary.
-        let pad = records_per_chunk - 4;
-        let tail_skipped = multi_of(DIM, 3, 8);
+        // Same for `update_from`: pad to one slot before the second boundary,
+        // then append a 3-record point that crosses it.
+        let pad = records_per_chunk - 3;
+        let straddling = multi_of(DIM, 3, 8);
        let blobs = [
            vec![0u8; pad * record_size],
-            oracle.encode_multi(&tail_skipped).concat(),
+            oracle.encode_multi(&straddling).concat(),
        ];
        let mut it = blobs.iter().map(|b| (Cow::from(b.as_slice()), false));
        storage.update_from(&mut it, &stopped).unwrap();
        let offset = storage.get_offset::<Random>(3).unwrap();
-        assert_eq!(offset.offset as usize, 2 * records_per_chunk);
-        // The skipped slot at the second boundary is never referenced.
-        assert_eq!(storage.storage.vectors_count(), 2 * records_per_chunk + 3);
+        assert_eq!(offset.offset as usize, 2 * records_per_chunk - 1);
+        assert_eq!(storage.storage.vectors_count(), 2 * records_per_chunk + 2);

-        // The range stays within one chunk: borrowed read, exact bytes.
        let blob = storage.get_multi_tq::<Random>(3);
-        assert!(matches!(blob, Cow::Borrowed(_)));
+        assert!(matches!(blob, Cow::Owned(_)));
        assert_eq!(
            blob.as_ref(),
-            oracle.encode_multi(&tail_skipped).concat().as_slice(),
+            oracle.encode_multi(&straddling).concat().as_slice(),
        );
    }

-    /// The largest multivector that fits one chunk is accepted; one subvector
-    /// more is rejected at write time on both write paths (`fresh_range_start`).
-    /// Cheap thanks to the small 512 KiB test-build `CHUNK_SIZE`.
+    /// A multivector larger than a whole chunk is stored and read back on both
+    /// write paths. Cheap thanks to the small 512 KiB test-build `CHUNK_SIZE`.
    #[test]
-    fn chunk_sized_multivector_accepted_one_larger_rejected() {
+    fn multivector_larger_than_a_chunk_round_trips() {
        const DIM: usize = 128;
        let distance = Distance::Dot;
        let dir = Builder::new()
@@ -1392,38 +1349,31 @@ mod tests {
        .unwrap();
        let record_size = storage.quantized_vector_size();
        let records_per_chunk = CHUNK_SIZE / record_size;
+        let oracle = Oracle::new(DIM, distance);

-        // Exactly one whole chunk of records is the maximum allowed.
-        let max_blob = vec![0u8; records_per_chunk * record_size];
-        let mut it = std::iter::once((Cow::from(max_blob.as_slice()), false));
+        // One record past a whole chunk, so the range covers two of them
+        let blob = vec![0u8; (records_per_chunk + 1) * record_size];
+        let mut it = std::iter::once((Cow::from(blob.as_slice()), false));
        assert_eq!(storage.update_from(&mut it, &stopped).unwrap(), 0..1);
-        let blob = storage.get_multi_tq::<Random>(0);
-        assert!(matches!(blob, Cow::Borrowed(_)));
-        assert_eq!(blob.as_ref(), max_blob.as_slice());

-        // One subvector more cannot fit any chunk: rejected via `update_from`...
-        let oversized_blob = vec![0u8; (records_per_chunk + 1) * record_size];
-        let mut it = std::iter::once((Cow::from(oversized_blob.as_slice()), false));
-        assert!(storage.update_from(&mut it, &stopped).is_err());
+        let read = storage.get_multi_tq::<Random>(0);
+        assert!(matches!(read, Cow::Owned(_)), "straddling read must copy");
+        assert_eq!(read.as_ref(), blob.as_slice());

-        // ...and via `insert_vector`, before any record is written.
-        let oversized = multi_of(DIM, records_per_chunk + 1, 11);
-        assert!(
-            storage
-                .insert_vector(
-                    1,
-                    TypedMultiDenseVectorRef::from(&oversized).into(),
-                    &hw_counter,
-                )
-                .is_err()
-        );
+        // Same through `insert_vector`
+        let multi = multi_of(DIM, records_per_chunk + 1, 11);
+        storage
+            .insert_vector(
+                1,
+                TypedMultiDenseVectorRef::from(&multi).into(),
+                &hw_counter,
+            )
+            .unwrap();

-        // The rejected points left no trace; the accepted one is intact.
-        assert_eq!(storage.total_vector_count(), 1);
-        assert_eq!(storage.storage.vectors_count(), records_per_chunk);
+        assert_eq!(storage.total_vector_count(), 2);
        assert_eq!(
-            storage.get_multi_tq::<Random>(0).as_ref(),
-            max_blob.as_slice()
+            storage.get_multi_tq::<Random>(1).as_ref(),
+            oracle.encode_multi(&multi).concat().as_slice(),
        );
    }

@@ -1663,12 +1613,6 @@ mod tests {

            let expected_blob = slot.encoded.concat();
            let blob = storage.get_multi_tq::<Random>(key);
-            // Ranges never straddle a chunk (`fresh_range_start`), so blob
-            // reads always borrow.
-            assert!(
-                matches!(blob, Cow::Borrowed(_)),
-                "{ctx}: blob not borrowed at {i}"
-            );
            assert_eq!(
                blob.as_ref(),
                expected_blob.as_slice(),
@@ -73,9 +73,6 @@ impl<S: UniversalAppend + 'static> UpdateOnlyMultiTurboVectorStorage<S> {
        hw_counter: &HardwareCounterCell,
    ) -> OperationResult<()> {
        let encoded_size = self.quantizer.quantized_size();
-        // As in the plain multivector storage, a run that would straddle a
-        // chunk skips to the next one; the gaps become explicit zero rows, so
-        // a single append lands every run where its offset entry points.
        let batch_start = self.next_row;
        let mut rows: Vec<u8> = Vec::new();
        let mut offsets = Vec::new();
@@ -92,15 +89,13 @@ impl<S: UniversalAppend + 'static> UpdateOnlyMultiTurboVectorStorage<S> {
            };

            let count = encoded.len() / encoded_size;
-            let row = self.place(count)?;
-            rows.resize(rows.len() + (row - self.next_row) * encoded_size, 0);
            rows.extend_from_slice(&encoded);
            offsets.push(MultivectorMmapOffset {
-                offset: row as PointOffsetType,
+                offset: self.next_row as PointOffsetType,
                count: count as PointOffsetType,
                capacity: count as PointOffsetType,
            });
-            self.next_row = row + count;
+            self.next_row += count;
        }

        self.vectors.append_many(
@@ -122,22 +117,6 @@ impl<S: UniversalAppend + 'static> UpdateOnlyMultiTurboVectorStorage<S> {
        self.deleted.flush(hw_counter)
    }

-    /// Where a run of `count` rows goes: at the end, or at the start of the next
-    /// chunk when it would otherwise straddle a chunk boundary.
-    fn place(&self, count: usize) -> OperationResult<usize> {
-        let remaining = self.vectors.remaining_chunk_keys(self.next_row);
-        if count > remaining {
-            let max = self.vectors.remaining_chunk_keys(0);
-            if count > max {
-                return Err(OperationError::service_error(format!(
-                    "Cannot insert a multi vector of {count} inner vectors, a chunk holds {max}",
-                )));
-            }
-            return Ok(self.next_row + remaining);
-        }
-        Ok(self.next_row)
-    }
-
    /// Encode every inner vector, back to back.
    fn encode_decoded(&mut self, vector: VectorRef<'_>) -> OperationResult<Vec<u8>> {
        let multi = TypedMultiDenseVectorRef::<VectorElementType>::try_from(vector)?;
@@ -1,3 +1,4 @@
+use std::borrow::Cow;
 use std::cmp::max;
 use std::collections::TryReserveError;
 use std::mem;
@@ -87,17 +88,33 @@ impl<T: Copy + Clone + Default> VolatileChunkedVectors<T> {
            })
    }

-    pub fn get_many(&self, key: VectorOffsetType, count: usize) -> Option<&[T]> {
-        if self.chunks.is_empty() {
-            return None;
+    /// Borrows a single chunk, or copies when the run straddles a boundary.
+    pub fn get_many(&self, key: VectorOffsetType, count: usize) -> Option<Cow<'_, [T]>> {
+        let mut key = key;
+        let mut left = count;
+        let mut vectors: Option<Cow<'_, [T]>> = None;
+
+        loop {
+            let chunk_data = self.chunks.get(key / self.chunk_capacity)?;
+            let idx = (key % self.chunk_capacity) * self.dim;
+            let part_count = left.min(self.chunk_left_keys(key));
+            let part = chunk_data.get(idx..idx + part_count * self.dim)?;
+
+            vectors = Some(match vectors {
+                None => Cow::Borrowed(part),
+                Some(mut vectors) => {
+                    vectors.to_mut().extend_from_slice(part);
+                    vectors
+                }
+            });
+
+            key += part_count;
+            left -= part_count;
+
+            if left == 0 {
+                return vectors;
+            }
        }
-        self.chunks
-            .get(key / self.chunk_capacity)
-            .and_then(|chunk_data| {
-                let idx = (key % self.chunk_capacity) * self.dim;
-                let range = idx..idx + count * self.dim;
-                chunk_data.get(range)
-            })
    }

    pub fn push(&mut self, vector: &[T]) -> Result<VectorOffsetType, TryReserveError> {
@@ -106,8 +123,8 @@ impl<T: Copy + Clone + Default> VolatileChunkedVectors<T> {
        Ok(new_id)
    }

-    // returns how many flattened vectors can be inserted starting from key
-    pub fn get_chunk_left_keys(&self, start_key: VectorOffsetType) -> usize {
+    /// How many vectors still fit in the chunk holding `start_key`.
+    fn chunk_left_keys(&self, start_key: VectorOffsetType) -> usize {
        self.chunk_capacity - (start_key % self.chunk_capacity)
    }

@@ -127,10 +144,6 @@ impl<T: Copy + Clone + Default> VolatileChunkedVectors<T> {
            vectors_count * self.dim,
            "Vector size mismatch"
        );
-        assert!(
-            self.get_chunk_left_keys(key) >= vectors_count,
-            "Index out of bounds"
-        );

        let desired_capacity = self.chunk_capacity * self.dim;
        let new_len = max(self.len, key + vectors_count);
@@ -162,30 +175,40 @@ impl<T: Copy + Clone + Default> VolatileChunkedVectors<T> {
            assert_eq!(self.chunks.len(), chunks_len);
        }

-        let chunk_idx = key / self.chunk_capacity;
-        let chunk_data = &mut self.chunks[chunk_idx];
-        let idx = (key % self.chunk_capacity) * self.dim;
+        // A run longer than the chunk's tail continues in the next one
+        let mut key = key;
+        let mut rest = vectors;
+        while !rest.is_empty() {
+            let chunk_idx = key / self.chunk_capacity;
+            let idx = (key % self.chunk_capacity) * self.dim;
+            let fits = self.chunk_left_keys(key) * self.dim;
+            let (part, tail) = rest.split_at(fits.min(rest.len()));

-        // Grow the current chunk if needed to fit the new vector.
-        //
-        // All chunks are dynamically resized to fit their vectors in it.
-        // Chunks have a size of zero by default. It's grown with zeroes to fit new vectors.
-        //
-        // The capacity for the first chunk is allocated normally to keep the memory footprint as
-        // small as possible, see
-        // <https://doc.rust-lang.org/std/vec/struct.Vec.html#capacity-and-reallocation>).
-        // All other chunks allocate their capacity in full on first use to prevent expensive
-        // reallocations when their data grows.
-        if chunk_data.len() < idx + vectors.len() {
-            // If the chunk is not the first one, allocate it fully on first use
-            if chunk_idx != 0 {
-                chunk_data.try_set_capacity_exact(desired_capacity)?;
+            let chunk_data = &mut self.chunks[chunk_idx];
+
+            // Grow the current chunk if needed to fit the new vector.
+            //
+            // All chunks are dynamically resized to fit their vectors in it.
+            // Chunks have a size of zero by default. It's grown with zeroes to fit new vectors.
+            //
+            // The capacity for the first chunk is allocated normally to keep the memory footprint as
+            // small as possible, see
+            // <https://doc.rust-lang.org/std/vec/struct.Vec.html#capacity-and-reallocation>).
+            // All other chunks allocate their capacity in full on first use to prevent expensive
+            // reallocations when their data grows.
+            if chunk_data.len() < idx + part.len() {
+                // If the chunk is not the first one, allocate it fully on first use
+                if chunk_idx != 0 {
+                    chunk_data.try_set_capacity_exact(desired_capacity)?;
+                }
+                chunk_data.resize_with(idx + part.len(), T::default);
            }
-            chunk_data.resize_with(idx + vectors.len(), T::default);
-        }

-        let data = &mut chunk_data[idx..idx + vectors.len()];
-        data.copy_from_slice(vectors);
+            chunk_data[idx..idx + part.len()].copy_from_slice(part);
+
+            key += part.len() / self.dim;
+            rest = tail;
+        }

        // Update `self.len` only after the vector is successfully inserted.
        // In case of OOM, `self.len` will not be updated.
@@ -196,27 +219,14 @@ impl<T: Copy + Clone + Default> VolatileChunkedVectors<T> {

    /// Append all flattened vectors in `vectors` to the end of the storage.
    ///
-    /// `vectors` holds `vectors.len() / dim` consecutive vectors. They are inserted
-    /// in batches of one chunk's remaining capacity, so each batch is a single
-    /// `copy_from_slice` and never crosses a chunk boundary.
+    /// `vectors` holds `vectors.len() / dim` consecutive vectors.
    pub fn extend(&mut self, vectors: &[T]) -> Result<(), TryReserveError> {
        assert!(
            vectors.len().is_multiple_of(self.dim),
            "Vector data size mismatch"
        );

-        let count = vectors.len() / self.dim;
-        let mut inserted = 0;
-        while inserted < count {
-            let key = self.len;
-            let batch = self.get_chunk_left_keys(key).min(count - inserted);
-            let start = inserted * self.dim;
-            let end = start + batch * self.dim;
-            self.insert_many(key, &vectors[start..end], batch)?;
-            inserted += batch;
-        }
-
-        Ok(())
+        self.insert_many(self.len, vectors, vectors.len() / self.dim)
    }
 }