upstream-archive/qdrant
1
0
Fork 0
mirror of https://github.com/qdrant/qdrant.git synced 2026-07-23 11:11:00 -05:00
Code Issues Packages Projects Releases Wiki Activity

feat/readonly map index with live reload (#9264)

Browse Source 
* live reload function for map index

* fmt
This commit is contained in:
Andrey Vasnetsov
2026-06-03 12:43:51 +02:00
committed by timvisee
parent 0905ef6575
commit cc8d1696a9
11 changed files with 130 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
lib/gridstore/src/gridstore/mod.rs
View File

@@ -8,6 +8,7 @@ use std::path::PathBuf;
use std::sync::Arc;
use ahash::AHashMap;
use common::counter::counter_cell::CounterCell;
use common::counter::hardware_counter::HardwareCounterCell;
use common::counter::referenced_counter::HwMetricRefCounter;
use common::fs::atomic_save_json;
@@ -380,14 +381,14 @@ impl<V: Blob> Gridstore<V> {
&self,
offsets: &[PointOffset],
callback: F,
hw_counter: &HardwareCounterCell,
hw_counter_cell: &CounterCell,
) -> std::result::Result<(), E>
where
P: AccessPattern,
F: FnMut(usize, Option<V>) -> std::result::Result<(), E>,
E: From<GridstoreError>,
{
self.with_view(|view| view.for_each_in_batch::<P, F, E>(offsets, callback, hw_counter))
self.with_view(|view| view.for_each_in_batch::<P, F, E>(offsets, callback, hw_counter_cell))
}
#[cfg(test)]

2
lib/gridstore/src/gridstore/tests.rs
View File

@@ -1380,7 +1380,7 @@ fn test_for_each_in_batch_congruent_with_get_value() {
batch_results[idx] = value;
Ok(())
},
&hw_counter,
hw_counter.payload_io_read_counter(),
)
.unwrap();

5
lib/gridstore/src/gridstore/view.rs
View File

