fix: fold appended points' deletion into read-only live_reload (#9948)

* fix: fold appended points' deletion into read-only live_reload

* fix: batch appended-deletion reads in read-only live_reload

* fix: clamp appended-deletion reload to persisted flag length

lib/segment/src/common/flags/in_memory_bitvec_flags.rs

@@ -1,7 +1,8 @@
-use std::path::Path;
+use std::path::{Path, PathBuf};
 
 use common::bitvec::{BitSlice, BitVec};
 use common::mmap::AdviceSetting;
+use common::sorted_slice::SortedSlice;
 use common::stored_bitslice::StoredBitSlice;
 use common::types::PointOffsetType;
 use common::universal_io::{
@@ -14,10 +15,15 @@ use crate::common::operation_error::{OperationError, OperationResult};
 /// In-memory counterpart of `BitvecFlags`: persisted flags materialized into an
 /// owned `BitVec`, no write path.
 ///
-/// Only consumer is the vector storages' `deleted` set, where the live-reload
-/// delta is authoritative (deleted offsets set, appended offsets live so absent).
-/// So unlike `ReadOnlyRoaringFlags` it keeps no file handle and never reopens —
-/// the owned bitvec is the whole state.
+/// The only consumer is the vector storages' `deleted` set. On live-reload the
+/// id-tracker delta supplies whole-point deletions ([`Self::insert_all`]), but
+/// an appended point can also carry a per-vector deletion recorded only in the
+/// on-disk flags file — a missing named vector is stored as a placeholder and
+/// its slot deleted. So the backing directory is retained and
+/// [`Self::reload_appended`] reopens the flags file to read the persisted bit
+/// of each appended offset. The set only ever grows, so a whole-point deletion
+/// already folded in but not yet flushed to the flags file is never lost to a
+/// re-read.
 #[derive(Debug)]
 #[allow(dead_code)] // pending: read-only vector storages will hold `deleted` as this
 pub struct InMemoryBitvecFlags {
@@ -25,6 +31,9 @@ pub struct InMemoryBitvecFlags {
     bitvec: BitVec,
     /// Set-flag count, kept in sync with `bitvec`.
     count: usize,
+    /// Backing directory of the dynamic flags file, so [`Self::reload_appended`]
+    /// can reopen it. `None` for flags built via [`Self::from_bitvec`].
+    directory: Option<PathBuf>,
 }
 
 /// Read-only mmap options: never writable, lazily paged, nothing populated.
@@ -56,6 +65,24 @@ impl InMemoryBitvecFlags {
         Ok(())
     }
 
+    /// Logical flag count from the status file; the flags file itself is padded
+    /// past this.
+    fn persisted_len<S: UniversalRead>(
+        fs: &impl UniversalReadFs<File = S>,
+        directory: &Path,
+    ) -> OperationResult<usize> {
+        // TypedStorage, not StoredStruct which is write-bound.
+        let status = TypedStorage::<S, DynamicFlagsStatus>::new(fs.open(
+            status_file(directory),
+            bitslice_open_options(Populate::No),
+            Default::default(),
+        )?);
+        Ok(status
+            .read_whole()?
+            .first()
+            .map_or(0, DynamicFlagsStatus::len))
+    }
+
     /// Open persisted flags read-only into an owned `BitVec`; creates and writes
     /// nothing. The flags file is padded past the logical length (held in the
     /// status file), so the bitvec is truncated to it and `count` is exact.
@@ -63,16 +90,7 @@ impl InMemoryBitvecFlags {
         fs: &impl UniversalReadFs<File = S>,
         directory: &Path,
     ) -> OperationResult<Self> {
-        // Length via TypedStorage; StoredStruct is write-bound.
-        let status = TypedStorage::<S, DynamicFlagsStatus>::new(fs.open(
-            status_file(directory),
-            bitslice_open_options(Populate::No),
-            Default::default(),
-        )?);
-        let len = status
-            .read_whole()?
-            .first()
-            .map_or(0, DynamicFlagsStatus::len);
+        let len = Self::persisted_len(fs, directory)?;
 
         let flags_path = directory.join(FLAGS_FILE);
         let flags = StoredBitSlice::<S>::open(
@@ -90,7 +108,11 @@ impl InMemoryBitvecFlags {
         })?;
         let count = bitvec.count_ones();
 
-        Ok(Self { bitvec, count })
+        Ok(Self {
+            bitvec,
+            count,
+            directory: Some(directory.to_path_buf()),
+        })
     }
 
     /// Wrap an already-materialized deletion `bitvec`, computing the set-flag
@@ -98,7 +120,11 @@ impl InMemoryBitvecFlags {
     /// flags read by [`Self::open`] (e.g. the immutable dense `deleted.dat`).
     pub fn from_bitvec(bitvec: BitVec) -> Self {
         let count = bitvec.count_ones();
-        Self { bitvec, count }
+        Self {
+            bitvec,
+            count,
+            directory: None,
+        }
     }
 
     /// Whether the flag at `key` is set; out-of-range keys read as unset.
@@ -129,6 +155,51 @@ impl InMemoryBitvecFlags {
             }
         }
     }
+
+    /// Fold the persisted deletion bit of each appended offset into the set.
+    ///
+    /// An appended point can carry a deleted vector slot whose flag lives only
+    /// in the on-disk flags file, not the id-tracker delta (see the type doc).
+    /// `new_points` is sorted, so the covering range up to the persisted length
+    /// is read in one batched read. A no-op for [`Self::from_bitvec`] flags.
+    pub fn reload_appended<S: UniversalRead>(
+        &mut self,
+        fs: &impl UniversalReadFs<File = S>,
+        new_points: &SortedSlice<'_, PointOffsetType>,
+    ) -> OperationResult<()> {
+        let Some(directory) = self.directory.clone() else {
+            return Ok(());
+        };
+        let (Some(&first), Some(&last)) = (new_points.first(), new_points.last()) else {
+            return Ok(());
+        };
+
+        let len = Self::persisted_len(fs, &directory)? as u64;
+        let start = u64::from(first);
+        let end = u64::from(last).saturating_add(1).min(len);
+        if start >= end {
+            return Ok(());
+        }
+
+        let flags = StoredBitSlice::<S>::open(
+            fs,
+            &directory.join(FLAGS_FILE),
+            bitslice_open_options(Populate::No),
+            Default::default(),
+        )?;
+        let bits = flags.read_bit_range(start..end)?;
+
+        let mut deleted = Vec::new();
+        for &point in new_points.iter() {
+            let index = (u64::from(point) - start) as usize;
+            if bits.get(index).as_deref().copied().unwrap_or(false) {
+                deleted.push(point);
+            }
+        }
+        self.insert_all(&deleted);
+
+        Ok(())
+    }
 }
 
 #[allow(clippy::default_constructed_unit_structs)]

