refactor(segment): split iter_points/check_error; add VectorDataRead trait (#8863)

* refactor(segment): split iter_points/check_error to StorageSegmentEntry; add VectorDataRead trait

Step 0 of the SegmentReadView migration.

* Move `iter_points` and `check_error` from `ReadSegmentEntry` to
  `StorageSegmentEntry`. They are not part of the read-only logic shared
  between `Segment` and the upcoming `ReadOnlySegment`: `iter_points` is
  consumed only by optimizers and tests, and `check_error` is an update-flow
  concern. Update implementations on `Segment` and `ProxySegment`, plus
  `StorageSegmentEntry` imports in test/optimizer call sites that hold a
  concrete `Segment`.

* Introduce `VectorDataRead` trait on `VectorData` exposing `vector_index()`
  and `vector_storage()` as `Deref`-bounded handles to `VectorIndexRead` and
  `VectorStorageRead` (via GATs). This is the abstraction the future
  `SegmentReadView` will use for per-vector access.

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

* refactor: drop iter_points from StorageSegmentEntry; make it Segment-only

`iter_points` was never implementable for `ProxySegment` (its impl was a
plain `unimplemented!()` panic). Instead of carrying a half-supported
trait method, expose `iter_points` as an inherent method on `Segment`
and force callers that operated on `dyn SegmentEntry` to first downcast
through `LockedSegment::Original`. Proxy-segment branches now panic
explicitly with a descriptive message — same observable behavior, but
the contract is clear at the call site.

Updated callers (all previously assumed non-proxy):
* `SegmentHolder::find_duplicated_points` (dedup)
* `vacuum_optimizer` test helper
* `collection_manager::tests` cow-segment assertion
* `deferred_points_dedup` test

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

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

lib/collection/src/collection_manager/optimizers/vacuum_optimizer.rs

@@ -13,7 +13,7 @@ mod tests {
 
     use common::counter::hardware_counter::HardwareCounterCell;
     use itertools::Itertools;
-    use segment::entry::{NonAppendableSegmentEntry as _, ReadSegmentEntry as _};
+    use segment::entry::NonAppendableSegmentEntry as _;
     use segment::id_tracker::IdTrackerRead;
     use segment::index::VectorIndexRead;
     use segment::payload_json;
@@ -100,13 +100,13 @@ mod tests {
 
         let hw_counter = HardwareCounterCell::new();
 
-        let original_segment_path = match segment {
-            LockedSegment::Original(s) => s.read().segment_path.clone(),
+        let original_segment = match segment {
+            LockedSegment::Original(s) => s,
             LockedSegment::Proxy(_) => panic!("Not expected"),
         };
+        let original_segment_path = original_segment.read().segment_path.clone();
 
-        let segment_points_to_delete = segment
-            .get()
+        let segment_points_to_delete = original_segment
             .read()
             .iter_points()
             .enumerate()
@@ -121,16 +121,14 @@ mod tests {
                 .unwrap();
         }
 
-        let segment_points_to_assign1 = segment
-            .get()
+        let segment_points_to_assign1 = original_segment
             .read()
             .iter_points()
             .enumerate()
             .filter_map(|(i, point_id)| (i % 20 == 0).then_some(point_id))
             .collect_vec();
 
-        let segment_points_to_assign2 = segment
-            .get()
+        let segment_points_to_assign2 = original_segment
             .read()
             .iter_points()
             .enumerate()

lib/collection/src/collection_manager/tests/mod.rs

@@ -173,9 +173,12 @@ fn test_move_points_to_copy_on_write() {
 
     // Copy-on-write segment should contain all 3 points
 
-    let cow_segment = segments_write.get(sid2).unwrap();
+    let cow_segment = match segments_write.get(sid2).unwrap() {
+        shard::locked_segment::LockedSegment::Original(segment) => segment.clone(),
+        shard::locked_segment::LockedSegment::Proxy(_) => panic!("cow segment must be Original"),
+    };
 
-    let cow_segment_read = cow_segment.get().read();
+    let cow_segment_read = cow_segment.read();
 
     let cow_points: HashSet<_> = cow_segment_read.iter_points().collect();
 

lib/collection/src/tests/deferred_points_dedup.rs

@@ -6,6 +6,7 @@ use common::counter::hardware_accumulator::HwMeasurementAcc;
 use common::save_on_disk::SaveOnDisk;
 use rand::rng;
 use segment::data_types::vectors::VectorStructInternal;
+use segment::entry::ReadSegmentEntry as _;
 use segment::fixtures::payload_fixtures::random_vector;
 use segment::types::{Distance, Filter, PointIdType};
 use tempfile::{Builder, TempDir};
@@ -149,8 +150,13 @@ fn assert_no_duplicate_point_ids(shard: &LocalShard) {
     > = std::collections::HashMap::new();
 
     for (seg_id, segment) in holder.iter() {
-        let seg = segment.get();
-        let seg_read = seg.read();
+        let original = match segment {
+            shard::locked_segment::LockedSegment::Original(segment) => segment,
+            shard::locked_segment::LockedSegment::Proxy(_) => {
+                panic!("test does not expect proxy segments")
+            }
+        };
+        let seg_read = original.read();
         for pid in seg_read.iter_points() {
             let is_deferred = seg_read.point_is_deferred(pid);
             point_occurrences

lib/segment/src/entry/entry_point.rs

@@ -101,9 +101,6 @@ pub trait ReadSegmentEntry: SnapshotEntry {
         hw_counter: &HardwareCounterCell,
     ) -> OperationResult<Payload>;
 
-    /// Iterator over all points in segment in ascending order.
-    fn iter_points(&self) -> Box<dyn Iterator<Item = PointIdType> + '_>;
-
     /// Paginate over points which satisfies filtering condition starting with `offset` id including.
     ///
     /// Cancelled by `is_stopped` flag.
@@ -236,9 +233,6 @@ pub trait ReadSegmentEntry: SnapshotEntry {
     /// Get indexed fields
     fn get_indexed_fields(&self) -> HashMap<PayloadKeyType, PayloadFieldSchema>;
 
-    /// Checks if segment errored during last operations
-    fn check_error(&self) -> Option<SegmentFailedState>;
-
     // Get collected telemetry data of segment
     fn get_telemetry_data(&self, detail: TelemetryDetail) -> SegmentTelemetry;
 
@@ -265,6 +259,9 @@ pub trait ReadSegmentEntry: SnapshotEntry {
 
 /// Segment with storage.
 pub trait StorageSegmentEntry: ReadSegmentEntry {
+    /// Checks if segment errored during last operations
+    fn check_error(&self) -> Option<SegmentFailedState>;
+
     /// Get current persistent version of the segment
     fn persistent_version(&self) -> SeqNumberType;
 

lib/segment/src/segment/entry.rs

@@ -265,14 +265,6 @@ impl ReadSegmentEntry for Segment {
         Ok(records)
     }
 
-    fn iter_points(&self) -> Box<dyn Iterator<Item = PointIdType> + '_> {
-        let mappings =
-            PointMappingsGuard::new(self.id_tracker.borrow(), |guard| guard.point_mappings());
-        Box::new(IterPointsIterator::new(mappings, |mappings| {
-            mappings.borrow_dependent().iter_external()
-        }))
-    }
-
     fn read_filtered<'a>(
         &'a self,
         offset: Option<PointIdType>,
@@ -564,10 +556,6 @@ impl ReadSegmentEntry for Segment {
         self.payload_index.borrow().indexed_fields()
     }
 
-    fn check_error(&self) -> Option<SegmentFailedState> {
-        self.error_status.clone()
-    }
-
     fn vector_names(&self) -> HashSet<VectorNameBuf> {
         self.vector_data.keys().cloned().collect()
     }
@@ -669,7 +657,22 @@ impl ReadSegmentEntry for Segment {
     }
 }
 
+impl Segment {
+    /// Iterator over all points in segment in ascending order.
+    pub fn iter_points(&self) -> Box<dyn Iterator<Item = PointIdType> + '_> {
+        let mappings =
+            PointMappingsGuard::new(self.id_tracker.borrow(), |guard| guard.point_mappings());
+        Box::new(IterPointsIterator::new(mappings, |mappings| {
+            mappings.borrow_dependent().iter_external()
+        }))
+    }
+}
+
 impl StorageSegmentEntry for Segment {
+    fn check_error(&self) -> Option<SegmentFailedState> {
+        self.error_status.clone()
+    }
+
     fn persistent_version(&self) -> SeqNumberType {
         (*self.persisted_version.lock()).unwrap_or(0)
     }

lib/segment/src/segment/mod.rs

@@ -7,6 +7,7 @@ mod sampling;
 mod scroll;
 mod search;
 mod segment_ops;
+pub mod vector_data_read;
 mod vector_name_ops;
 mod version_tracker;
 

lib/segment/src/segment/vector_data_read.rs

@@ -0,0 +1,39 @@
+use std::ops::Deref;
+
+use atomic_refcell::AtomicRef;
+
+use crate::index::{VectorIndexEnum, VectorIndexRead};
+use crate::segment::VectorData;
+use crate::vector_storage::{VectorStorageEnum, VectorStorageRead};
+
+/// Read-only view over a single named vector entry of a segment.
+///
+/// Abstracts the concrete `VectorData` so that `SegmentReadView` can be
+/// constructed from any segment-like type (e.g. a future `ReadOnlySegment`)
+/// that exposes its per-vector storages through this trait.
+pub trait VectorDataRead {
+    type IndexRef<'a>: Deref<Target: VectorIndexRead>
+    where
+        Self: 'a;
+
+    type StorageRef<'a>: Deref<Target: VectorStorageRead>
+    where
+        Self: 'a;
+
+    fn vector_index(&self) -> Self::IndexRef<'_>;
+
+    fn vector_storage(&self) -> Self::StorageRef<'_>;
+}
+
+impl VectorDataRead for VectorData {
+    type IndexRef<'a> = AtomicRef<'a, VectorIndexEnum>;
+    type StorageRef<'a> = AtomicRef<'a, VectorStorageEnum>;
+
+    fn vector_index(&self) -> Self::IndexRef<'_> {
+        self.vector_index.borrow()
+    }
+
+    fn vector_storage(&self) -> Self::StorageRef<'_> {
+        self.vector_storage.borrow()
+    }
+}

lib/shard/src/proxy_segment/segment_entry.rs

@@ -267,13 +267,6 @@ impl ReadSegmentEntry for ProxySegment {
         )
     }
 
-    /// Not implemented for proxy
-    fn iter_points(&self) -> Box<dyn Iterator<Item = PointIdType> + '_> {
-        // get_points is not available for Proxy implementation
-        // Due to internal locks it is almost impossible to return iterator with proper owning, lifetimes, e.t.c.
-        unimplemented!("call to get_points is not implemented for Proxy segment")
-    }
-
     fn read_filtered<'a>(
         &'a self,
         offset: Option<PointIdType>,
@@ -679,10 +672,6 @@ impl ReadSegmentEntry for ProxySegment {
         indexed_fields
     }
 
-    fn check_error(&self) -> Option<SegmentFailedState> {
-        self.wrapped_segment.get().read().check_error()
-    }
-
     fn vector_names(&self) -> HashSet<VectorNameBuf> {
         self.wrapped_segment.get().read().vector_names()
     }
@@ -730,6 +719,10 @@ impl ReadSegmentEntry for ProxySegment {
 }
 
 impl StorageSegmentEntry for ProxySegment {
+    fn check_error(&self) -> Option<SegmentFailedState> {
+        self.wrapped_segment.get().read().check_error()
+    }
+
     fn persistent_version(&self) -> SeqNumberType {
         self.wrapped_segment.get().read().persistent_version()
     }

lib/shard/src/segment_holder/mod.rs

@@ -916,13 +916,24 @@ impl SegmentHolder {
             is_deferred: bool,
         }
 
+        // Dedup needs to enumerate all points in every segment, which is only
+        // available on a concrete `Segment`. Proxy segments cannot be enumerated
+        // this way (their internals span a wrapped read segment plus an
+        // in-memory write segment), so we panic if one shows up here — matching
+        // the pre-existing behavior when `iter_points` was a trait method that
+        // `unimplemented!()`'d for proxies.
         let segments = self
             .iter()
-            .map(|(segment_id, locked_segment)| (segment_id, locked_segment.get()))
+            .map(|(segment_id, locked_segment)| match locked_segment {
+                LockedSegment::Original(segment) => (segment_id, segment.as_ref()),
+                LockedSegment::Proxy(_) => panic!(
+                    "deduplicate_points cannot enumerate points of proxy segment {segment_id}",
+                ),
+            })
             .collect::<Vec<_>>();
         let locked_segments = segments
             .iter()
-            .map(|(segment_id, locked_segment)| (*segment_id, locked_segment.read()))
+            .map(|(segment_id, segment)| (*segment_id, segment.read()))
             .collect::<BTreeMap<_, _>>();
 
         // Iterator produces groups of points by point ID