[UpdateOnly] implement UpdateOnlyChunkedVectors (#10114)

* AI + manual: impl `UpdateOnlyChunkedVectors`

* AI: simplify

* graceful handling of unexpected file lengths

fix test

* incorporate updates from #10119

* drop the unused status read on open

The vector count loaded at open was never consulted: every batch carries the
offset it starts at, and the chunks are reconciled against that offset. Drop
the field and the read, and fold both watermark writes into `save_len`.

A corrupt status file no longer blocks opening the writer — the first batch
overwrites it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix clippy: ensure_chunk_lengths no longer needs &mut self

Dropping the status field left it with nothing to mutate. `append_many` keeps
`&mut self` — nothing in this module is exported, so the lint reaches it too,
but the exclusive borrow is what enforces the single-writer contract the
appends rest on.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: generall <andrey@vasnetsov.com>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

lib/segment/src/vector_storage/chunked_vectors/chunks.rs

@@ -3,8 +3,8 @@ use std::path::{Path, PathBuf};
 use ahash::AHashMap;
 use common::mmap::{AdviceSetting, MULTI_MMAP_IS_SUPPORTED, create_and_ensure_length};
 use common::universal_io::{
-    OpenOptions, Populate, TypedStorage, UniversalIoError, UniversalRead, UniversalReadFs,
-    UniversalWrite,
+    ListedFile, OpenOptions, Populate, TypedStorage, UniversalIoError, UniversalRead,
+    UniversalReadFileOps, UniversalReadFs, UniversalWrite,
 };
 
 use super::config::{MMAP_CHUNKS_PATTERN_END, MMAP_CHUNKS_PATTERN_START};
@@ -47,6 +47,26 @@ pub fn read_chunks<T: bytemuck::Pod + Send, S: UniversalRead>(
     read_chunks_from(fs, directory, 0, advice, populate, writeable)
 }
 
+/// List the chunk files under `directory`, keyed by chunk id.
+pub(super) fn list_chunk_files(
+    fs: &impl UniversalReadFileOps,
+    directory: &Path,
+) -> Result<AHashMap<usize, ListedFile>, UniversalIoError> {
+    let mut chunks_files = AHashMap::new();
+    for listed in fs.list_files(&chunks_prefix(directory))? {
+        let chunk_id = listed
+            .path
+            .file_name()
+            .and_then(|file_name| file_name.to_str())
+            .and_then(check_mmap_file_name_pattern);
+
+        if let Some(chunk_id) = chunk_id {
+            chunks_files.insert(chunk_id, listed);
+        }
+    }
+    Ok(chunks_files)
+}
+
 /// Open chunk files with id `>= start_chunk_id`, in ascending order.
 pub fn read_chunks_from<T: bytemuck::Pod + Send, S: UniversalRead>(
     fs: &impl UniversalReadFs<File = S>,
@@ -56,28 +76,19 @@ pub fn read_chunks_from<T: bytemuck::Pod + Send, S: UniversalRead>(
     populate: Populate,
     writeable: bool,
 ) -> Result<Vec<TypedStorage<S, T>>, UniversalIoError> {
-    let mut chunks_files: AHashMap<usize, _> = AHashMap::new();
-    for listed in fs.list_files(&chunks_prefix(directory))? {
-        let path = listed.path;
-        let chunk_id = path
-            .file_name()
-            .and_then(|file_name| file_name.to_str())
-            .and_then(check_mmap_file_name_pattern);
-
-        if let Some(chunk_id) = chunk_id {
-            chunks_files.insert(chunk_id, path);
-        }
-    }
-
+    let mut chunks_files = list_chunk_files(fs, directory)?;
     let num_chunks = chunks_files.len();
     let mut result = Vec::with_capacity(num_chunks.saturating_sub(start_chunk_id));
     for chunk_id in start_chunk_id..num_chunks {
-        let chunk_path = chunks_files.remove(&chunk_id).ok_or_else(|| {
-            UniversalIoError::Io(std::io::Error::new(
-                std::io::ErrorKind::NotFound,
-                format!("Missing chunk {chunk_id} in {}", directory.display(),),
-            ))
-        })?;
+        let chunk_path = chunks_files
+            .remove(&chunk_id)
+            .ok_or_else(|| {
+                UniversalIoError::Io(std::io::Error::new(
+                    std::io::ErrorKind::NotFound,
+                    format!("Missing chunk {chunk_id} in {}", directory.display(),),
+                ))
+            })?
+            .path;
 
         let chunk = TypedStorage::open(
             fs,

lib/segment/src/vector_storage/chunked_vectors/config.rs

@@ -1,9 +1,12 @@
 use std::path::{Path, PathBuf};
 
-use common::universal_io::{UniversalIoError, UniversalReadFs, read_json_via, read_whole_via};
+use common::universal_io::{
+    UniversalIoError, UniversalReadFs, UniversalWriteFileOps, read_json_via, read_whole_via,
+};
 use serde::{Deserialize, Serialize};
 
-use crate::common::operation_error::OperationResult;
+use crate::common::operation_error::{OperationError, OperationResult};
+use crate::vector_storage::common::CHUNK_SIZE;
 
 const CONFIG_FILE_NAME: &str = "config.json";
 const STATUS_FILE_NAME: &str = "status.dat";
@@ -12,8 +15,8 @@ pub(super) const MMAP_CHUNKS_PATTERN_START: &str = "chunk_";
 // TODO: rename for other storages?
 pub(super) const MMAP_CHUNKS_PATTERN_END: &str = ".mmap";
 
-/// Contents of the status file: the number of stored vectors, mapped writable
-/// and updated in place by [`ChunkedVectors`](super::ChunkedVectors).
+/// Contents of the status file: the number of stored vectors, updated in
+/// place by the writers.
 #[derive(Debug, Clone, Copy, bytemuck::Pod, bytemuck::Zeroable)]
 #[repr(C)]
 pub struct Status {
@@ -31,6 +34,80 @@ pub(super) struct ChunkedVectorsConfig {
     pub(super) populate: Option<bool>,
 }
 
+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
+    }
+
+    /// 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
+    }
+}
+
+/// Load the stored config, or create it (and its file) on first open.
+pub(super) fn ensure_config<T, Fs>(
+    fs: &Fs,
+    directory: &Path,
+    dim: usize,
+    populate: bool,
+) -> OperationResult<ChunkedVectorsConfig>
+where
+    Fs: UniversalReadFs + UniversalWriteFileOps,
+{
+    let config_file = config_file(directory);
+    match load_config(fs, &config_file) {
+        Ok(Some(config)) => {
+            if config.dim == dim {
+                Ok(config)
+            } else {
+                Err(OperationError::service_error(format!(
+                    "Wrong configuration in {}: expected {}, found {dim}",
+                    config_file.display(),
+                    config.dim,
+                )))
+            }
+        }
+        Ok(None) => create_config::<T>(fs, &config_file, dim, populate),
+        Err(e) => {
+            log::error!("Failed to deserialize config file {config_file:?}: {e}");
+            create_config::<T>(fs, &config_file, dim, populate)
+        }
+    }
+}
+
+fn create_config<T>(
+    fs: &impl UniversalWriteFileOps,
+    config_file: &Path,
+    dim: usize,
+    populate: bool,
+) -> OperationResult<ChunkedVectorsConfig> {
+    if dim == 0 {
+        return Err(OperationError::service_error(
+            "The vector's dimension cannot be 0",
+        ));
+    }
+
+    let chunk_size_bytes = CHUNK_SIZE;
+    let vector_size_bytes = dim * std::mem::size_of::<T>();
+    let chunk_size_vectors = chunk_size_bytes / vector_size_bytes;
+    let corrected_chunk_size_bytes = chunk_size_vectors * vector_size_bytes;
+
+    let config = ChunkedVectorsConfig {
+        chunk_size_bytes: corrected_chunk_size_bytes,
+        chunk_size_vectors,
+        dim,
+        populate: Some(populate),
+    };
+    fs.atomic_save(config_file, &serde_json::to_vec(&config)?)?;
+    Ok(config)
+}
+
 pub(super) fn config_file(directory: &Path) -> PathBuf {
     directory.join(CONFIG_FILE_NAME)
 }

lib/segment/src/vector_storage/chunked_vectors/lifecycle.rs

@@ -1,18 +1,17 @@
-use std::path::{Path, PathBuf};
+use std::path::Path;
 
-use common::fs::atomic_save_json;
 use common::mmap::AdviceSetting;
 use common::universal_io::{
     OpenOptions, Populate, StoredStruct, UniversalKind, UniversalReadFileOps, UniversalWrite,
+    UniversalWriteFileOps,
 };
 
 use super::ChunkedVectors;
 use super::chunks::read_chunks;
-use super::config::{ChunkedVectorsConfig, Status, config_file, load_config, status_file};
+use super::config::{Status, ensure_config, status_file};
 use super::read_only::ReadOnlyChunkedVectors;
 use crate::common::Flusher;
-use crate::common::operation_error::{OperationError, OperationResult};
-use crate::vector_storage::common::CHUNK_SIZE;
+use crate::common::operation_error::OperationResult;
 
 impl<T, S> ChunkedVectors<T, S>
 where
@@ -23,71 +22,6 @@ where
         S::kind()
     }
 
-    pub fn ensure_status_file(fs: &S::Fs, directory: &Path) -> OperationResult<PathBuf> {
-        let status_file = status_file(directory);
-        if !fs.exists(&status_file)? {
-            {
-                let length = std::mem::size_of::<Status>();
-                // TODO(uio): migrate when UniversalWriteFileOps is available
-                common::mmap::create_and_ensure_length(&status_file, length)?;
-            }
-        }
-        Ok(status_file)
-    }
-
-    fn ensure_config(
-        fs: &S::Fs,
-        directory: &Path,
-        dim: usize,
-        populate: bool,
-    ) -> OperationResult<ChunkedVectorsConfig> {
-        let config_file = config_file(directory);
-        match load_config(fs, &config_file) {
-            Ok(Some(config)) => {
-                if config.dim == dim {
-                    Ok(config)
-                } else {
-                    Err(OperationError::service_error(format!(
-                        "Wrong configuration in {}: expected {}, found {dim}",
-                        config_file.display(),
-                        config.dim,
-                    )))
-                }
-            }
-            Ok(None) => Self::create_config(&config_file, dim, populate),
-            Err(e) => {
-                log::error!("Failed to deserialize config file {config_file:?}: {e}");
-                Self::create_config(&config_file, dim, populate)
-            }
-        }
-    }
-
-    fn create_config(
-        config_file: &Path,
-        dim: usize,
-        populate: bool,
-    ) -> OperationResult<ChunkedVectorsConfig> {
-        if dim == 0 {
-            return Err(OperationError::service_error(
-                "The vector's dimension cannot be 0",
-            ));
-        }
-
-        let chunk_size_bytes = CHUNK_SIZE;
-        let vector_size_bytes = dim * std::mem::size_of::<T>();
-        let chunk_size_vectors = chunk_size_bytes / vector_size_bytes;
-        let corrected_chunk_size_bytes = chunk_size_vectors * vector_size_bytes;
-
-        let config = ChunkedVectorsConfig {
-            chunk_size_bytes: corrected_chunk_size_bytes,
-            chunk_size_vectors,
-            dim,
-            populate: Some(populate),
-        };
-        atomic_save_json(config_file, &config)?;
-        Ok(config)
-    }
-
     pub fn open(
         fs: S::Fs,
         directory: &Path,
@@ -96,7 +30,10 @@ where
         populate: Populate,
     ) -> OperationResult<Self> {
         fs_err::create_dir_all(directory)?;
-        let status_path = Self::ensure_status_file(&fs, directory)?;
+        let status_path = status_file(directory);
+        if !fs.exists(&status_path)? {
+            fs.create(&status_path, size_of::<Status>())?;
+        }
 
         let status: StoredStruct<S, Status> = StoredStruct::open(
             &fs,
@@ -110,7 +47,7 @@ where
             Default::default(),
         )?;
 
-        let config = Self::ensure_config(&fs, directory, dim, populate.to_bool::<S>())?;
+        let config = ensure_config::<T, _>(&fs, directory, dim, populate.to_bool::<S>())?;
         let chunks = read_chunks(&fs, directory, advice, populate, true)?;
         let inner = ReadOnlyChunkedVectors {
             config,

lib/segment/src/vector_storage/chunked_vectors/mod.rs

@@ -1,25 +1,29 @@
-//! Fixed-dimension vectors stored flattened across a directory of
-//! equally-sized chunk files.
+//! 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 preallocated to the configured chunk size. A
-//! vector never straddles a chunk boundary.
+//! and `chunk_<n>.mmap` files. A vector never straddles a chunk boundary.
 //!
-//! Two types share that layout:
+//! Three types share that layout:
 //!
-//! - [`ChunkedVectors`] — the writable storage. Its impls are split across
-//!   [`lifecycle`] (open, config creation, flush) and [`write_ops`] (insert,
-//!   push).
-//! - [`read_only::ReadOnlyChunkedVectors`] — the read-only view, which
-//!   [`ChunkedVectors`] wraps and derefs to for every read.
+//! - [`ChunkedVectors`] — the writable storage, which preallocates chunks to
+//!   the configured size and writes in place. Its impls are split across
+//!   [`lifecycle`] (open, flush) and [`write_ops`] (insert, push).
+//! - [`update_only::UpdateOnlyChunkedVectors`] — a short-lived batch writer
+//!   that grows chunks by appends instead, so chunk files end at the data.
+//! - [`read_only::ReadOnlyChunkedVectors`] — the read-only view over either
+//!   writer's output, which [`ChunkedVectors`] wraps and derefs to for every
+//!   read.
 //!
-//! [`chunks`] and [`config`] hold what both sides need: the chunk files and
+//! [`chunks`] and [`config`] hold what all sides need: the chunk files and
 //! the on-disk metadata files respectively.
 
 mod chunks;
 mod config;
 mod lifecycle;
 pub mod read_only;
+#[cfg(test)]
+mod test_utils;
+pub mod update_only;
 mod write_ops;
 
 use std::ops::Deref;

lib/segment/src/vector_storage/chunked_vectors/read_only/mod.rs

@@ -53,14 +53,10 @@ mod tests {
     use tempfile::Builder;
 
     use super::super::chunks::chunk_name;
+    use super::super::test_utils::{append_range, make_vec};
+    use super::super::update_only::UpdateOnlyChunkedVectors;
     use super::*;
     use crate::common::live_reload::LiveReload;
-    use crate::vector_storage::VectorOffsetType;
-    use crate::vector_storage::chunked_vectors::ChunkedVectors;
-
-    fn make_vec(seed: usize, dim: usize) -> Vec<f32> {
-        (0..dim).map(|i| (seed * dim + i) as f32).collect()
-    }
 
     /// A read-only view picks up writer-appended vectors after `live_reload`.
     #[test]
@@ -69,21 +65,9 @@ mod tests {
         let dir = Builder::new().prefix("chunked_reload").tempdir().unwrap();
         let hw = HardwareCounterCell::disposable();
 
-        let first: Vec<Vec<f32>> = (0..100).map(|s| make_vec(s, DIM)).collect();
-        let second: Vec<Vec<f32>> = (100..250).map(|s| make_vec(s, DIM)).collect();
-
-        let mut writer = ChunkedVectors::<f32, MmapFile>::open(
-            MmapFs,
-            dir.path(),
-            DIM,
-            AdviceSetting::Global,
-            Populate::No,
-        )
-        .unwrap();
-        for vector in &first {
-            writer.push(vector.as_slice(), &hw).unwrap();
-        }
-        writer.flusher()().unwrap();
+        let mut writer =
+            UpdateOnlyChunkedVectors::<f32, MmapFile>::open(MmapFs, dir.path(), DIM).unwrap();
+        append_range(&mut writer, 0, 0..100, DIM, &hw);
 
         let mut reader = ReadOnlyChunkedVectors::<f32, MmapFile>::open(
             &MmapFs,
@@ -93,30 +77,23 @@ mod tests {
             Populate::No,
         )
         .unwrap();
-        assert_eq!(reader.len(), first.len());
+        assert_eq!(reader.len(), 100);
 
         // Append more through the writer, then reload the read-only view.
-        for vector in &second {
-            writer.push(vector.as_slice(), &hw).unwrap();
-        }
-        writer.flusher()().unwrap();
+        append_range(&mut writer, 100, 100..250, DIM, &hw);
 
         let empty = SortedSlice::new(&[]).unwrap();
         reader.live_reload(&MmapFs, &empty, &empty, &hw).unwrap();
 
-        assert_eq!(reader.len(), first.len() + second.len());
-        let got = reader
-            .get::<Random>(first.len() as VectorOffsetType)
-            .unwrap();
-        assert_eq!(got.as_ref(), second[0].as_slice());
+        assert_eq!(reader.len(), 250);
+        let got = reader.get::<Random>(100).unwrap();
+        assert_eq!(got.as_ref(), make_vec(100, DIM).as_slice());
     }
 
-    /// Case-5 regression of the live-reload staleness audit: chunk files are
-    /// preallocated to full size, so appended vectors are in-place writes
-    /// within the existing file length. A reader over a caching backend that
-    /// fetched a block straddling the old tail (any read near the tail pulls
-    /// a 16KiB block extending into then-unwritten space) would keep serving
-    /// those stale bytes for vectors appended later into that block —
+    /// Case-5 regression of the live-reload staleness audit: a reader over a
+    /// caching backend that fetched a block straddling the old tail (any read
+    /// near the tail pulls a 16KiB block covering space appended into later)
+    /// would keep serving those stale bytes for vectors landing in that block —
     /// `live_reload` must re-open the last held chunk, not keep the handle.
     /// This drives it over `DiskCacheFs`, where the failure actually
     /// reproduces (mmap readers are read-through and can't catch it).
@@ -140,18 +117,9 @@ mod tests {
 
         // The writer works on the "remote" directly; the reader mirrors it
         // into `local_root` through the disk cache.
-        let mut writer = ChunkedVectors::<f32, MmapFile>::open(
-            MmapFs,
-            &dir,
-            DIM,
-            AdviceSetting::Global,
-            Populate::No,
-        )
-        .unwrap();
-        for s in 0..100 {
-            writer.push(make_vec(s, DIM).as_slice(), &hw).unwrap();
-        }
-        writer.flusher()().unwrap();
+        let mut writer =
+            UpdateOnlyChunkedVectors::<f32, MmapFile>::open(MmapFs, &dir, DIM).unwrap();
+        append_range(&mut writer, 0, 0..100, DIM, &hw);
 
         let cache_fs = DiskCacheFs::<MmapFile>::from_context(DiskCacheFsContext {
             config: Arc::new(DiskCacheConfig::new(remote_root, local_root).unwrap()),
@@ -168,17 +136,14 @@ mod tests {
         .unwrap();
         assert_eq!(reader.len(), 100);
 
-        // Read the tail vector: the fetched block extends past it into
-        // then-unwritten space — the stale bytes this test must escape are
-        // now in the reader's local cache.
+        // Read the tail vector: the fetched block ends at the old tail —
+        // the stale bytes this test must escape are now in the reader's
+        // local cache.
         let got = reader.get::<Random>(99).unwrap();
         assert_eq!(got.as_ref(), make_vec(99, DIM).as_slice());
 
         // Append into that same block region, then reload.
-        for s in 100..150 {
-            writer.push(make_vec(s, DIM).as_slice(), &hw).unwrap();
-        }
-        writer.flusher()().unwrap();
+        append_range(&mut writer, 100, 100..150, DIM, &hw);
 
         let empty = SortedSlice::new(&[]).unwrap();
         reader.live_reload(&cache_fs, &empty, &empty, &hw).unwrap();
@@ -205,18 +170,9 @@ mod tests {
             .unwrap();
         let hw = HardwareCounterCell::disposable();
 
-        let mut writer = ChunkedVectors::<f32, MmapFile>::open(
-            MmapFs,
-            dir.path(),
-            DIM,
-            AdviceSetting::Global,
-            Populate::No,
-        )
-        .unwrap();
-        for s in 0..4000 {
-            writer.push(make_vec(s, DIM).as_slice(), &hw).unwrap();
-        }
-        writer.flusher()().unwrap();
+        let mut writer =
+            UpdateOnlyChunkedVectors::<f32, MmapFile>::open(MmapFs, dir.path(), DIM).unwrap();
+        append_range(&mut writer, 0, 0..4000, DIM, &hw);
 
         let mut reader = ReadOnlyChunkedVectors::<f32, MmapFile>::open(
             &MmapFs,
@@ -229,10 +185,8 @@ mod tests {
         assert_eq!(reader.len(), 4000);
         assert_eq!(reader.chunks.len(), 1);
 
-        for s in 4000..9000 {
-            writer.push(make_vec(s, DIM).as_slice(), &hw).unwrap();
-        }
-        writer.flusher()().unwrap();
+        // Straddles two chunk boundaries: fills chunk 0, spans 1, starts 2.
+        append_range(&mut writer, 4000, 4000..9000, DIM, &hw);
 
         let empty = SortedSlice::new(&[]).unwrap();
         reader.live_reload(&MmapFs, &empty, &empty, &hw).unwrap();
@@ -264,18 +218,9 @@ mod tests {
             .unwrap();
         let hw = HardwareCounterCell::disposable();
 
-        let mut writer = ChunkedVectors::<f32, MmapFile>::open(
-            MmapFs,
-            dir.path(),
-            DIM,
-            AdviceSetting::Global,
-            Populate::No,
-        )
-        .unwrap();
-        for s in 0..100 {
-            writer.push(make_vec(s, DIM).as_slice(), &hw).unwrap();
-        }
-        writer.flusher()().unwrap();
+        let mut writer =
+            UpdateOnlyChunkedVectors::<f32, MmapFile>::open(MmapFs, dir.path(), DIM).unwrap();
+        append_range(&mut writer, 0, 0..100, DIM, &hw);
 
         let mut reader = ReadOnlyChunkedVectors::<f32, MmapFile>::open(
             &MmapFs,
@@ -292,10 +237,7 @@ mod tests {
         );
 
         // Grow within the same chunk so the reload takes the slow path.
-        for s in 100..150 {
-            writer.push(make_vec(s, DIM).as_slice(), &hw).unwrap();
-        }
-        writer.flusher()().unwrap();
+        append_range(&mut writer, 100, 100..150, DIM, &hw);
 
         // Inject a transient error: chunk 0 still exists but cannot be opened.
         let chunk_file = chunk_name(dir.path(), 0);

lib/segment/src/vector_storage/chunked_vectors/read_only/read_ops.rs

@@ -14,18 +14,6 @@ use crate::vector_storage::query_scorer::is_read_with_prefetch_efficient;
 use crate::vector_storage::{VectorOffset, VectorOffsetType};
 
 impl<T: bytemuck::Pod + Send, S: UniversalRead> ReadOnlyChunkedVectors<T, S> {
-    #[inline]
-    pub(in crate::vector_storage::chunked_vectors) fn get_chunk_index(&self, key: usize) -> usize {
-        key / self.config.chunk_size_vectors
-    }
-
-    /// Returns the byte offset of the vector in the chunk
-    #[inline]
-    pub(in crate::vector_storage::chunked_vectors) fn get_chunk_offset(&self, key: usize) -> usize {
-        let chunk_vector_idx = key % self.config.chunk_size_vectors;
-        chunk_vector_idx * self.config.dim
-    }
-
     #[inline]
     pub fn max_vector_size_bytes(&self) -> usize {
         self.config.chunk_size_bytes
@@ -43,9 +31,7 @@ impl<T: bytemuck::Pod + Send, S: UniversalRead> ReadOnlyChunkedVectors<T, S> {
 
     // returns how many vectors can be inserted starting from key
     pub fn get_remaining_chunk_keys(&self, start_key: VectorOffsetType) -> usize {
-        let start_key = start_key.as_();
-        let chunk_vector_idx = self.get_chunk_offset(start_key) / self.config.dim;
-        self.config.chunk_size_vectors - chunk_vector_idx
+        self.config.remaining_chunk_capacity(start_key.as_())
     }
 
     #[inline]
@@ -54,12 +40,12 @@ impl<T: bytemuck::Pod + Send, S: UniversalRead> ReadOnlyChunkedVectors<T, S> {
             return None;
         }
 
-        let chunk_idx = self.get_chunk_index(offset);
+        let chunk_idx = self.config.get_chunk_index(offset);
         if chunk_idx >= self.chunks.len() {
             return None;
         }
 
-        let element_offset = self.get_chunk_offset(offset);
+        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;

lib/segment/src/vector_storage/chunked_vectors/test_utils.rs

@@ -0,0 +1,25 @@
+//! Helpers shared by the `read_only` and `update_only` test modules.
+
+use common::counter::hardware_counter::HardwareCounterCell;
+use common::universal_io::MmapFile;
+
+use super::update_only::UpdateOnlyChunkedVectors;
+use crate::vector_storage::VectorOffsetType;
+
+pub(super) fn make_vec(seed: usize, dim: usize) -> Vec<f32> {
+    (0..dim).map(|i| (seed * dim + i) as f32).collect()
+}
+
+/// Append one durable batch of `make_vec(seed)` vectors through the writer.
+pub(super) fn append_range(
+    writer: &mut UpdateOnlyChunkedVectors<f32, MmapFile>,
+    start_key: VectorOffsetType,
+    seeds: std::ops::Range<usize>,
+    dim: usize,
+    hw: &HardwareCounterCell,
+) {
+    let batch: Vec<Vec<f32>> = seeds.map(|seed| make_vec(seed, dim)).collect();
+    writer
+        .append_many(start_key, batch.iter().map(|vector| vector.as_slice()), hw)
+        .unwrap();
+}

lib/segment/src/vector_storage/chunked_vectors/update_only/mod.rs

@@ -0,0 +1,212 @@
+#[cfg(test)]
+mod tests;
+
+use std::marker::PhantomData;
+use std::path::{Path, PathBuf};
+
+use common::counter::hardware_counter::HardwareCounterCell;
+use common::mmap::AdviceSetting;
+use common::universal_io::{
+    OpenOptions, Populate, UniversalAppend, UniversalReadFileOps, UniversalReadFs, UniversalWrite,
+    UniversalWriteFileOps,
+};
+
+use crate::common::operation_error::{OperationError, OperationResult};
+use crate::vector_storage::VectorOffsetType;
+use crate::vector_storage::chunked_vectors::chunks::{chunk_name, list_chunk_files};
+use crate::vector_storage::chunked_vectors::config::{
+    ChunkedVectorsConfig, Status, ensure_config, status_file,
+};
+
+/// Short-lived append-only writer for chunked vectors storage.
+///
+/// Holds no chunk handles: each [`append_many`](Self::append_many) opens the
+/// touched chunks, appends, and persists the vector count, so a batch is
+/// durable when it returns and there is nothing to flush.
+#[derive(Debug)]
+#[cfg_attr(not(test), expect(dead_code))]
+pub struct UpdateOnlyChunkedVectors<T, S: UniversalAppend> {
+    directory: PathBuf,
+    config: ChunkedVectorsConfig,
+    fs: S::Fs,
+    _t: PhantomData<T>,
+}
+
+/// Options for the short-lived append-only handles
+fn append_options() -> OpenOptions {
+    OpenOptions {
+        writeable: true,
+        need_sequential: false,
+        populate: Populate::No,
+        advice: AdviceSetting::Global,
+    }
+}
+
+#[cfg_attr(not(test), expect(dead_code))]
+impl<T, S> UpdateOnlyChunkedVectors<T, S>
+where
+    T: bytemuck::Pod + Send,
+    S: UniversalAppend + UniversalWrite + 'static,
+{
+    /// Open a chunked-vectors directory for appending, creating it if missing.
+    pub fn open(fs: S::Fs, directory: &Path, dim: usize) -> OperationResult<Self> {
+        let status_path = status_file(directory);
+        // An absent status file marks the first open. Writing it eagerly keeps
+        // the directory readable even if no batch ever lands.
+        if !fs.exists(&status_path)? {
+            fs.create_dir(directory)?;
+            fs.atomic_save(&status_path, bytemuck::bytes_of(&Status { len: 0 }))?;
+        }
+        let config = ensure_config::<T, _>(&fs, directory, dim, false)?;
+
+        Ok(Self {
+            directory: directory.to_owned(),
+            config,
+            fs,
+            _t: PhantomData,
+        })
+    }
+
+    /// Replace the stored vector count.
+    fn save_len(&self, len: usize) -> OperationResult<()> {
+        self.fs.atomic_save(
+            &status_file(&self.directory),
+            bytemuck::bytes_of(&Status { len }),
+        )?;
+        Ok(())
+    }
+
+    /// Compare every chunk file's length against an external total length.
+    ///
+    /// Ensures every file is at the expected length by truncating or filling with zeroes.
+    fn ensure_chunk_lengths(&self, target_len: usize) -> OperationResult<()> {
+        let total_bytes = target_len * self.config.dim * size_of::<T>();
+        let num_chunks = target_len.div_ceil(self.config.chunk_size_vectors);
+
+        let mut listed = list_chunk_files(&self.fs, &self.directory)?;
+
+        for chunk_id in 0..num_chunks {
+            let expected = self
+                .config
+                .chunk_size_bytes
+                .min(total_bytes.saturating_sub(chunk_id * self.config.chunk_size_bytes))
+                as u64;
+            match listed.remove(&chunk_id) {
+                Some(file_info) => {
+                    match file_info.size.cmp(&expected) {
+                        std::cmp::Ordering::Equal => {
+                            // Ok
+                        }
+                        std::cmp::Ordering::Less => {
+                            // fill with zeroes
+                            log::warn!(
+                                "Expected larger chunk, filling chunk {chunk_id} with zeroes"
+                            );
+                            let data = vec![0u8; (expected - file_info.size) as usize];
+                            let mut file =
+                                self.fs.open_append(&file_info.path, append_options())?;
+                            file.append(file_info.size, &data)?;
+                            file.flusher()()?;
+                        }
+                        std::cmp::Ordering::Greater => {
+                            // truncate
+                            log::warn!("Expected smaller chunk, truncating chunk {chunk_id}");
+                            let file = self.fs.open_append(&file_info.path, append_options())?;
+                            let content = file.read_whole::<u8>()?;
+                            let Some(truncated) = content.get(..expected as usize) else {
+                                return Err(OperationError::service_error(format!(
+                                    "Chunk {chunk_id} is {} bytes, shorter than the expected \
+                                     truncation length {expected}",
+                                    content.len(),
+                                )));
+                            };
+                            let truncated = truncated.to_vec();
+                            drop(file);
+                            self.fs.atomic_save(&file_info.path, &truncated)?;
+                        }
+                    }
+                }
+                None => {
+                    // create and fill with zeroes
+                    log::warn!(
+                        "Expected non-existing chunk {chunk_id}, creating and filling with zeroes"
+                    );
+                    let mut file = self.open_chunk_for_append(chunk_id, true)?;
+                    file.append(0, &vec![0u8; expected as usize])?;
+                    file.flusher()()?;
+                }
+            }
+        }
+
+        // Files past the boundary hold no data this writer should serve
+        for (chunk_id, file) in listed {
+            log::warn!(
+                "Chunk {chunk_id} past the target vector count ({} bytes). Removing.",
+                file.size,
+            );
+            self.fs.remove(&file.path)?;
+        }
+
+        self.save_len(target_len)?;
+
+        Ok(())
+    }
+
+    /// Append a batch of vectors at the end of the storage, one file append per
+    /// touched chunk, then persist the new vector count.
+    ///
+    /// This method trusts the `start_key` to be the source of truth, so it will
+    /// fill with zeroes or truncate chunks if necessary to make chunks' sizes
+    /// 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>(
+        &mut self,
+        start_key: VectorOffsetType,
+        vectors: impl IntoIterator<Item = &'a [T]>,
+        hw_counter: &HardwareCounterCell,
+    ) -> OperationResult<()> {
+        self.ensure_chunk_lengths(start_key)?;
+
+        let mut vectors = vectors.into_iter().peekable();
+        let mut len = start_key;
+
+        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);
+
+            let batch: Vec<&[T]> = vectors.by_ref().take(capacity).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)?;
+            chunk.append_batch(
+                (chunk_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)?;
+
+        Ok(())
+    }
+
+    /// Open the chunk for appending; a `new` chunk is past the watermark, so
+    /// it is created, truncating any leftover from a crashed writer.
+    fn open_chunk_for_append(&self, chunk_idx: usize, new: bool) -> OperationResult<S> {
+        let path = chunk_name(&self.directory, chunk_idx);
+        if new {
+            self.fs.create(&path, 0)?;
+        }
+        Ok(self.fs.open(&path, append_options(), Default::default())?)
+    }
+}

lib/segment/src/vector_storage/chunked_vectors/update_only/tests.rs

@@ -0,0 +1,282 @@
+use common::counter::hardware_counter::HardwareCounterCell;
+use common::generic_consts::Random;
+use common::mmap::AdviceSetting;
+use common::universal_io::{MmapFile, MmapFs, Populate};
+use tempfile::{Builder, TempDir};
+
+use super::UpdateOnlyChunkedVectors;
+use crate::vector_storage::VectorOffsetType;
+use crate::vector_storage::chunked_vectors::ChunkedVectors;
+use crate::vector_storage::chunked_vectors::chunks::chunk_name;
+use crate::vector_storage::chunked_vectors::config::{config_file, status_file};
+use crate::vector_storage::chunked_vectors::read_only::ReadOnlyChunkedVectors;
+use crate::vector_storage::chunked_vectors::test_utils::{append_range, make_vec};
+
+const DIM: usize = 32;
+/// Spans three test chunks (4096 vectors each), ending mid-chunk.
+const COUNT: usize = 9000;
+
+/// Write the same `COUNT` vectors through both writers: `ChunkedVectors` into
+/// the first directory, `UpdateOnlyChunkedVectors` into the second — the
+/// latter over two sessions to also exercise reopening mid-chunk.
+fn write_both() -> (TempDir, TempDir) {
+    let hw = HardwareCounterCell::disposable();
+    let plain_dir = Builder::new().prefix("chunked_plain").tempdir().unwrap();
+    let appended_dir = Builder::new().prefix("chunked_appended").tempdir().unwrap();
+
+    let mut plain = ChunkedVectors::<f32, MmapFile>::open(
+        MmapFs,
+        plain_dir.path(),
+        DIM,
+        AdviceSetting::Global,
+        Populate::No,
+    )
+    .unwrap();
+    for seed in 0..COUNT {
+        plain.push(make_vec(seed, DIM).as_slice(), &hw).unwrap();
+    }
+    plain.flusher()().unwrap();
+
+    for range in [0..COUNT / 2, COUNT / 2..COUNT] {
+        let mut writer =
+            UpdateOnlyChunkedVectors::<f32, MmapFile>::open(MmapFs, appended_dir.path(), DIM)
+                .unwrap();
+        append_range(&mut writer, range.start, range, DIM, &hw);
+    }
+
+    (plain_dir, appended_dir)
+}
+
+/// Both writers produce the same on-disk format: identical config and status
+/// files, and byte-identical chunk data — an appended chunk is the
+/// preallocated chunk minus the not-yet-written tail.
+#[test]
+fn writes_congruent_format() {
+    let (plain_dir, appended_dir) = write_both();
+
+    assert_eq!(
+        fs_err::read(config_file(plain_dir.path())).unwrap(),
+        fs_err::read(config_file(appended_dir.path())).unwrap(),
+        "config files differ",
+    );
+    assert_eq!(
+        fs_err::read(status_file(plain_dir.path())).unwrap(),
+        fs_err::read(status_file(appended_dir.path())).unwrap(),
+        "status files differ",
+    );
+
+    let mut remaining = COUNT * DIM * size_of::<f32>();
+    let mut chunk_id = 0;
+    while remaining > 0 {
+        let plain = fs_err::read(chunk_name(plain_dir.path(), chunk_id)).unwrap();
+        let appended = fs_err::read(chunk_name(appended_dir.path(), chunk_id)).unwrap();
+
+        let data_len = remaining.min(plain.len());
+        assert_eq!(appended.len(), data_len, "chunk {chunk_id} length");
+        assert!(appended[..] == plain[..data_len], "chunk {chunk_id} data");
+
+        remaining -= data_len;
+        chunk_id += 1;
+    }
+    assert!(chunk_id > 1, "the data must span multiple chunks");
+
+    // Neither directory has chunk files past the data
+    assert!(!chunk_name(plain_dir.path(), chunk_id).exists());
+    assert!(!chunk_name(appended_dir.path(), chunk_id).exists());
+}
+
+/// A reader over an append-written directory serves exactly what one over a
+/// `ChunkedVectors`-written directory does.
+#[test]
+fn directory_reads_congruently() {
+    let (plain_dir, appended_dir) = write_both();
+
+    let open_reader = |dir: &std::path::Path| {
+        ReadOnlyChunkedVectors::<f32, MmapFile>::open(
+            &MmapFs,
+            dir,
+            DIM,
+            AdviceSetting::Global,
+            Populate::No,
+        )
+        .unwrap()
+    };
+    let plain = open_reader(plain_dir.path());
+    let appended = open_reader(appended_dir.path());
+
+    assert_eq!(plain.len(), COUNT);
+    assert_eq!(appended.len(), COUNT);
+    for key in 0..COUNT {
+        let expected = make_vec(key, DIM);
+        let key = key as VectorOffsetType;
+        assert_eq!(
+            plain.get::<Random>(key).unwrap().as_ref(),
+            expected.as_slice(),
+        );
+        assert_eq!(
+            appended.get::<Random>(key).unwrap().as_ref(),
+            expected.as_slice(),
+        );
+    }
+}
+
+/// A preallocated (`ChunkedVectors`-written) directory is repaired on open:
+/// chunk files longer than the stored vector count implies are truncated back
+/// to the data, after which appends continue where the count left off.
+#[test]
+fn repairs_preallocated_chunks() {
+    let hw = HardwareCounterCell::disposable();
+    let dir = Builder::new().prefix("chunked_prealloc").tempdir().unwrap();
+
+    let mut plain = ChunkedVectors::<f32, MmapFile>::open(
+        MmapFs,
+        dir.path(),
+        DIM,
+        AdviceSetting::Global,
+        Populate::No,
+    )
+    .unwrap();
+    plain.push(make_vec(0, DIM).as_slice(), &hw).unwrap();
+    plain.flusher()().unwrap();
+    drop(plain);
+
+    let vector_bytes = (DIM * size_of::<f32>()) as u64;
+    let chunk = chunk_name(dir.path(), 0);
+    assert!(
+        fs_err::metadata(&chunk).unwrap().len() > vector_bytes,
+        "chunk must be preallocated past the single stored vector",
+    );
+
+    let mut writer =
+        UpdateOnlyChunkedVectors::<f32, MmapFile>::open(MmapFs, dir.path(), DIM).unwrap();
+
+    // Appends continue after it
+    append_range(&mut writer, 1, 1..3, DIM, &hw);
+
+    // File gets truncated before inserting the new vectors
+    assert!(
+        fs_err::metadata(&chunk).unwrap().len() == 3 * vector_bytes,
+        "file should be re-sized to the new length",
+    );
+
+    let reader = ReadOnlyChunkedVectors::<f32, MmapFile>::open(
+        &MmapFs,
+        dir.path(),
+        DIM,
+        AdviceSetting::Global,
+        Populate::No,
+    )
+    .unwrap();
+    assert_eq!(reader.len(), 3);
+    for key in 0..3 {
+        assert_eq!(
+            reader.get::<Random>(key).unwrap().as_ref(),
+            make_vec(key, DIM).as_slice(),
+        );
+    }
+}
+
+/// A batch whose first offset lands *behind* the persisted watermark is a
+/// replay of an already-applied range (e.g. a WAL resending a batch after a
+/// crash that happened before the outer commit pointer advanced, but after
+/// this writer's data was durable). `append_many` must shrink the chunk back
+/// to that offset and overwrite it, rather than blindly appending after the
+/// existing data and corrupting the offset-to-vector mapping.
+#[test]
+fn replaying_an_already_applied_range_overwrites_it() {
+    let hw = HardwareCounterCell::disposable();
+    let dir = Builder::new().prefix("chunked_replay").tempdir().unwrap();
+
+    let mut writer =
+        UpdateOnlyChunkedVectors::<f32, MmapFile>::open(MmapFs, dir.path(), DIM).unwrap();
+    append_range(&mut writer, 0, 0..100, DIM, &hw);
+
+    // Replay offsets 50..100 with different vectors than landed the first
+    // time, so the overwrite is observable.
+    append_range(&mut writer, 50, 1050..1100, DIM, &hw);
+
+    let reader = ReadOnlyChunkedVectors::<f32, MmapFile>::open(
+        &MmapFs,
+        dir.path(),
+        DIM,
+        AdviceSetting::Global,
+        Populate::No,
+    )
+    .unwrap();
+    // Not doubled: the watermark lands back at 100, not 150.
+    assert_eq!(reader.len(), 100);
+    for key in 0..50 {
+        assert_eq!(
+            reader
+                .get::<Random>(key as VectorOffsetType)
+                .unwrap()
+                .as_ref(),
+            make_vec(key, DIM).as_slice(),
+            "untouched prefix should be unchanged",
+        );
+    }
+    for key in 50..100 {
+        assert_eq!(
+            reader
+                .get::<Random>(key as VectorOffsetType)
+                .unwrap()
+                .as_ref(),
+            make_vec(key + 1000, DIM).as_slice(),
+            "replayed range should reflect the newer batch",
+        );
+    }
+}
+
+/// A batch whose first offset lands *ahead* of the persisted watermark
+/// skips a range (e.g. points deleted before ever getting a vector).
+/// `append_many` pads the gap with zero vectors instead of shifting later
+/// offsets down to close it.
+#[test]
+fn extends_across_a_gap_with_zeroes() {
+    let hw = HardwareCounterCell::disposable();
+    let dir = Builder::new().prefix("chunked_gap").tempdir().unwrap();
+
+    let mut writer =
+        UpdateOnlyChunkedVectors::<f32, MmapFile>::open(MmapFs, dir.path(), DIM).unwrap();
+    append_range(&mut writer, 0, 0..10, DIM, &hw);
+    // Offsets 10..15 are skipped; the next batch picks up at 15.
+    append_range(&mut writer, 15, 15..20, DIM, &hw);
+
+    let reader = ReadOnlyChunkedVectors::<f32, MmapFile>::open(
+        &MmapFs,
+        dir.path(),
+        DIM,
+        AdviceSetting::Global,
+        Populate::No,
+    )
+    .unwrap();
+    assert_eq!(reader.len(), 20);
+    for key in 0..10 {
+        assert_eq!(
+            reader
+                .get::<Random>(key as VectorOffsetType)
+                .unwrap()
+                .as_ref(),
+            make_vec(key, DIM).as_slice(),
+        );
+    }
+    for key in 10..15 {
+        assert_eq!(
+            reader
+                .get::<Random>(key as VectorOffsetType)
+                .unwrap()
+                .as_ref(),
+            vec![0.0f32; DIM].as_slice(),
+            "skipped offset {key} should read back as zeroes",
+        );
+    }
+    for key in 15..20 {
+        assert_eq!(
+            reader
+                .get::<Random>(key as VectorOffsetType)
+                .unwrap()
+                .as_ref(),
+            make_vec(key, DIM).as_slice(),
+        );
+    }
+}

lib/segment/src/vector_storage/chunked_vectors/write_ops.rs

@@ -50,8 +50,8 @@ where
         );
 
         let start_key = start_key.as_();
-        let chunk_idx = self.inner.get_chunk_index(start_key);
-        let chunk_offset = self.inner.get_chunk_offset(start_key);
+        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()