chore: Refactor mutable map index inner (#9319)

* replace `ingest` with `add_many_to_map`

* rename to InMemoryMapIndex

* rename module

* fmt

* clippy

* don't index if there's no values

lib/segment/src/index/field_index/map_index/mutable_map_index/in_memory.rs

@@ -17,7 +17,7 @@ use crate::index::payload_config::StorageType;
 /// Both wrappers add a different backing storage (`Gridstore` vs
 /// `GridstoreReader`); the in-memory layout that serves every
 /// [`MapIndexRead`] method is the same, so it lives here once.
-pub(in crate::index::field_index::map_index) struct MutableMapIndexInner<N: MapIndexKey + ?Sized>
+pub(in crate::index::field_index::map_index) struct InMemoryMapIndex<N: MapIndexKey + ?Sized>
 where
     Vec<<N as MapIndexKey>::Owned>: Blob + Send + Sync,
 {
@@ -30,7 +30,7 @@ where
     pub(in crate::index::field_index::map_index) values_count: usize,
 }
 
-impl<N: MapIndexKey + ?Sized> MutableMapIndexInner<N>
+impl<N: MapIndexKey + ?Sized> InMemoryMapIndex<N>
 where
     Vec<<N as MapIndexKey>::Owned>: Blob + Send + Sync,
 {
@@ -43,29 +43,25 @@ where
         }
     }
 
-    /// Apply one `(point_id, value)` pair from a backing store iteration.
-    ///
-    /// Used by `MutableMapIndex::open_gridstore` to populate the in-memory
-    /// state from on-disk data; the matching loader for
-    /// [`super::read_only::ReadOnlyAppendableMapIndex`] will reuse this helper
-    /// once its lifecycle layer lands.
-    pub(in crate::index::field_index::map_index) fn ingest(
-        &mut self,
-        idx: PointOffsetType,
-        value: <N as MapIndexKey>::Owned,
-    ) {
+    pub fn add_many_to_map(&mut self, idx: u32, values: Vec<<N as MapIndexKey>::Owned>) {
+        if values.is_empty() {
+            return;
+        }
+
+        self.values_count += values.len();
         if self.point_to_values.len() <= idx as usize {
-            self.point_to_values.resize_with(idx as usize + 1, Vec::new);
+            self.point_to_values.resize_with(idx as usize + 1, Vec::new)
         }
-        let point_values = &mut self.point_to_values[idx as usize];
 
-        if point_values.is_empty() {
-            self.indexed_points += 1;
+        self.point_to_values[idx as usize] = Vec::with_capacity(values.len());
+
+        for value in values {
+            let entry = self.map.entry(value);
+            self.point_to_values[idx as usize].push(entry.key().clone());
+            entry.or_default().insert(idx);
         }
-        self.values_count += 1;
 
-        point_values.push(value.clone());
-        self.map.entry(value).or_default().insert(idx);
+        self.indexed_points += 1;
     }
 
     /// Remove a point from the in-memory state, updating the value map and
@@ -107,7 +103,7 @@ where
     }
 }
 
-impl<N: MapIndexKey + ?Sized> MapIndexRead<N> for MutableMapIndexInner<N>
+impl<N: MapIndexKey + ?Sized> MapIndexRead<N> for InMemoryMapIndex<N>
 where
     Vec<<N as MapIndexKey>::Owned>: Blob + Send + Sync,
 {

lib/segment/src/index/field_index/map_index/mutable_map_index/lifecycle.rs

@@ -8,7 +8,7 @@ use gridstore::{Blob, Gridstore};
 
 use super::super::MapIndexKey;
 use super::MutableMapIndex;
-use super::inner::MutableMapIndexInner;
+use super::in_memory::InMemoryMapIndex;
 use crate::common::Flusher;
 use crate::common::operation_error::{OperationError, OperationResult};
 
@@ -52,16 +52,14 @@ where
         };
 
         // Load in-memory index from Gridstore
-        let mut inner = MutableMapIndexInner::<N>::empty();
+        let mut in_memory_index = InMemoryMapIndex::<N>::empty();
 
         let hw_counter = HardwareCounterCell::disposable();
         let hw_counter_ref = hw_counter.ref_payload_index_io_write_counter();
         store
             .iter::<_, GridstoreError>(
                 |idx, values: Vec<_>| {
-                    for value in values {
-                        inner.ingest(idx, value);
-                    }
+                    in_memory_index.add_many_to_map(idx, values);
                     Ok(true)
                 },
                 hw_counter_ref,
@@ -70,7 +68,7 @@ where
             .unwrap();
 
         Ok(Some(Self {
-            inner,
+            in_memory_index,
             storage: store,
         }))
     }