lib/segment/src/vector_storage/dense/read_only/live_reload.rs

@@ -13,8 +13,9 @@ impl<T: PrimitiveVectorElement, S: UniversalRead> LiveReload
 {
     type Fs = S::Fs;
 
-    /// Reload the chunked vectors and apply `deleted_points`; appended points are
-    /// served from the refreshed chunks, so `new_points` is unused.
+    /// Reload the chunked vectors, apply `deleted_points`, and fold in the
+    /// persisted deletion of each appended offset — a live point may have a
+    /// deleted vector slot recorded only on disk.
     fn live_reload(
         &mut self,
         fs: &S::Fs,
@@ -25,6 +26,7 @@ impl<T: PrimitiveVectorElement, S: UniversalRead> LiveReload
         self.vectors
             .live_reload(fs, deleted_points, new_points, hw_counter)?;
         self.deleted.insert_all(deleted_points);
+        self.deleted.reload_appended::<S>(fs, new_points)?;
 
         Ok(())
     }

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

@@ -194,4 +194,124 @@ mod tests {
         }
         assert!(!reader.is_deleted_vector(0));
     }
+
+    #[test]
+    fn live_reload_picks_up_appended_vector_deletion() {
+        const DIM: usize = 4;
+        let dir = Builder::new()
+            .prefix("ro_dense_appended_deleted")
+            .tempdir()
+            .unwrap();
+        let hw = HardwareCounterCell::disposable();
+
+        let mut writer = open_appendable_memmap_vector_storage_impl::<VectorElementType>(
+            dir.path(),
+            DIM,
+            Distance::Dot,
+            AdviceSetting::Global,
+            false,
+        )
+        .unwrap();
+        writer
+            .insert_vector(0, VectorRef::from(&vec![1.0; DIM]), &hw)
+            .unwrap();
+        writer.flusher()().unwrap();
+
+        let mut reader = ReadOnlyChunkedDenseVectorStorage::<VectorElementType, MmapFile>::open(
+            &MmapFs,
+            dir.path(),
+            DIM,
+            Distance::Dot,
+            AdviceSetting::Global,
+            Populate::No,
+        )
+        .unwrap();
+
+        writer
+            .insert_vector(1, VectorRef::from(&vec![0.0; DIM]), &hw)
+            .unwrap();
+        writer.delete_vector(1).unwrap();
+        writer.flusher()().unwrap();
+
+        let deleted_ids: Vec<PointOffsetType> = vec![];
+        let new_ids: Vec<PointOffsetType> = vec![1];
+        reader
+            .live_reload(
+                &MmapFs,
+                &SortedSlice::new(&deleted_ids).unwrap(),
+                &SortedSlice::new(&new_ids).unwrap(),
+                &hw,
+            )
+            .unwrap();
+
+        assert_eq!(reader.total_vector_count(), 2);
+        assert!(reader.is_deleted_vector(1));
+    }
+
+    #[test]
+    fn live_reload_batches_appended_vector_deletions() {
+        const DIM: usize = 4;
+        let dir = Builder::new()
+            .prefix("ro_dense_appended_batch")
+            .tempdir()
+            .unwrap();
+        let hw = HardwareCounterCell::disposable();
+
+        let mut writer = open_appendable_memmap_vector_storage_impl::<VectorElementType>(
+            dir.path(),
+            DIM,
+            Distance::Dot,
+            AdviceSetting::Global,
+            false,
+        )
+        .unwrap();
+        for id in 0..3u32 {
+            writer
+                .insert_vector(id, VectorRef::from(&vec![1.0; DIM]), &hw)
+                .unwrap();
+        }
+        writer.flusher()().unwrap();
+
+        let mut reader = ReadOnlyChunkedDenseVectorStorage::<VectorElementType, MmapFile>::open(
+            &MmapFs,
+            dir.path(),
+            DIM,
+            Distance::Dot,
+            AdviceSetting::Global,
+            Populate::No,
+        )
+        .unwrap();
+
+        for id in 3..8u32 {
+            writer
+                .insert_vector(id, VectorRef::from(&vec![0.0; DIM]), &hw)
+                .unwrap();
+        }
+        let deleted_appended: Vec<PointOffsetType> = vec![4, 6];
+        for &id in &deleted_appended {
+            writer.delete_vector(id).unwrap();
+        }
+        writer.flusher()().unwrap();
+
+        let deleted_ids: Vec<PointOffsetType> = vec![];
+        let new_ids: Vec<PointOffsetType> = (3..8).collect();
+        reader
+            .live_reload(
+                &MmapFs,
+                &SortedSlice::new(&deleted_ids).unwrap(),
+                &SortedSlice::new(&new_ids).unwrap(),
+                &hw,
+            )
+            .unwrap();
+
+        assert_eq!(reader.total_vector_count(), 8);
+        for id in 3..8 {
+            assert_eq!(
+                reader.is_deleted_vector(id),
+                deleted_appended.contains(&id),
+                "appended offset {id}",
+            );
+        }
+        assert!(!reader.is_deleted_vector(0));
+    }
 }