@@ -1,6 +1,7 @@
use std::borrow::Cow;
use std::ops::ControlFlow;
use common::counter::counter_cell::CounterCell;
use common::counter::hardware_counter::HardwareCounterCell;
use common::counter::referenced_counter::HwMetricRefCounter;
use common::generic_consts::{AccessPattern, Sequential};
@@ -112,7 +113,7 @@ impl<'a, V: Blob, S: UniversalRead> GridstoreView<'a, V, S> {
&self,
point_offsets: &[PointOffset],
mut callback: F,
hw_counter: &HardwareCounterCell,
hw_counter_cell: &CounterCell,
) -> std::result::Result<(), E>
where
P: AccessPattern,
@@ -129,7 +130,7 @@ impl<'a, V: Blob, S: UniversalRead> GridstoreView<'a, V, S> {
self.pages
.read_batch_from_pages::<P, _, E>(pointers, self.config, |idx, raw_opt| {
let value = raw_opt.map(|raw| {
hw_counter.payload_io_read_counter().incr_delta(raw.len());
hw_counter_cell.incr_delta(raw.len());
let decompressed = self.decompress(raw);
V::from_bytes(&decompressed)
});

26
lib/segment/src/index/field_index/map_index/mutable_map_index/inner.rs
View File

@@ -68,6 +68,32 @@ where
self.map.entry(value).or_default().insert(idx);
}
/// Remove a point from the in-memory state, updating the value map and
/// counters accordingly.
///
/// Returns `true` if the point was within the indexed range; callers backed
/// by a store should delete the on-disk entry only in that case.
pub fn remove_point(&mut self, idx: PointOffsetType) -> bool {
if self.point_to_values.len() <= idx as usize {
return false;
}
let removed_values = std::mem::take(&mut self.point_to_values[idx as usize]);
if !removed_values.is_empty() {
self.indexed_points -= 1;
}
self.values_count -= removed_values.len();
for value in &removed_values {
if let Some(vals) = self.map.get_mut(value.borrow()) {
vals.remove(idx);
}
}
true
}
pub(in crate::index::field_index::map_index) fn for_points_values(
&self,
points: impl Iterator<Item = PointOffsetType>,

16
lib/segment/src/index/field_index/map_index/mutable_map_index/lifecycle.rs
View File

@@ -1,4 +1,3 @@
use std::borrow::Borrow;
use std::path::PathBuf;
use common::counter::hardware_counter::HardwareCounterCell;
@@ -120,23 +119,10 @@ where
}
pub fn remove_point(&mut self, idx: PointOffsetType) -> OperationResult<()> {
if self.inner.point_to_values.len() <= idx as usize {
if !self.inner.remove_point(idx) {
return Ok(());
}
let removed_values = std::mem::take(&mut self.inner.point_to_values[idx as usize]);
if !removed_values.is_empty() {
self.inner.indexed_points -= 1;
}
self.inner.values_count -= removed_values.len();
for value in &removed_values {
if let Some(vals) = self.inner.map.get_mut(value.borrow()) {
vals.remove(idx);
}
}
self.storage.delete_value(idx)?;
Ok(())

50
lib/segment/src/index/field_index/map_index/mutable_map_index/read_only/live_reload.rs Normal file
View File

@@ -0,0 +1,50 @@
use common::counter::hardware_counter::HardwareCounterCell;
use common::generic_consts::Random;
use common::types::PointOffsetType;
use common::universal_io::UniversalRead;
use gridstore::Blob;
use gridstore::error::GridstoreError;
use crate::common::operation_error::OperationResult;
use crate::index::field_index::map_index::MapIndexKey;
use crate::index::field_index::map_index::mutable_map_index::read_only::ReadOnlyAppendableMapIndex;
impl<N: MapIndexKey + ?Sized, S: UniversalRead> ReadOnlyAppendableMapIndex<N, S>
where
Vec<<N as MapIndexKey>::Owned>: Blob + Send + Sync,
{
pub fn live_reload(
&mut self,
fs: &S::Fs,
deleted_points: &[PointOffsetType],
new_points: &[PointOffsetType],
hw_counter: &HardwareCounterCell,
) -> OperationResult<()> {
self.storage.live_reload(fs)?;
let in_memory_storage = &mut self.inner;
for deleted_point in deleted_points {
in_memory_storage.remove_point(*deleted_point);
}
self.storage
.view()
.for_each_in_batch::<Random, _, GridstoreError>(
new_points,
|idx, maybe_values: Option<Vec<_>>| {
let Some(values) = maybe_values else {
return Ok(());
};
let point_offset = new_points[idx];
for value in values {
in_memory_storage.ingest(point_offset, value);
}
Ok(())
},
hw_counter.payload_index_io_read_counter(),
)?;
Ok(())
}
}

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

@@ -5,6 +5,7 @@ use super::super::MapIndexKey;
use super::inner::MutableMapIndexInner;
mod lifecycle;
mod live_reload;
mod read_ops;
/// Read-only counterpart to [`super::MutableMapIndex`].

24
lib/segment/src/index/field_index/map_index/universal_map_index/lifecycle.rs
View File

@@ -99,6 +99,18 @@ where
is_on_disk,
}))
}
/// Marks `idx` as deleted in the in-memory deletion bitvec.
///
/// Not persisted: on reopen, deletions must be re-supplied via the
/// `deleted_points` argument to [`Self::open`].
pub fn remove_point(&mut self, idx: PointOffsetType) {
let idx = idx as usize;
if idx < self.storage.deleted.len() && !self.storage.deleted.get_bit(idx).unwrap_or(true) {
self.storage.deleted.set(idx, true);
self.deleted_count += 1;
}
}
}
impl<N, S> UniversalMapIndex<N, S>
@@ -219,18 +231,6 @@ where
files
}
/// Marks `idx` as deleted in the in-memory deletion bitvec.
///
/// Not persisted: on reopen, deletions must be re-supplied via the
/// `deleted_points` argument to [`Self::open`].
pub fn remove_point(&mut self, idx: PointOffsetType) {
let idx = idx as usize;
if idx < self.storage.deleted.len() && !self.storage.deleted.get_bit(idx).unwrap_or(true) {
self.storage.deleted.set(idx, true);
self.deleted_count += 1;
}
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {

31
lib/segment/src/index/field_index/map_index/universal_map_index/live_reload.rs Normal file
View File

@@ -0,0 +1,31 @@
use common::counter::hardware_counter::HardwareCounterCell;
use common::persisted_hashmap::Key;
use common::types::PointOffsetType;
use common::universal_io::UniversalRead;
use crate::common::operation_error::OperationResult;
use crate::index::field_index::map_index::MapIndexKey;
use crate::index::field_index::map_index::universal_map_index::UniversalMapIndex;
impl<N, S> UniversalMapIndex<N, S>
where
N: MapIndexKey + Key + ?Sized,
S: UniversalRead,
{
pub fn live_reload(
&mut self,
_fs: &S::Fs,
deleted_points: &[PointOffsetType],
_new_points: &[PointOffsetType],
_hw_counter: &HardwareCounterCell,
) -> OperationResult<()> {
// No on-disk state is changing when we live-reload, as
// this UniversalMapIndex is not mutable.
// We only patch in-memory deleted bitslice representation.
for deleted_point in deleted_points {
self.remove_point(*deleted_point)
}
Ok(())
}
}

1
lib/segment/src/index/field_index/map_index/universal_map_index/mod.rs
View File

@@ -10,6 +10,7 @@ use super::MapIndexKey;
use crate::index::field_index::stored_point_to_values::StoredPointToValues;
mod lifecycle;
mod live_reload;
mod read_ops;
pub(super) const DELETED_PATH: &str = "deleted.bin";

2
lib/segment/src/vector_storage/sparse/mmap_sparse_vector_storage.rs
View File

@@ -218,7 +218,7 @@ impl SparseVectorStorage for MmapSparseVectorStorage {
}
Ok(())
},
&HardwareCounterCell::disposable(),
HardwareCounterCell::disposable().vector_io_read(),
)
}
}