@@ -88,23 +86,8 @@ where
             return Ok(());
         }
 
-        self.inner.values_count += values.len();
-        if self.inner.point_to_values.len() <= idx as usize {
-            self.inner
-                .point_to_values
-                .resize_with(idx as usize + 1, Vec::new)
-        }
-
-        self.inner.point_to_values[idx as usize] = Vec::with_capacity(values.len());
-
         let hw_counter_ref = hw_counter.ref_payload_index_io_write_counter();
 
-        for value in values.clone() {
-            let entry = self.inner.map.entry(value.into());
-            self.inner.point_to_values[idx as usize].push(entry.key().clone());
-            entry.or_default().insert(idx);
-        }
-
         let values = values.into_iter().map(Into::into).collect::<Vec<_>>();
         self.storage
             .put_value(idx, &values, hw_counter_ref)
@@ -114,12 +97,13 @@ where
                 ))
             })?;
 
-        self.inner.indexed_points += 1;
+        self.in_memory_index.add_many_to_map(idx, values);
+
         Ok(())
     }
 
     pub fn remove_point(&mut self, idx: PointOffsetType) -> OperationResult<()> {
-        if !self.inner.remove_point(idx) {
+        if !self.in_memory_index.remove_point(idx) {
             return Ok(());
         }
 

lib/segment/src/index/field_index/map_index/mutable_map_index/mod.rs

@@ -1,9 +1,9 @@
 use gridstore::{Blob, Gridstore};
 
-use self::inner::MutableMapIndexInner;
+use self::in_memory::InMemoryMapIndex;
 use super::MapIndexKey;
 
-pub(super) mod inner;
+pub(super) mod in_memory;
 mod lifecycle;
 pub mod read_only;
 mod read_ops;
@@ -12,6 +12,6 @@ pub struct MutableMapIndex<N: MapIndexKey + ?Sized>
 where
     Vec<<N as MapIndexKey>::Owned>: Blob + Send + Sync,
 {
-    pub(super) inner: MutableMapIndexInner<N>,
+    pub(super) in_memory_index: InMemoryMapIndex<N>,
     pub(super) storage: Gridstore<Vec<<N as MapIndexKey>::Owned>>,
 }

lib/segment/src/index/field_index/map_index/mutable_map_index/read_only/lifecycle.rs

@@ -6,7 +6,7 @@ use gridstore::error::GridstoreError;
 use gridstore::{Blob, GridstoreReader};
 
 use super::super::MapIndexKey;
-use super::super::inner::MutableMapIndexInner;
+use super::super::in_memory::InMemoryMapIndex;
 use super::ReadOnlyAppendableMapIndex;
 use crate::common::operation_error::OperationResult;
 
@@ -37,19 +37,20 @@ where
             return Ok(None);
         };
 
-        let mut inner = MutableMapIndexInner::<N>::empty();
+        let mut in_memory_index = InMemoryMapIndex::<N>::empty();
         let hw_counter = HardwareCounterCell::disposable();
         storage.iter::<_, GridstoreError>(
             storage.max_point_offset(),
             |idx, values: Vec<_>| {
-                for value in values {
-                    inner.ingest(idx, value);
-                }
+                in_memory_index.add_many_to_map(idx, values);
                 Ok(true)
             },
             hw_counter.ref_payload_index_io_write_counter(),
         )?;
 
-        Ok(Some(Self { inner, storage }))
+        Ok(Some(Self {
+            in_memory_index,
+            storage,
+        }))
     }
 }

lib/segment/src/index/field_index/map_index/mutable_map_index/read_only/live_reload.rs