lib/segment/src/vector_storage/multi_dense/read_only/live_reload.rs

@@ -13,8 +13,9 @@ impl<T: PrimitiveVectorElement, S: UniversalRead> LiveReload
 {
     type Fs = S::Fs;
 
-    /// Reload the vectors and offsets and apply `deleted_points`; appended points
-    /// are served from the refreshed chunks, so `new_points` is unused.
+    /// Reload the vectors and offsets, apply `deleted_points`, and fold in the
+    /// persisted deletion of each appended offset — a live point may have a
+    /// deleted vector slot recorded only on disk.
     fn live_reload(
         &mut self,
         fs: &S::Fs,
@@ -27,6 +28,7 @@ impl<T: PrimitiveVectorElement, S: UniversalRead> LiveReload
         self.offsets
             .live_reload(fs, deleted_points, new_points, hw_counter)?;
         self.deleted.insert_all(deleted_points);
+        self.deleted.reload_appended::<S>(fs, new_points)?;
 
         Ok(())
     }

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

@@ -257,4 +257,64 @@ mod tests {
         }
         assert!(!reader.is_deleted_vector(0));
     }
+
+    #[test]
+    fn live_reload_picks_up_appended_vector_deletion() {
+        const DIM: usize = 8;
+        let dir = Builder::new()
+            .prefix("ro_multi_appended_deleted")
+            .tempdir()
+            .unwrap();
+        let hw = HardwareCounterCell::disposable();
+
+        let multi = |value: VectorElementType| {
+            MultiDenseVectorInternal::try_from(vec![vec![value; DIM]]).unwrap()
+        };
+
+        let mut writer = open_appendable_memmap_multi_vector_storage_impl::<VectorElementType>(
+            dir.path(),
+            DIM,
+            Distance::Dot,
+            MultiVectorConfig::default(),
+            AdviceSetting::Global,
+            false,
+        )
+        .unwrap();
+        writer
+            .insert_vector(0, VectorRef::from(&multi(1.0)), &hw)
+            .unwrap();
+        writer.flusher()().unwrap();
+
+        let mut reader =
+            ReadOnlyChunkedMultiDenseVectorStorage::<VectorElementType, MmapFile>::open(
+                &MmapFs,
+                dir.path(),
+                DIM,
+                Distance::Dot,
+                MultiVectorConfig::default(),
+                AdviceSetting::Global,
+                Populate::No,
+            )
+            .unwrap();
+
+        writer
+            .insert_vector(1, VectorRef::from(&multi(0.0)), &hw)
+            .unwrap();
+        writer.delete_vector(1).unwrap();
+        writer.flusher()().unwrap();
+
+        let deleted_ids: Vec<PointOffsetType> = vec![];
+        let new_ids: Vec<PointOffsetType> = vec![1];
+        reader
+            .live_reload(
+                &MmapFs,
+                &SortedSlice::new(&deleted_ids).unwrap(),
+                &SortedSlice::new(&new_ids).unwrap(),
+                &hw,
+            )
+            .unwrap();
+
+        assert_eq!(reader.total_vector_count(), 2);
+        assert!(reader.is_deleted_vector(1));
+    }
 }