@@ -25,10 +25,10 @@ where
     ) -> OperationResult<()> {
         self.storage.live_reload(fs)?;
 
-        let in_memory_storage = &mut self.inner;
+        let in_memory_index = &mut self.in_memory_index;
 
         for deleted_point in deleted_points {
-            in_memory_storage.remove_point(*deleted_point);
+            in_memory_index.remove_point(*deleted_point);
         }
 
         self.storage
@@ -37,9 +37,7 @@ where
                 new_points.iter().copied().enumerate(),
                 |_, point_offset, maybe_values: Option<Vec<_>>| {
                     let values = maybe_values.unwrap_or_default();
-                    for value in values {
-                        in_memory_storage.ingest(point_offset, value);
-                    }
+                    in_memory_index.add_many_to_map(point_offset, values);
                     Ok(())
                 },
                 hw_counter.payload_index_io_read_counter(),

lib/segment/src/index/field_index/map_index/mutable_map_index/read_only/mod.rs

@@ -2,7 +2,7 @@ use common::universal_io::UniversalRead;
 use gridstore::{Blob, GridstoreReader};
 
 use super::super::MapIndexKey;
-use super::inner::MutableMapIndexInner;
+use super::in_memory::InMemoryMapIndex;
 
 mod lifecycle;
 mod live_reload;
@@ -23,7 +23,7 @@ pub struct ReadOnlyAppendableMapIndex<N: MapIndexKey + ?Sized, S: UniversalRead>
 where
     Vec<<N as MapIndexKey>::Owned>: Blob + Send + Sync,
 {
-    pub(super) inner: MutableMapIndexInner<N>,
+    pub(super) in_memory_index: InMemoryMapIndex<N>,
     /// Backing Gridstore reader, populated by [`Self::open`]. Held to keep the
     /// storage mapped; the `files` / `populate` / `clear_cache` wiring that
     /// reads it lands with the parent dispatcher (it isn't part of the

lib/segment/src/index/field_index/map_index/mutable_map_index/read_only/read_ops.rs

@@ -21,7 +21,8 @@ where
         hw_counter: &HardwareCounterCell,
         check_fn: impl Fn(&N) -> bool,
     ) -> bool {
-        self.inner.check_values_any(idx, hw_counter, check_fn)
+        self.in_memory_index
+            .check_values_any(idx, hw_counter, check_fn)
     }
 
     fn get_values<'a>(
@@ -32,35 +33,35 @@ where
     where
         N: 'a,
     {
-        self.inner.get_values(idx, hw_counter)
+        self.in_memory_index.get_values(idx, hw_counter)
     }
 
     fn values_count(&self, idx: PointOffsetType) -> Option<usize> {
-        self.inner.values_count(idx)
+        self.in_memory_index.values_count(idx)
     }
 
     fn get_indexed_points(&self) -> usize {
-        self.inner.get_indexed_points()
+        self.in_memory_index.get_indexed_points()
     }
 
     fn get_values_count(&self) -> usize {
-        self.inner.get_values_count()
+        self.in_memory_index.get_values_count()
     }
 
     fn get_unique_values_count(&self) -> usize {
-        self.inner.get_unique_values_count()
+        self.in_memory_index.get_unique_values_count()
     }
 
     fn get_count_for_value(&self, value: &N, hw_counter: &HardwareCounterCell) -> Option<usize> {
-        self.inner.get_count_for_value(value, hw_counter)
+        self.in_memory_index.get_count_for_value(value, hw_counter)
     }
 
     fn get_iterator(&self, value: &N, hw_counter: &HardwareCounterCell) -> IdIter<'_> {
-        self.inner.get_iterator(value, hw_counter)
+        self.in_memory_index.get_iterator(value, hw_counter)
     }
 
     fn for_each_value(&self, f: impl FnMut(&N) -> OperationResult<()>) -> OperationResult<()> {
-        self.inner.for_each_value(f)
+        self.in_memory_index.for_each_value(f)
     }
 
     fn for_each_count_per_value(
@@ -68,7 +69,8 @@ where
         deferred_internal_id: Option<PointOffsetType>,
         f: impl FnMut(&N, usize) -> OperationResult<()>,
     ) -> OperationResult<()> {
-        self.inner.for_each_count_per_value(deferred_internal_id, f)
+        self.in_memory_index
+            .for_each_count_per_value(deferred_internal_id, f)
     }
 
     fn for_each_value_map(
@@ -76,7 +78,7 @@ where
         hw_counter: &HardwareCounterCell,
         f: impl FnMut(&N, &mut dyn Iterator<Item = PointOffsetType>) -> OperationResult<()>,
     ) -> OperationResult<()> {
-        self.inner.for_each_value_map(hw_counter, f)
+        self.in_memory_index.for_each_value_map(hw_counter, f)
     }
 
     fn storage_type(&self) -> StorageType {
@@ -86,7 +88,7 @@ where
     }
 
     fn ram_usage_bytes(&self) -> usize {
-        self.inner.ram_usage_bytes()
+        self.in_memory_index.ram_usage_bytes()
     }
 
     fn telemetry_index_type(&self) -> &'static str {
@@ -103,6 +105,6 @@ where
         points: impl Iterator<Item = PointOffsetType>,
         f: impl FnMut(PointOffsetType, &[<N as MapIndexKey>::Owned]),
     ) {
-        self.inner.for_points_values(points, f);
+        self.in_memory_index.for_points_values(points, f);
     }
 }

lib/segment/src/index/field_index/map_index/mutable_map_index/read_ops.rs

@@ -20,7 +20,8 @@ where
         hw_counter: &HardwareCounterCell,
         check_fn: impl Fn(&N) -> bool,
     ) -> bool {
-        self.inner.check_values_any(idx, hw_counter, check_fn)
+        self.in_memory_index
+            .check_values_any(idx, hw_counter, check_fn)
     }
 
     fn get_values<'a>(
@@ -31,35 +32,35 @@ where
     where
         N: 'a,
     {
-        self.inner.get_values(idx, hw_counter)
+        self.in_memory_index.get_values(idx, hw_counter)
     }
 
     fn values_count(&self, idx: PointOffsetType) -> Option<usize> {
-        self.inner.values_count(idx)
+        self.in_memory_index.values_count(idx)
     }
 
     fn get_indexed_points(&self) -> usize {
-        self.inner.get_indexed_points()
+        self.in_memory_index.get_indexed_points()
     }
 
     fn get_values_count(&self) -> usize {
-        self.inner.get_values_count()
+        self.in_memory_index.get_values_count()
     }
 
     fn get_unique_values_count(&self) -> usize {
-        self.inner.get_unique_values_count()
+        self.in_memory_index.get_unique_values_count()
     }
 
     fn get_count_for_value(&self, value: &N, hw_counter: &HardwareCounterCell) -> Option<usize> {
-        self.inner.get_count_for_value(value, hw_counter)
+        self.in_memory_index.get_count_for_value(value, hw_counter)
     }
 
     fn get_iterator(&self, value: &N, hw_counter: &HardwareCounterCell) -> IdIter<'_> {
-        self.inner.get_iterator(value, hw_counter)
+        self.in_memory_index.get_iterator(value, hw_counter)
     }
 
     fn for_each_value(&self, f: impl FnMut(&N) -> OperationResult<()>) -> OperationResult<()> {
-        self.inner.for_each_value(f)
+        self.in_memory_index.for_each_value(f)
     }
 
     fn for_each_count_per_value(
@@ -67,7 +68,8 @@ where
         deferred_internal_id: Option<PointOffsetType>,
         f: impl FnMut(&N, usize) -> OperationResult<()>,
     ) -> OperationResult<()> {
-        self.inner.for_each_count_per_value(deferred_internal_id, f)
+        self.in_memory_index
+            .for_each_count_per_value(deferred_internal_id, f)
     }
 
     fn for_each_value_map(
@@ -75,7 +77,7 @@ where
         hw_counter: &HardwareCounterCell,
         f: impl FnMut(&N, &mut dyn Iterator<Item = PointOffsetType>) -> OperationResult<()>,
     ) -> OperationResult<()> {
-        self.inner.for_each_value_map(hw_counter, f)
+        self.in_memory_index.for_each_value_map(hw_counter, f)
     }
 
     fn storage_type(&self) -> StorageType {
@@ -83,7 +85,7 @@ where
     }
 
     fn ram_usage_bytes(&self) -> usize {
-        self.inner.ram_usage_bytes()
+        self.in_memory_index.ram_usage_bytes()
     }
 
     fn telemetry_index_type(&self) -> &'static str {
@@ -100,6 +102,6 @@ where
         points: impl Iterator<Item = PointOffsetType>,
         f: impl FnMut(PointOffsetType, &[<N as MapIndexKey>::Owned]),
     ) {
-        self.inner.for_points_values(points, f);
+        self.in_memory_index.for_points_values(points, f);
     }
 }