lib/segment/src/vector_storage/sparse/read_only/live_reload.rs

@@ -10,18 +10,18 @@ use crate::common::operation_error::OperationResult;
 impl<S: UniversalRead> LiveReload for ReadOnlySparseVectorStorage<S> {
     type Fs = S::Fs;
 
-    /// Reload the Gridstore, apply `deleted_points`, and recompute
-    /// `next_point_offset`; appended points come from the Gridstore, so
-    /// `new_points` is unused.
+    /// Reload the Gridstore, apply `deleted_points`, fold in the persisted
+    /// deletion of each appended offset, and recompute `next_point_offset`.
     fn live_reload(
         &mut self,
         fs: &S::Fs,
         deleted_points: &SortedSlice<'_, PointOffsetType>,
-        _new_points: &SortedSlice<'_, PointOffsetType>,
+        new_points: &SortedSlice<'_, PointOffsetType>,
         _hw_counter: &HardwareCounterCell,
     ) -> OperationResult<()> {
         self.storage.live_reload(fs)?;
         self.deleted.insert_all(deleted_points);
+        self.deleted.reload_appended::<S>(fs, new_points)?;
 
         self.next_point_offset = self
             .deleted

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

@@ -235,4 +235,49 @@ mod tests {
         }
         assert!(!reader.is_deleted_vector(0));
     }
+
+    #[test]
+    fn live_reload_picks_up_appended_vector_deletion() {
+        let dir = Builder::new()
+            .prefix("ro_sparse_appended_deleted")
+            .tempdir()
+            .unwrap();
+        let hw = HardwareCounterCell::disposable();
+
+        fn make(id: usize) -> SparseVector {
+            SparseVector {
+                indices: vec![1, 5, 9],
+                values: vec![id as f32 + 0.1, id as f32 + 0.2, id as f32 + 0.3],
+            }
+        }
+
+        let mut writer = MmapSparseVectorStorage::open_or_create(dir.path()).unwrap();
+        writer
+            .insert_vector(0, VectorRef::from(&make(0)), &hw)
+            .unwrap();
+        writer.flusher()().unwrap();
+
+        let mut reader =
+            ReadOnlySparseVectorStorage::<MmapFile>::open(&MmapFs, dir.path(), Populate::No)
+                .unwrap();
+
+        writer
+            .insert_vector(1, VectorRef::from(&make(1)), &hw)
+            .unwrap();
+        writer.delete_vector(1).unwrap();
+        writer.flusher()().unwrap();
+
+        let deleted_ids: Vec<PointOffsetType> = vec![];
+        let new_ids: Vec<PointOffsetType> = vec![1];
+        reader
+            .live_reload(
+                &MmapFs,
+                &SortedSlice::new(&deleted_ids).unwrap(),
+                &SortedSlice::new(&new_ids).unwrap(),
+                &hw,
+            )
+            .unwrap();
+
+        assert!(reader.is_deleted_vector(1));
+    }
 }

lib/segment/src/vector_storage/turbo/read_only/live_reload.rs

@@ -11,7 +11,9 @@ impl<S: UniversalRead> LiveReload for ReadOnlyTurboVectorStorage<S> {
     type Fs = S::Fs;
 
     /// Pick up vectors a writer appended (chunked backend only; the single-file
-    /// layout is immutable) and patch the in-memory deletion flags.
+    /// layout is immutable), patch the in-memory deletion flags, and fold in the
+    /// persisted deletion of each appended offset — a live point may have a
+    /// deleted vector slot recorded only on disk.
     fn live_reload(
         &mut self,
         fs: &S::Fs,
@@ -23,6 +25,7 @@ impl<S: UniversalRead> LiveReload for ReadOnlyTurboVectorStorage<S> {
             storage.live_reload(fs, deleted_points, new_points, hw_counter)?;
         }
         self.deleted.insert_all(deleted_points);
+        self.deleted.reload_appended::<S>(fs, new_points)?;
         Ok(())
     }
 }
@@ -30,8 +33,10 @@ impl<S: UniversalRead> LiveReload for ReadOnlyTurboVectorStorage<S> {
 impl<S: UniversalRead> LiveReload for ReadOnlyTurboMultiVectorStorage<S> {
     type Fs = S::Fs;
 
-    /// Pick up multivectors a writer appended (records + offsets) and patch the
-    /// in-memory deletion flags.
+    /// Pick up multivectors a writer appended (records + offsets), patch the
+    /// in-memory deletion flags, and fold in the persisted deletion of each
+    /// appended offset — a live point may have a deleted vector slot recorded
+    /// only on disk.
     fn live_reload(
         &mut self,
         fs: &S::Fs,
@@ -44,6 +49,7 @@ impl<S: UniversalRead> LiveReload for ReadOnlyTurboMultiVectorStorage<S> {
         self.offsets
             .live_reload(fs, deleted_points, new_points, hw_counter)?;
         self.deleted.insert_all(deleted_points);
+        self.deleted.reload_appended::<S>(fs, new_points)?;
         Ok(())
     }
 }