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

disk cache hygiene (#6323)

Browse Source 
* wip: implement explicit populate and clear_cache functions for all components

* fmt

* implement clear and populate for vector storages

* fmt

* implement clear and populate for payload storage

* wip: implement explicit populate and clear_cache functions payload indexes

* implement explicit populate and clear_cache functions payload indexes

* fix clippy on CI

* only compile posix_fadvise on linux

* only compile posix_fadvise on linux

* implement explicit populate and clear_cache functions for quantized vectors

* fmt

* remove post-load prefault

* fix typo

* implement is-on-disk for payload indexes, implement clear on drop for segment, implement clear after segment build

* fmt

* also evict quantized vectors after optimization

* re-use and replace advise_dontneed
This commit is contained in:
Andrey Vasnetsov
2025-04-09 10:54:30 +02:00
committed by GitHub
parent c11be87d7a
commit 0266508736
65 changed files with 1052 additions and 250 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.dockerignore
View File

@@ -1,6 +1,7 @@
tests/
target/
storage/
qdrant-storage/
snapshots/
lib/segment/target
lib/api/target

3
Cargo.lock generated
View File

@@ -1241,6 +1241,7 @@ dependencies = [
"lazy_static",
"log",
"memmap2",
"memory",
"num-traits",
"num_cpus",
"ordered-float 5.0.0",
@@ -3054,7 +3055,6 @@ version = "0.0.0"
dependencies = [
"atomicwrites",
"bincode",
"nix 0.29.0",
"semver",
"serde",
"serde_json",
@@ -3826,6 +3826,7 @@ dependencies = [
"bitvec",
"log",
"memmap2",
"nix 0.29.0",
"parking_lot",
"rand 0.9.0",
"serde",

1
Cargo.toml
View File

@@ -222,6 +222,7 @@ zerocopy = { version = "0.8.24", features = ["derive"] }
atomic_refcell = "0.1.13"
byteorder = "1.5.0"
thiserror = "2.0.12"
libc = "0.2"
bitvec = "1.0.1"
merge = "0.1.0"
dashmap = "6.1"

2
lib/collection/src/collection_manager/optimizers/segment_optimizer.rs
View File

@@ -735,8 +735,6 @@ pub trait SegmentOptimizer {
.unwrap();
}
optimized_segment.prefault_mmap_pages();
let point_count = optimized_segment.available_point_count();
let (_, proxies) = write_segments_guard.swap_new(optimized_segment, &proxy_ids);

65
lib/collection/src/shards/local_shard/mod.rs
View File

@@ -8,7 +8,6 @@ pub(super) mod search;
pub(super) mod shard_ops;
use std::collections::{BTreeSet, HashMap};
use std::mem::size_of;
use std::ops::Deref;
use std::path::{Path, PathBuf};
use std::sync::Arc;
@@ -27,16 +26,14 @@ use indicatif::{ProgressBar, ProgressStyle};
use itertools::Itertools;
use parking_lot::{Mutex as ParkingMutex, RwLock};
use segment::data_types::segment_manifest::SegmentManifests;
use segment::data_types::vectors::VectorElementType;
use segment::entry::entry_point::SegmentEntry as _;
use segment::index::field_index::CardinalityEstimation;
use segment::segment::Segment;
use segment::segment_constructor::{build_segment, load_segment};
use segment::types::{
CompressionRatio, Filter, PayloadIndexInfo, PayloadKeyType, PointIdType, QuantizationConfig,
SegmentConfig, SegmentType, SnapshotFormat,
Filter, PayloadIndexInfo, PayloadKeyType, PointIdType, SegmentConfig, SegmentType,
SnapshotFormat,
};
use segment::utils::mem::Mem;
use segment::vector_storage::common::get_async_scorer;
use tokio::fs::{create_dir_all, remove_dir_all, remove_file};
use tokio::runtime::Handle;
@@ -434,27 +431,6 @@ impl LocalShard {
// Apply outstanding operations from WAL
local_shard.load_from_wal(collection_id).await?;
let available_memory_bytes = Mem::new().available_memory_bytes() as usize;
let vectors_size_bytes = local_shard.estimate_vector_data_size().await;
// Simple heuristic to exclude mmap prefaulting for collections that won't benefit from it.
//
// We assume that mmap prefaulting is beneficial if we can put significant part of data
// into RAM in advance. However, if we can see that the data is too big to fit into RAM,
// it is better to avoid prefaulting, because it will only cause extra disk IO.
//
// This heuristic is not perfect, but it exclude cases when we don't have enough RAM
// even to store half of the vector data.
let do_mmap_prefault = available_memory_bytes * 2 > vectors_size_bytes;
if do_mmap_prefault {
for (_, segment) in local_shard.segments.read().iter() {
if let LockedSegment::Original(segment) = segment {
segment.read().prefault_mmap_pages();
}
}
}
Ok(local_shard)
}
@@ -1082,43 +1058,6 @@ impl LocalShard {
}
}
/// Returns estimated size of vector data in bytes
async fn estimate_vector_data_size(&self) -> usize {
let info = self.local_shard_info().await;
let vector_size: usize = info
.config
.params
.vectors
.params_iter()
.map(|(_, value)| {
let vector_size = value.size.get() as usize;
let quantization_config = value
.quantization_config
.as_ref()
.or(info.config.quantization_config.as_ref());
let quantized_size_bytes = match quantization_config {
None => 0,
Some(QuantizationConfig::Scalar(_)) => vector_size,
Some(QuantizationConfig::Product(pq)) => match pq.product.compression {
CompressionRatio::X4 => vector_size,
CompressionRatio::X8 => vector_size / 2,
CompressionRatio::X16 => vector_size / 4,
CompressionRatio::X32 => vector_size / 8,
CompressionRatio::X64 => vector_size / 16,
},
Some(QuantizationConfig::Binary(_)) => vector_size / 8,
};
vector_size * size_of::<VectorElementType>() + quantized_size_bytes
})
.sum();
vector_size * info.points_count
}
pub async fn local_shard_status(&self) -> (ShardStatus, OptimizersStatus) {
{
let segments = self.segments().read();

2
lib/common/common/Cargo.toml
View File

@@ -36,6 +36,8 @@ thiserror = { workspace = true }
zerocopy = { workspace = true }
log = { workspace = true }
memory = { path = "../memory" }
[dev-dependencies]
common = { path = ".", features = ["testing"] }
criterion = { workspace = true }

2
lib/common/common/benches/mmap_hashmap.rs
View File

@@ -18,7 +18,7 @@ fn bench_mmap_hashmap(c: &mut Criterion) {
)
.unwrap();
let mmap = MmapHashMap::<str, u32>::open(&mmap_path).unwrap();
let mmap = MmapHashMap::<str, u32>::open(&mmap_path, true).unwrap();
let mut it = keys.iter().cycle();
c.bench_function("get", |b| {

22
lib/common/common/src/mmap_hashmap.rs
View File

@@ -1,6 +1,5 @@
#[cfg(any(test, feature = "testing"))]
use std::collections::{BTreeMap, BTreeSet};
use std::fs::File;
use std::hash::Hash;
use std::io::{self, Cursor, Write};
use std::marker::PhantomData;
@@ -9,6 +8,8 @@ use std::path::Path;
use std::str;
use memmap2::Mmap;
use memory::madvise::{AdviceSetting, Madviseable};
use memory::mmap_ops::open_read_mmap;
use ph::fmph::Function;
#[cfg(any(test, feature = "testing"))]
use rand::Rng as _;
@@ -193,12 +194,8 @@ impl<K: Key + ?Sized, V: Sized + FromBytes + Immutable + IntoBytes + KnownLayout
}
/// Load the hash map from file.
pub fn open(path: &Path) -> io::Result<Self> {
let file = File::open(path)?;
// SAFETY: Assume other processes do not modify the file.
// See https://docs.rs/memmap2/latest/memmap2/struct.Mmap.html#safety
let mmap = unsafe { Mmap::map(&file)? };
pub fn open(path: &Path, populate: bool) -> io::Result<Self> {
let mmap = open_read_mmap(path, AdviceSetting::Global, populate)?;
let (header, _) =
Header::read_from_prefix(mmap.as_ref()).map_err(|_| io::ErrorKind::InvalidData)?;
@@ -339,6 +336,13 @@ impl<K: Key + ?Sized, V: Sized + FromBytes + Immutable + IntoBytes + KnownLayout
)
})
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> io::Result<()> {
self.mmap.populate();
Ok(())
}
}
/// A key that can be stored in the hash map.
@@ -514,7 +518,7 @@ mod tests {
map.iter().map(|(k, v)| (as_ref(k), v.iter().copied())),
)
.unwrap();
let mmap = MmapHashMap::<K, u32>::open(&tmpdir.path().join("map")).unwrap();
let mmap = MmapHashMap::<K, u32>::open(&tmpdir.path().join("map"), false).unwrap();
// Non-existing keys should return None
for _ in 0..1000 {
@@ -561,7 +565,7 @@ mod tests {
)
.unwrap();
let mmap = MmapHashMap::<i64, u64>::open(&tmpdir.path().join("map")).unwrap();
let mmap = MmapHashMap::<i64, u64>::open(&tmpdir.path().join("map"), true).unwrap();
for (k, v) in map {
assert_eq!(

1
lib/common/io/Cargo.toml
View File

@@ -15,7 +15,6 @@ workspace = true
[dependencies]
atomicwrites = { workspace = true }
bincode = { workspace = true }
nix = { workspace = true }
semver = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }

29
lib/common/io/src/file_operations.rs
View File

@@ -31,35 +31,6 @@ pub fn read_json<T: DeserializeOwned>(path: &Path) -> Result<T> {
Ok(value)
}
/// Advise the operating system that the file is no longer needed to be in the page cache.
pub fn advise_dontneed(path: &Path) -> Result<()> {
// https://github.com/nix-rust/nix/blob/v0.29.0/src/fcntl.rs#L35-L42
#[cfg(any(
target_os = "linux",
target_os = "freebsd",
target_os = "android",
target_os = "fuchsia",
target_os = "emscripten",
target_os = "wasi",
target_env = "uclibc",
))]
{
use std::os::fd::AsRawFd as _;
use nix::fcntl;
let file = File::open(path)?;
let fd = file.as_raw_fd();
fcntl::posix_fadvise(fd, 0, 0, fcntl::PosixFadviseAdvice::POSIX_FADV_DONTNEED)
.map_err(io::Error::from)?;
}
_ = path;
Ok(())
}
pub type FileOperationResult<T> = Result<T>;
pub type FileStorageError = Error;

1
lib/common/memory/Cargo.toml
View File

@@ -19,6 +19,7 @@ parking_lot = { workspace = true }
serde = { workspace = true }
bitvec = { workspace = true }
thiserror = { workspace = true }
nix = { workspace = true }
[dev-dependencies]
rand = { workspace = true }

5
lib/common/memory/src/chunked_utils.rs
View File

@@ -1,4 +1,5 @@
use std::collections::HashMap;
use std::io;
use std::path::{Path, PathBuf};
use crate::madvise::{Advice, AdviceSetting};
@@ -40,6 +41,10 @@ impl<T: Sized + 'static> UniversalMmapChunk<T> {
pub fn is_empty(&self) -> bool {
self.mmap.is_empty()
}
pub fn populate(&self) -> io::Result<()> {
self.mmap.populate()
}
}
/// Checks if the file name matches the pattern for mmap chunks

33
lib/common/memory/src/madvise.rs
View File

@@ -1,9 +1,11 @@
//! Platform-independent abstractions over [`memmap2::Mmap::advise`]/[`memmap2::MmapMut::advise`]
//! and [`memmap2::Advice`].
use std::fs::File;
use std::hint::black_box;
use std::io;
use std::num::Wrapping;
use std::path::Path;
use serde::Deserialize;
@@ -169,3 +171,34 @@ fn populate_simple(slice: &[u8]) {
.sum::<Wrapping<u8>>(),
);
}
/// For given file path, clear disk cache with `posix_fadvise`
///
/// If `posix_fadvise` is not supported, this function does nothing.
pub fn clear_disk_cache(file_path: &Path) -> io::Result<()> {
// https://github.com/nix-rust/nix/blob/v0.29.0/src/fcntl.rs#L35-L42
#[cfg(any(
target_os = "linux",
target_os = "freebsd",
target_os = "android",
target_os = "fuchsia",
target_os = "emscripten",
target_os = "wasi",
target_env = "uclibc",
))]
{
use std::os::fd::AsRawFd as _;
use nix::fcntl;
let file = File::open(file_path)?;
let fd = file.as_raw_fd();
fcntl::posix_fadvise(fd, 0, 0, fcntl::PosixFadviseAdvice::POSIX_FADV_DONTNEED)
.map_err(io::Error::from)?;
}
_ = file_path;
Ok(())
}

43
lib/common/memory/src/mmap_ops.rs
View File

@@ -1,8 +1,7 @@
use std::fs::{File, OpenOptions};
use std::mem::{align_of, size_of};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::{io, mem, ops, ptr, time};
use std::path::Path;
use std::{io, mem, ptr};
use memmap2::{Mmap, MmapMut};
@@ -77,44 +76,6 @@ pub fn open_write_mmap(path: &Path, advice: AdviceSetting, populate: bool) -> io
Ok(mmap)
}
#[derive(Clone, Debug)]
pub struct PrefaultMmapPages {
mmap: Arc<Mmap>,
path: Option<PathBuf>,
}
impl PrefaultMmapPages {
pub fn new(mmap: Arc<Mmap>, path: Option<impl Into<PathBuf>>) -> Self {
Self {
mmap,
path: path.map(Into::into),
}
}
pub fn exec(&self) {
prefault_mmap_pages(self.mmap.as_ref(), self.path.as_deref());
}
}
fn prefault_mmap_pages<T>(mmap: &T, path: Option<&Path>)
where
T: Madviseable + ops::Deref<Target = [u8]>,
{
let separator = path.map_or("", |_| " "); // space if `path` is `Some` or nothing
let path = path.unwrap_or(Path::new("")); // path if `path` is `Some` or nothing
log::trace!("Reading mmap{separator}{path:?} to populate cache...");
let instant = time::Instant::now();
mmap.populate();
log::trace!(
"Reading mmap{separator}{path:?} to populate cache took {:?}",
instant.elapsed()
);
}
pub fn transmute_from_u8<T>(v: &[u8]) -> &T {
debug_assert_eq!(v.len(), size_of::<T>());

21
lib/common/memory/src/mmap_type.rs
View File

@@ -30,7 +30,7 @@ use std::{fmt, mem, slice};
use bitvec::slice::BitSlice;
use memmap2::MmapMut;
use crate::madvise::{Advice, AdviceSetting};
use crate::madvise::{Advice, AdviceSetting, Madviseable};
use crate::mmap_ops;
/// Result for mmap errors.
@@ -182,6 +182,11 @@ where
pub unsafe fn unchecked_advise(&self, advice: memmap2::UncheckedAdvice) -> std::io::Result<()> {
unsafe { self.mmap.unchecked_advise(advice) }
}
pub fn populate(&self) -> std::io::Result<()> {
self.mmap.populate();
Ok(())
}
}
impl<T> Deref for MmapType<T>
@@ -293,6 +298,13 @@ impl<T> MmapSlice<T> {
Ok(())
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> std::io::Result<()> {
self.mmap.populate()?;
Ok(())
}
}
impl<T> Deref for MmapSlice<T> {
@@ -392,6 +404,13 @@ impl MmapBitSlice {
Ok(())
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> std::io::Result<()> {
self.mmap.populate()?;
Ok(())
}
}
impl Deref for MmapBitSlice {

15
lib/gridstore/src/bitmask/gaps.rs
View File

@@ -2,7 +2,7 @@ use std::ops::Range;
use std::path::{Path, PathBuf};
use itertools::Itertools;
use memory::madvise::{Advice, AdviceSetting};
use memory::madvise::{Advice, AdviceSetting, clear_disk_cache};
use memory::mmap_ops::{create_and_ensure_length, open_write_mmap};
use memory::mmap_type::{self, MmapSlice};
@@ -253,6 +253,19 @@ impl BitmaskGaps {
}
})
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> std::io::Result<()> {
self.mmap_slice.populate()?;
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> std::io::Result<()> {
clear_disk_cache(&self.path)?;
Ok(())
}
}
#[cfg(test)]

17
lib/gridstore/src/bitmask/mod.rs
View File

@@ -6,7 +6,7 @@ use std::path::{Path, PathBuf};
use bitvec::slice::BitSlice;
use gaps::{BitmaskGaps, RegionGaps};
use itertools::Itertools;
use memory::madvise::{Advice, AdviceSetting};
use memory::madvise::{Advice, AdviceSetting, clear_disk_cache};
use memory::mmap_ops::{create_and_ensure_length, open_write_mmap};
use memory::mmap_type::{self, MmapBitSlice};
@@ -500,6 +500,21 @@ impl Bitmask {
RegionGaps::new(leading as u16, trailing as u16, max as u16)
}
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> std::io::Result<()> {
self.bitslice.populate()?;
self.regions_gaps.populate()?;
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> std::io::Result<()> {
clear_disk_cache(&self.path)?;
self.regions_gaps.clear_cache()?;
Ok(())
}
}
#[cfg(test)]

19
lib/gridstore/src/gridstore.rs
View File

@@ -527,6 +527,25 @@ impl<V> Gridstore<V> {
Ok(())
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> std::io::Result<()> {
for page in &self.pages {
page.populate();
}
self.bitmask.read().populate()?;
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> std::io::Result<()> {
for page in &self.pages {
page.clear_cache()?;
}
self.bitmask.read().clear_cache()?;
Ok(())
}
}
#[cfg(test)]

14
lib/gridstore/src/page.rs
View File

@@ -1,7 +1,7 @@
use std::path::{Path, PathBuf};
use memmap2::MmapMut;
use memory::madvise::{Advice, AdviceSetting};
use memory::madvise::{Advice, AdviceSetting, Madviseable, clear_disk_cache};
use memory::mmap_ops::{create_and_ensure_length, open_write_mmap};
use crate::tracker::BlockOffset;
@@ -112,4 +112,16 @@ impl Page {
drop(self.mmap);
std::fs::remove_file(&self.path).unwrap();
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) {
self.mmap.populate();
}
/// Drop disk cache.
pub fn clear_cache(&self) -> std::io::Result<()> {
clear_disk_cache(&self.path)?;
Ok(())
}
}

4
lib/quantization/src/encoded_vectors_binary.rs
View File

@@ -167,6 +167,10 @@ impl BitsStoreType for u128 {
impl<TBitsStoreType: BitsStoreType, TStorage: EncodedStorage>
EncodedVectorsBin<TBitsStoreType, TStorage>
{
pub fn storage(&self) -> &TStorage {
&self.encoded_vectors
}
pub fn encode<'a>(
orig_data: impl Iterator<Item = impl AsRef<[f32]> + 'a> + Clone,
mut storage_builder: impl EncodedStorageBuilder<TStorage>,

4
lib/quantization/src/encoded_vectors_pq.rs
View File

@@ -42,6 +42,10 @@ pub struct Metadata {
}
impl<TStorage: EncodedStorage> EncodedVectorsPQ<TStorage> {
pub fn storage(&self) -> &TStorage {
&self.encoded_vectors
}
/// Encode vector data using product quantization.
///
/// # Arguments

4
lib/quantization/src/encoded_vectors_u8.rs
View File

@@ -34,6 +34,10 @@ struct Metadata {
}
impl<TStorage: EncodedStorage> EncodedVectorsU8<TStorage> {
pub fn storage(&self) -> &TStorage {
&self.encoded_vectors
}
pub fn encode<'a>(
orig_data: impl Iterator<Item = impl AsRef<[f32]> + 'a> + Clone,
mut storage_builder: impl EncodedStorageBuilder<TStorage>,

1
lib/segment/benches/multi_vector_search.rs
View File

@@ -129,6 +129,7 @@ fn make_segment_index<R: Rng + ?Sized>(rnd: &mut R, distance: Distance) -> HNSWI
},
)
.unwrap();
hnsw_index.populate().unwrap();
hnsw_index
}

21
lib/segment/src/index/field_index/bool_index/mmap_bool_index.rs
View File

@@ -280,6 +280,27 @@ impl MmapBoolIndex {
.flatten()
.collect()
}
pub fn is_on_disk(&self) -> bool {
!self.populated
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
self.trues_slice.populate()?;
self.falses_slice.populate()?;
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
self.trues_slice.clear_cache()?;
self.falses_slice.clear_cache()?;
Ok(())
}
}
/// Set or insert a flag in the given flags. Returns previous value.

27
lib/segment/src/index/field_index/bool_index/mod.rs
View File

@@ -6,6 +6,7 @@ use simple_bool_index::SimpleBoolIndex;
use super::facet_index::FacetIndex;
use super::map_index::IdIter;
use super::{PayloadFieldIndex, ValueIndexer};
use crate::common::operation_error::OperationResult;
use crate::data_types::facets::{FacetHit, FacetValueRef};
use crate::telemetry::PayloadIndexTelemetry;
@@ -81,6 +82,32 @@ impl BoolIndex {
BoolIndex::Mmap(index) => index.values_is_empty(point_id),
}
}
pub fn is_on_disk(&self) -> bool {
match self {
BoolIndex::Simple(_) => false,
BoolIndex::Mmap(index) => index.is_on_disk(),
}
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
match self {
BoolIndex::Simple(_) => {} // Not a mmap
BoolIndex::Mmap(index) => index.populate()?,
}
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
match self {
BoolIndex::Simple(_) => {} // Not a mmap
BoolIndex::Mmap(index) => index.clear_cache()?,
}
Ok(())
}
}
impl PayloadFieldIndex for BoolIndex {

51
lib/segment/src/index/field_index/field_index_base.rs
View File

@@ -419,6 +419,57 @@ impl FieldIndex {
| FieldIndex::NullIndex(_) => None,
}
}
pub fn is_on_disk(&self) -> bool {
match self {
FieldIndex::IntIndex(index) => index.is_on_disk(),
FieldIndex::DatetimeIndex(index) => index.is_on_disk(),
FieldIndex::IntMapIndex(index) => index.is_on_disk(),
FieldIndex::KeywordIndex(index) => index.is_on_disk(),
FieldIndex::FloatIndex(index) => index.is_on_disk(),
FieldIndex::GeoIndex(index) => index.is_on_disk(),
FieldIndex::BoolIndex(index) => index.is_on_disk(),
FieldIndex::FullTextIndex(index) => index.is_on_disk(),
FieldIndex::UuidIndex(index) => index.is_on_disk(),
FieldIndex::UuidMapIndex(index) => index.is_on_disk(),
FieldIndex::NullIndex(index) => index.is_on_disk(),
}
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
match self {
FieldIndex::IntIndex(index) => index.populate(),
FieldIndex::DatetimeIndex(index) => index.populate(),
FieldIndex::IntMapIndex(index) => index.populate(),
FieldIndex::KeywordIndex(index) => index.populate(),
FieldIndex::FloatIndex(index) => index.populate(),
FieldIndex::GeoIndex(index) => index.populate(),
FieldIndex::BoolIndex(index) => index.populate(),
FieldIndex::FullTextIndex(index) => index.populate(),
FieldIndex::UuidIndex(index) => index.populate(),
FieldIndex::UuidMapIndex(index) => index.populate(),
FieldIndex::NullIndex(index) => index.populate(),
}
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
match self {
FieldIndex::IntIndex(index) => index.clear_cache(),
FieldIndex::DatetimeIndex(index) => index.clear_cache(),
FieldIndex::IntMapIndex(index) => index.clear_cache(),
FieldIndex::KeywordIndex(index) => index.clear_cache(),
FieldIndex::FloatIndex(index) => index.clear_cache(),
FieldIndex::GeoIndex(index) => index.clear_cache(),
FieldIndex::BoolIndex(index) => index.clear_cache(),
FieldIndex::FullTextIndex(index) => index.clear_cache(),
FieldIndex::UuidIndex(index) => index.clear_cache(),
FieldIndex::UuidMapIndex(index) => index.clear_cache(),
FieldIndex::NullIndex(index) => index.clear_cache(),
}
}
}
/// Common interface for all index builders.

8
lib/segment/src/index/field_index/full_text_index/mmap_inverted_index/mmap_postings.rs
View File

@@ -7,7 +7,7 @@ use common::counter::hardware_counter::HardwareCounterCell;
use common::types::PointOffsetType;
use common::zeros::WriteZerosExt;
use memmap2::Mmap;
use memory::madvise::{Advice, AdviceSetting};
use memory::madvise::{Advice, AdviceSetting, Madviseable};
use memory::mmap_ops::open_read_mmap;
use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout};
@@ -234,4 +234,10 @@ impl MmapPostings {
on_disk: !populate,
})
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) {
self.mmap.populate();
}
}

27
lib/segment/src/index/field_index/full_text_index/mmap_inverted_index/mod.rs
View File

@@ -5,7 +5,7 @@ use bitvec::vec::BitVec;
use common::counter::hardware_counter::HardwareCounterCell;
use common::mmap_hashmap::{MmapHashMap, READ_ENTRY_OVERHEAD};
use common::types::PointOffsetType;
use memory::madvise::AdviceSetting;
use memory::madvise::{AdviceSetting, clear_disk_cache};
use memory::mmap_ops;
use memory::mmap_type::{MmapBitSlice, MmapSlice};
use mmap_postings::MmapPostings;
@@ -87,7 +87,7 @@ impl MmapInvertedIndex {
let deleted_points_path = path.join(DELETED_POINTS_FILE);
let postings = MmapPostings::open(&postings_path, populate)?;
let vocab = MmapHashMap::<str, TokenId>::open(&vocab_path)?;
let vocab = MmapHashMap::<str, TokenId>::open(&vocab_path, false)?;
let point_to_tokens_count = unsafe {
MmapSlice::try_from(mmap_ops::open_write_mmap(
@@ -136,6 +136,29 @@ impl MmapInvertedIndex {
self.path.join(DELETED_POINTS_FILE),
]
}
pub fn is_on_disk(&self) -> bool {
self.is_on_disk
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
self.postings.populate();
self.vocab.populate()?;
self.point_to_tokens_count.populate()?;
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
let files = self.files();
for file in files {
clear_disk_cache(&file)?;
}
Ok(())
}
}
impl InvertedIndex for MmapInvertedIndex {

17
lib/segment/src/index/field_index/full_text_index/mmap_text_index.rs
View File

@@ -60,6 +60,23 @@ impl MmapFullTextIndex {
pub fn flusher(&self) -> Flusher {
self.inverted_index.deleted_points.flusher()
}
pub fn is_on_disk(&self) -> bool {
self.inverted_index.is_on_disk()
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
self.inverted_index.populate()?;
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
self.inverted_index.clear_cache()?;
Ok(())
}
}
pub struct FullTextMmapIndexBuilder {

29
lib/segment/src/index/field_index/full_text_index/text_index.rs
View File

@@ -260,6 +260,35 @@ impl FullTextIndex {
let parsed_query = self.parse_query(query, hw_counter);
self.filter(parsed_query, hw_counter)
}
pub fn is_on_disk(&self) -> bool {
match self {
FullTextIndex::Mutable(_) => false,
FullTextIndex::Immutable(_) => false,
FullTextIndex::Mmap(index) => index.is_on_disk(),
}
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
match self {
FullTextIndex::Mutable(_) => {} // Not a mmap
FullTextIndex::Immutable(_) => {} // Not a mmap
FullTextIndex::Mmap(index) => index.populate()?,
}
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
match self {
FullTextIndex::Mutable(_) => {} // Not a mmap
FullTextIndex::Immutable(_) => {} // Not a mmap
FullTextIndex::Mmap(index) => index.clear_cache()?,
}
Ok(())
}
}
pub struct FullTextIndexBuilder(FullTextIndex);

35
lib/segment/src/index/field_index/geo_index/mmap_geo_index.rs
View File

@@ -6,7 +6,7 @@ use common::counter::hardware_counter::HardwareCounterCell;
use common::types::PointOffsetType;
use io::file_operations::{atomic_save_json, read_json};
use memmap2::MmapMut;
use memory::madvise::AdviceSetting;
use memory::madvise::{AdviceSetting, clear_disk_cache};
use memory::mmap_ops::{create_and_ensure_length, open_write_mmap};
use memory::mmap_type::{MmapBitSlice, MmapSlice};
use serde::{Deserialize, Serialize};
@@ -224,7 +224,7 @@ impl MmapGeoMapIndex {
populate,
)?)?
};
let point_to_values = MmapPointToValues::open(path)?;
let point_to_values = MmapPointToValues::open(path, true)?;
let deleted = open_write_mmap(&deleted_path, AdviceSetting::Global, populate)?;
let deleted = MmapBitSlice::from(deleted, 0);
@@ -397,4 +397,35 @@ impl MmapGeoMapIndex {
) -> ConditionedCounter<'a> {
ConditionedCounter::new(self.is_on_disk, hw_counter)
}
pub fn is_on_disk(&self) -> bool {
self.is_on_disk
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
self.counts_per_hash.populate()?;
self.points_map.populate()?;
self.points_map_ids.populate()?;
self.point_to_values.populate();
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
let deleted_path = self.path.join(DELETED_PATH);
let counts_per_hash_path = self.path.join(COUNTS_PER_HASH);
let points_map_path = self.path.join(POINTS_MAP);
let points_map_ids_path = self.path.join(POINTS_MAP_IDS);
clear_disk_cache(&deleted_path)?;
clear_disk_cache(&counts_per_hash_path)?;
clear_disk_cache(&points_map_path)?;
clear_disk_cache(&points_map_ids_path)?;
self.point_to_values.clear_cache()?;
Ok(())
}
}

29
lib/segment/src/index/field_index/geo_index/mod.rs
View File

@@ -342,6 +342,35 @@ impl GeoMapIndex {
pub fn values_is_empty(&self, idx: PointOffsetType) -> bool {
self.values_count(idx) == 0
}
pub fn is_on_disk(&self) -> bool {
match self {
GeoMapIndex::Mutable(_) => false,
GeoMapIndex::Immutable(_) => false,
GeoMapIndex::Mmap(index) => index.is_on_disk(),
}
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
match self {
GeoMapIndex::Mutable(_) => {} // Not a mmap
GeoMapIndex::Immutable(_) => {} // Not a mmap
GeoMapIndex::Mmap(index) => index.populate()?,
}
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
match self {
GeoMapIndex::Mutable(_) => {} // Not a mmap
GeoMapIndex::Immutable(_) => {} // Not a mmap
GeoMapIndex::Mmap(index) => index.clear_cache()?,
}
Ok(())
}
}
pub struct GeoMapIndexBuilder(GeoMapIndex);

33
lib/segment/src/index/field_index/map_index/mmap_map_index.rs
View File

@@ -13,7 +13,7 @@ use common::types::PointOffsetType;
use io::file_operations::{atomic_save_json, read_json};
use itertools::Itertools;
use memmap2::MmapMut;
use memory::madvise::AdviceSetting;
use memory::madvise::{AdviceSetting, clear_disk_cache};
use memory::mmap_ops::{self, create_and_ensure_length};
use memory::mmap_type::MmapBitSlice;
use serde::{Deserialize, Serialize};
@@ -51,10 +51,11 @@ impl<N: MapIndexKey + Key + ?Sized> MmapMapIndex<N> {
let config: MmapMapIndexConfig = read_json(&config_path)?;
let hashmap = MmapHashMap::open(&hashmap_path)?;
let point_to_values = MmapPointToValues::open(path)?;
let do_populate = !is_on_disk;
let hashmap = MmapHashMap::open(&hashmap_path, do_populate)?;
let point_to_values = MmapPointToValues::open(path, do_populate)?;
let deleted = mmap_ops::open_write_mmap(&deleted_path, AdviceSetting::Global, do_populate)?;
let deleted = MmapBitSlice::from(deleted, 0);
let deleted_count = deleted.count_ones();
@@ -332,4 +333,28 @@ impl<N: MapIndexKey + Key + ?Sized> MmapMapIndex<N> {
) -> ConditionedCounter<'a> {
ConditionedCounter::new(self.is_on_disk, hw_counter)
}
pub fn is_on_disk(&self) -> bool {
self.is_on_disk
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
self.value_to_points.populate()?;
self.point_to_values.populate();
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
let value_to_points_path = self.path.join(HASHMAP_PATH);
let deleted_path = self.path.join(DELETED_PATH);
clear_disk_cache(&value_to_points_path)?;
clear_disk_cache(&deleted_path)?;
self.point_to_values.clear_cache()?;
Ok(())
}
}

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

@@ -434,6 +434,35 @@ impl<N: MapIndexKey + ?Sized> MapIndex<N> {
.unique(),
)
}
pub fn is_on_disk(&self) -> bool {
match self {
MapIndex::Mutable(_) => false,
MapIndex::Immutable(_) => false,
MapIndex::Mmap(index) => index.is_on_disk(),
}
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
match self {
MapIndex::Mutable(_) => {} // Not a mmap
MapIndex::Immutable(_) => {} // Not a mmap
MapIndex::Mmap(index) => index.populate()?,
}
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
match self {
MapIndex::Mutable(_) => {} // Not a mmap
MapIndex::Immutable(_) => {} // Not a mmap
MapIndex::Mmap(index) => index.clear_cache()?,
}
Ok(())
}
}
pub struct MapIndexBuilder<N: MapIndexKey + ?Sized>(MapIndex<N>);

22
lib/segment/src/index/field_index/mmap_point_to_values.rs
View File

@@ -3,7 +3,7 @@ use std::path::{Path, PathBuf};
use common::counter::hardware_counter::HardwareCounterCell;
use common::types::PointOffsetType;
use memmap2::Mmap;
use memory::madvise::AdviceSetting;
use memory::madvise::{AdviceSetting, Madviseable, clear_disk_cache};
use memory::mmap_ops::{create_and_ensure_length, open_write_mmap};
use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout};
@@ -259,9 +259,9 @@ impl<T: MmapValue + ?Sized> MmapPointToValues<T> {
})
}
pub fn open(path: &Path) -> OperationResult<Self> {
pub fn open(path: &Path, populate: bool) -> OperationResult<Self> {
let file_name = path.join(POINT_TO_VALUES_PATH);
let mmap = open_write_mmap(&file_name, AdviceSetting::Global, false)?;
let mmap = open_write_mmap(&file_name, AdviceSetting::Global, populate)?;
let (header, _) = Header::read_from_prefix(mmap.as_ref()).map_err(|_| {
OperationError::InconsistentStorage {
description: NOT_ENOUGHT_BYTES_ERROR_MESSAGE.to_owned(),
@@ -367,6 +367,18 @@ impl<T: MmapValue + ?Sized> MmapPointToValues<T> {
None
}
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) {
self.mmap.populate();
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
clear_disk_cache(&self.file_name)?;
Ok(())
}
}
#[cfg(test)]
@@ -429,7 +441,7 @@ mod tests {
.map(|(id, values)| (id as PointOffsetType, values.iter().map(|s| s.as_str()))),
)
.unwrap();
let point_to_values = MmapPointToValues::<str>::open(dir.path()).unwrap();
let point_to_values = MmapPointToValues::<str>::open(dir.path(), false).unwrap();
for (idx, values) in values.iter().enumerate() {
let iter = point_to_values.get_values(idx as PointOffsetType);
@@ -486,7 +498,7 @@ mod tests {
.map(|(id, values)| (id as PointOffsetType, values.iter().cloned())),
)
.unwrap();
let point_to_values = MmapPointToValues::<GeoPoint>::open(dir.path()).unwrap();
let point_to_values = MmapPointToValues::<GeoPoint>::open(dir.path(), false).unwrap();
for (idx, values) in values.iter().enumerate() {
let iter = point_to_values.get_values(idx as PointOffsetType);

20
lib/segment/src/index/field_index/null_index/mmap_null_index.rs
View File

@@ -177,6 +177,26 @@ impl MmapNullIndex {
histogram_bucket_size: None,
}
}
pub fn is_on_disk(&self) -> bool {
!POPULATE_NULL_INDEX
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
self.is_null_slice.populate()?;
self.has_values_slice.populate()?;
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
self.is_null_slice.clear_cache()?;
self.has_values_slice.clear_cache()?;
Ok(())
}
}
impl PayloadFieldIndex for MmapNullIndex {

29
lib/segment/src/index/field_index/numeric_index/mmap_numeric_index.rs
View File

@@ -8,7 +8,7 @@ use common::counter::iterator_hw_measurement::HwMeasurementIteratorExt;
use common::types::PointOffsetType;
use io::file_operations::{atomic_save_json, read_json};
use memmap2::MmapMut;
use memory::madvise::AdviceSetting;
use memory::madvise::{AdviceSetting, clear_disk_cache};
use memory::mmap_ops::{self, create_and_ensure_length};
use memory::mmap_type::{MmapBitSlice, MmapSlice};
use serde::{Deserialize, Serialize};
@@ -168,7 +168,7 @@ impl<T: Encodable + Numericable + Default + MmapValue> MmapNumericIndex<T> {
do_populate,
)?)?
};
let point_to_values = MmapPointToValues::open(path)?;
let point_to_values = MmapPointToValues::open(path, do_populate)?;
Ok(Self {
pairs: map,
@@ -355,4 +355,29 @@ impl<T: Encodable + Numericable + Default + MmapValue> MmapNumericIndex<T> {
) -> ConditionedCounter<'a> {
ConditionedCounter::new(self.is_on_disk, hw_counter)
}
pub fn is_on_disk(&self) -> bool {
self.is_on_disk
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
self.pairs.populate()?;
self.point_to_values.populate();
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
let pairs_path = self.path.join(PAIRS_PATH);
let deleted_path = self.path.join(DELETED_PATH);
clear_disk_cache(&pairs_path)?;
clear_disk_cache(&deleted_path)?;
self.point_to_values.clear_cache()?;
Ok(())
}
}

32
lib/segment/src/index/field_index/numeric_index/mod.rs
View File

@@ -421,6 +421,35 @@ impl<T: Encodable + Numericable + MmapValue + Default> NumericIndexInner<T> {
}
}
}
pub fn is_on_disk(&self) -> bool {
match self {
NumericIndexInner::Mutable(_) => false,
NumericIndexInner::Immutable(_) => false,
NumericIndexInner::Mmap(index) => index.is_on_disk(),
}
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
match self {
NumericIndexInner::Mutable(_) => {} // Not a mmap
NumericIndexInner::Immutable(_) => {} // Not a mmap
NumericIndexInner::Mmap(index) => index.populate()?,
}
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
match self {
NumericIndexInner::Mutable(_) => {} // Not a mmap
NumericIndexInner::Immutable(_) => {} // Not a mmap
NumericIndexInner::Mmap(index) => index.clear_cache()?,
}
Ok(())
}
}
pub struct NumericIndex<T: Encodable + Numericable + MmapValue + Default, P> {
@@ -495,6 +524,9 @@ impl<T: Encodable + Numericable + MmapValue + Default, P> NumericIndex<T, P> {
pub fn values_count(&self, idx: PointOffsetType) -> usize;
pub fn get_values(&self, idx: PointOffsetType) -> Option<Box<dyn Iterator<Item = T> + '_>>;
pub fn values_is_empty(&self, idx: PointOffsetType) -> bool;
pub fn is_on_disk(&self) -> bool;
pub fn populate(&self) -> OperationResult<()>;
pub fn clear_cache(&self) -> OperationResult<()>;
}
}
}

6
lib/segment/src/index/hnsw_index/graph_layers.rs
View File

@@ -6,7 +6,6 @@ use common::fixed_length_priority_queue::FixedLengthPriorityQueue;
use common::types::{PointOffsetType, ScoredPointOffset};
use io::file_operations::read_bin;
use itertools::Itertools;
use memory::mmap_ops;
use serde::{Deserialize, Serialize};
use super::entry_points::EntryPoint;
@@ -342,8 +341,9 @@ impl GraphLayers {
.to_graph_links_ram();
}
pub fn prefault_mmap_pages(&self, path: &Path) -> Option<mmap_ops::PrefaultMmapPages> {
self.links.prefault_mmap_pages(path)
pub fn populate(&self) -> OperationResult<()> {
self.links.populate()?;
Ok(())
}
}

17
lib/segment/src/index/hnsw_index/graph_links.rs
View File

@@ -3,8 +3,7 @@ use std::sync::Arc;
use common::types::PointOffsetType;
use memmap2::Mmap;
use memory::madvise::{Advice, AdviceSetting};
use memory::mmap_ops;
use memory::madvise::{Advice, AdviceSetting, Madviseable};
use memory::mmap_ops::open_read_mmap;
use crate::common::operation_error::OperationResult;
@@ -145,14 +144,14 @@ impl GraphLinks {
edges
}
pub fn prefault_mmap_pages(&self, path: &Path) -> Option<mmap_ops::PrefaultMmapPages> {
/// Populate the disk cache with data, if applicable.
/// This is a blocking operation.
pub fn populate(&self) -> OperationResult<()> {
match self.borrow_owner() {
GraphLinksEnum::Mmap(mmap) => Some(mmap_ops::PrefaultMmapPages::new(
Arc::clone(mmap),
Some(path.to_owned()),
)),
GraphLinksEnum::Ram(_) => None,
}
GraphLinksEnum::Mmap(mmap) => mmap.populate(),
GraphLinksEnum::Ram(_) => {}
};
Ok(())
}
}

37
lib/segment/src/index/hnsw_index/hnsw.rs
View File

@@ -14,7 +14,7 @@ use common::cpu::linux_low_thread_priority;
use common::ext::BitSliceExt as _;
use common::types::{PointOffsetType, ScoredPointOffset, TelemetryDetail};
use log::debug;
use memory::mmap_ops;
use memory::madvise::clear_disk_cache;
use parking_lot::Mutex;
use rayon::ThreadPool;
use rayon::prelude::*;
@@ -81,6 +81,7 @@ pub struct HNSWIndex {
path: PathBuf,
graph: GraphLayers,
searches_telemetry: HNSWSearchesTelemetry,
is_on_disk: bool,
}
#[derive(Debug)]
@@ -157,7 +158,9 @@ impl HNSWIndex {
let do_convert = LINK_COMPRESSION_CONVERT_EXISTING;
let graph = GraphLayers::load(path, hnsw_config.on_disk.unwrap_or(false), do_convert)?;
let is_on_disk = hnsw_config.on_disk.unwrap_or(false);
let graph = GraphLayers::load(path, is_on_disk, do_convert)?;
Ok(HNSWIndex {
id_tracker,
@@ -168,9 +171,14 @@ impl HNSWIndex {
path: path.to_owned(),
graph,
searches_telemetry: HNSWSearchesTelemetry::new(),
is_on_disk,
})
}
pub fn is_on_disk(&self) -> bool {
self.is_on_disk
}
#[cfg(test)]
pub(super) fn graph(&self) -> &GraphLayers {
&self.graph
@@ -475,11 +483,12 @@ impl HNSWIndex {
config.indexed_vector_count.replace(indexed_vectors);
let graph: GraphLayers = graph_layers_builder.into_graph_layers(
path,
LINK_COMPRESSION_FORMAT,
hnsw_config.on_disk.unwrap_or(false),
)?;
// Always skip loading graph to RAM on build
// as it will be discarded anyway
let is_on_disk = true;
let graph: GraphLayers =
graph_layers_builder.into_graph_layers(path, LINK_COMPRESSION_FORMAT, is_on_disk)?;
#[cfg(debug_assertions)]
{
@@ -510,6 +519,7 @@ impl HNSWIndex {
path: path.to_owned(),
graph,
searches_telemetry: HNSWSearchesTelemetry::new(),
is_on_disk,
})
}
@@ -1123,8 +1133,17 @@ impl HNSWIndex {
Ok(postprocess_result)
}
pub fn prefault_mmap_pages(&self) -> Option<mmap_ops::PrefaultMmapPages> {
self.graph.prefault_mmap_pages(&self.path)
/// Read underlying data from disk into disk cache.
pub fn populate(&self) -> OperationResult<()> {
self.graph.populate()
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
for file in self.graph.files(&self.path) {
clear_disk_cache(&file)?
}
Ok(())
}
}

8
lib/segment/src/index/sparse_index/sparse_vector_index.rs
View File

@@ -74,10 +74,6 @@ impl<TInvertedIndex: InvertedIndex> SparseVectorIndex<TInvertedIndex> {
&self.payload_index
}
pub fn inverted_index(&self) -> &TInvertedIndex {
&self.inverted_index
}
pub fn indices_tracker(&self) -> &IndicesTracker {
&self.indices_tracker
}
@@ -244,6 +240,10 @@ impl<TInvertedIndex: InvertedIndex> SparseVectorIndex<TInvertedIndex> {
))
}
pub fn inverted_index(&self) -> &TInvertedIndex {
&self.inverted_index
}
/// Returns the maximum number of results that can be returned by the index for a given sparse vector
/// Warning: the cost of this function grows with the number of dimensions in the query vector
#[cfg(feature = "testing")]

29
lib/segment/src/index/struct_payload_index.rs
View File

@@ -441,6 +441,35 @@ impl StructPayloadIndex {
key: key.to_string(),
})
}
pub fn populate(&self) -> OperationResult<()> {
for (_, field_indexes) in self.field_indexes.iter() {
for index in field_indexes {
index.populate()?;
}
}
Ok(())
}
pub fn clear_cache(&self) -> OperationResult<()> {
for (_, field_indexes) in self.field_indexes.iter() {
for index in field_indexes {
index.clear_cache()?;
}
}
Ok(())
}
pub fn clear_cache_if_on_disk(&self) -> OperationResult<()> {
for (_, field_indexes) in self.field_indexes.iter() {
for index in field_indexes {
if index.is_on_disk() {
index.clear_cache()?;
}
}
}
Ok(())
}
}
impl PayloadIndex for StructPayloadIndex {

53
lib/segment/src/index/vector_index_base.rs
View File

@@ -5,6 +5,7 @@ use common::counter::hardware_counter::HardwareCounterCell;
use common::types::{PointOffsetType, ScoredPointOffset, TelemetryDetail};
use half::f16;
use sparse::common::types::{DimId, QuantizedU8};
use sparse::index::inverted_index::InvertedIndex;
use sparse::index::inverted_index::inverted_index_compressed_immutable_ram::InvertedIndexCompressedImmutableRam;
use sparse::index::inverted_index::inverted_index_compressed_mmap::InvertedIndexCompressedMmap;
use sparse::index::inverted_index::inverted_index_immutable_ram::InvertedIndexImmutableRam;
@@ -97,6 +98,58 @@ impl VectorIndexEnum {
}
}
/// Returns true if underlying storage is configured to be stored on disk without
/// actively holding data in RAM
pub fn is_on_disk(&self) -> bool {
match self {
Self::Plain(_) => false,
Self::Hnsw(index) => index.is_on_disk(),
Self::SparseRam(index) => index.inverted_index().is_on_disk(),
Self::SparseImmutableRam(index) => index.inverted_index().is_on_disk(),
Self::SparseMmap(index) => index.inverted_index().is_on_disk(),
Self::SparseCompressedImmutableRamF32(index) => index.inverted_index().is_on_disk(),
Self::SparseCompressedImmutableRamF16(index) => index.inverted_index().is_on_disk(),
Self::SparseCompressedImmutableRamU8(index) => index.inverted_index().is_on_disk(),
Self::SparseCompressedMmapF32(index) => index.inverted_index().is_on_disk(),
Self::SparseCompressedMmapF16(index) => index.inverted_index().is_on_disk(),
Self::SparseCompressedMmapU8(index) => index.inverted_index().is_on_disk(),
}
}
pub fn populate(&self) -> OperationResult<()> {
match self {
Self::Plain(_) => {}
Self::Hnsw(index) => index.populate()?,
Self::SparseRam(_) => {}
Self::SparseImmutableRam(_) => {}
Self::SparseMmap(index) => index.inverted_index().populate()?,
Self::SparseCompressedImmutableRamF32(_) => {}
Self::SparseCompressedImmutableRamF16(_) => {}
Self::SparseCompressedImmutableRamU8(_) => {}
Self::SparseCompressedMmapF32(index) => index.inverted_index().populate()?,
Self::SparseCompressedMmapF16(index) => index.inverted_index().populate()?,
Self::SparseCompressedMmapU8(index) => index.inverted_index().populate()?,
};
Ok(())
}
pub fn clear_cache(&self) -> OperationResult<()> {
match self {
Self::Plain(_) => {}
Self::Hnsw(index) => index.clear_cache()?,
Self::SparseRam(_) => {}
Self::SparseImmutableRam(_) => {}
Self::SparseMmap(index) => index.inverted_index().clear_cache()?,
Self::SparseCompressedImmutableRamF32(_) => {}
Self::SparseCompressedImmutableRamF16(_) => {}
Self::SparseCompressedImmutableRamU8(_) => {}
Self::SparseCompressedMmapF32(index) => index.inverted_index().clear_cache()?,
Self::SparseCompressedMmapF16(index) => index.inverted_index().clear_cache()?,
Self::SparseCompressedMmapU8(index) => index.inverted_index().clear_cache()?,
};
Ok(())
}
pub fn fill_idf_statistics(
&self,
idf: &mut HashMap<DimId, usize>,

13
lib/segment/src/payload_storage/mmap_payload_storage.rs
View File

@@ -59,6 +59,19 @@ impl MmapPayloadStorage {
let storage = Arc::new(RwLock::new(storage));
Ok(Self { storage })
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
self.storage.read().populate()?;
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
self.storage.read().clear_cache()?;
Ok(())
}
}
impl PayloadStorage for MmapPayloadStorage {

27
lib/segment/src/payload_storage/payload_storage_enum.rs
View File

@@ -207,6 +207,33 @@ impl PayloadStorage for PayloadStorageEnum {
}
}
impl PayloadStorageEnum {
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
match self {
#[cfg(feature = "testing")]
PayloadStorageEnum::InMemoryPayloadStorage(_) => {}
PayloadStorageEnum::SimplePayloadStorage(_) => {}
PayloadStorageEnum::OnDiskPayloadStorage(_) => {}
PayloadStorageEnum::MmapPayloadStorage(s) => s.populate()?,
}
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
match self {
#[cfg(feature = "testing")]
PayloadStorageEnum::InMemoryPayloadStorage(_) => {}
PayloadStorageEnum::SimplePayloadStorage(_) => {}
PayloadStorageEnum::OnDiskPayloadStorage(_) => {}
PayloadStorageEnum::MmapPayloadStorage(s) => s.clear_cache()?,
}
Ok(())
}
}
#[cfg(test)]
mod tests {
use tempfile::Builder;

48
lib/segment/src/segment/mod.rs
View File

@@ -20,7 +20,6 @@ use std::thread::JoinHandle;
use atomic_refcell::AtomicRefCell;
use io::storage_version::StorageVersion;
use memory::mmap_ops;
use parking_lot::{Mutex, RwLock};
use rocksdb::DB;
@@ -94,27 +93,42 @@ impl fmt::Debug for VectorData {
}
}
impl VectorData {
pub fn prefault_mmap_pages(&self) -> impl Iterator<Item = mmap_ops::PrefaultMmapPages> {
let index_task = match &*self.vector_index.borrow() {
VectorIndexEnum::Hnsw(index) => index.prefault_mmap_pages(),
_ => None,
};
let storage_task = match &*self.vector_storage.borrow() {
VectorStorageEnum::DenseMemmap(storage) => storage.prefault_mmap_pages(),
_ => None,
};
index_task.into_iter().chain(storage_task)
}
}
impl Drop for Segment {
fn drop(&mut self) {
if let Err(flushing_err) = self.lock_flushing() {
log::error!("Failed to flush segment during drop: {flushing_err}");
}
// Try to remove everything from the disk cache, as it might pollute the cache
if let Err(e) = self.payload_storage.borrow().clear_cache() {
log::error!("Failed to clear cache of payload_storage: {e}");
}
if let Err(e) = self.payload_index.borrow().clear_cache() {
log::error!("Failed to clear cache of payload_index: {e}");
}
for (name, vector_data) in &self.vector_data {
let VectorData {
vector_index,
vector_storage,
quantized_vectors,
} = vector_data;
if let Err(e) = vector_index.borrow().clear_cache() {
log::error!("Failed to clear cache of vector index {name}: {e}");
}
if let Err(e) = vector_storage.borrow().clear_cache() {
log::error!("Failed to clear cache of vector storage {name}: {e}");
}
if let Some(quantized_vectors) = quantized_vectors.borrow().as_ref() {
if let Err(e) = quantized_vectors.clear_cache() {
log::error!("Failed to clear cache of quantized vectors {name}: {e}");
}
}
}
}
}

18
lib/segment/src/segment/segment_ops.rs
View File

@@ -2,13 +2,12 @@ use std::cmp::max;
use std::collections::{HashMap, HashSet};
use std::fs;
use std::path::Path;
use std::thread::{self, JoinHandle};
use std::thread::JoinHandle;
use bitvec::prelude::BitVec;
use common::counter::hardware_counter::HardwareCounterCell;
use common::types::PointOffsetType;
use io::file_operations::{atomic_save_json, read_json};
use memory::mmap_ops;
use super::{
DB_BACKUP_PATH, PAYLOAD_DB_BACKUP_PATH, SEGMENT_STATE_FILE, SNAPSHOT_FILES_PATH, SNAPSHOT_PATH,
@@ -653,21 +652,6 @@ impl Segment {
self.id_tracker.borrow().total_point_count()
}
pub fn prefault_mmap_pages(&self) {
let tasks: Vec<_> = self
.vector_data
.values()
.flat_map(|data| data.prefault_mmap_pages())
.collect();
let _ = thread::Builder::new()
.name(format!(
"segment-{:?}-prefault-mmap-pages",
self.current_path,
))
.spawn(move || tasks.iter().for_each(mmap_ops::PrefaultMmapPages::exec));
}
pub fn cleanup_versions(&mut self) -> OperationResult<()> {
self.id_tracker.borrow_mut().cleanup_versions()
}

49
lib/segment/src/segment_constructor/segment_builder.rs
View File

@@ -534,7 +534,7 @@ impl SegmentBuilder {
let payload_index_path = get_payload_index_path(temp_dir.path());
let mut payload_index = StructPayloadIndex::open(
payload_storage_arc,
payload_storage_arc.clone(),
id_tracker_arc.clone(),
vector_storages_arc.clone(),
&payload_index_path,
@@ -564,16 +564,18 @@ impl SegmentBuilder {
let permit = Arc::new(permit);
for (vector_name, vector_config) in &segment_config.vector_data {
build_vector_index(
let vector_storage = vector_storages_arc.remove(vector_name).unwrap();
let quantized_vectors =
Arc::new(AtomicRefCell::new(quantized_vectors.remove(vector_name)));
let index = build_vector_index(
vector_config,
VectorIndexOpenArgs {
path: &get_vector_index_path(temp_dir.path(), vector_name),
id_tracker: id_tracker_arc.clone(),
vector_storage: vector_storages_arc.remove(vector_name).unwrap(),
vector_storage: vector_storage.clone(),
payload_index: payload_index_arc.clone(),
quantized_vectors: Arc::new(AtomicRefCell::new(
quantized_vectors.remove(vector_name),
)),
quantized_vectors: quantized_vectors.clone(),
},
VectorIndexBuildArgs {
permit: permit.clone(),
@@ -582,6 +584,20 @@ impl SegmentBuilder {
stopped,
},
)?;
if vector_storage.borrow().is_on_disk() {
// If vector storage is expected to be on-disk, we need to clear cache
// to avoid cache pollution
vector_storage.borrow().clear_cache()?;
}
if let Some(quantized_vectors) = quantized_vectors.borrow().as_ref() {
quantized_vectors.clear_cache()?;
}
// Index if always loaded on-disk=true from build function
// So we may clear unconditionally
index.clear_cache()?;
}
for (vector_name, sparse_vector_config) in &segment_config.sparse_vector_data {
@@ -589,7 +605,7 @@ impl SegmentBuilder {
let vector_storage_arc = vector_storages_arc.remove(vector_name).unwrap();
create_sparse_vector_index(SparseVectorIndexOpenArgs {
let index = create_sparse_vector_index(SparseVectorIndexOpenArgs {
config: sparse_vector_config.index,
id_tracker: id_tracker_arc.clone(),
vector_storage: vector_storage_arc.clone(),
@@ -598,8 +614,27 @@ impl SegmentBuilder {
stopped,
tick_progress: || (),
})?;
if sparse_vector_config.storage_type.is_on_disk() {
// If vector storage is expected to be on-disk, we need to clear cache
// to avoid cache pollution
vector_storage_arc.borrow().clear_cache()?;
}
if sparse_vector_config.index.index_type.is_on_disk() {
index.clear_cache()?;
}
}
if segment_config.payload_storage_type.is_on_disk() {
// If payload storage is expected to be on-disk, we need to clear cache
// to avoid cache pollution
payload_storage_arc.borrow().clear_cache()?;
}
// Clear cache for payload index to avoid cache pollution
payload_index_arc.borrow().clear_cache_if_on_disk()?;
// We're done with CPU-intensive tasks, release CPU permit
debug_assert_eq!(
Arc::strong_count(&permit),

19
lib/segment/src/types.rs
View File

@@ -548,6 +548,13 @@ impl Indexes {
Indexes::Hnsw(_) => true,
}
}
pub fn is_on_disk(&self) -> bool {
match self {
Indexes::Plain {} => false,
Indexes::Hnsw(config) => config.on_disk.unwrap_or_default(),
}
}
}
/// Config of HNSW index
@@ -1375,6 +1382,18 @@ pub enum SparseVectorStorageType {
Mmap,
}
impl SparseVectorStorageType {
/// Whether this storage type is a mmap on disk
pub fn is_on_disk(&self) -> bool {
match self {
// Both options are on disk, but we keep it explicit for the case if someone adds a new
// storage type in the future
Self::OnDisk => true,
Self::Mmap => true,
}
}
}
/// Config of single sparse vector data storage
#[derive(Debug, Deserialize, Serialize, JsonSchema, Anonymize, Clone, Validate)]
#[serde(rename_all = "snake_case")]

17
lib/segment/src/vector_storage/chunked_mmap_vectors.rs
View File

@@ -8,7 +8,7 @@ use common::maybe_uninit::maybe_uninit_fill_from;
use io::file_operations::atomic_save_json;
use memmap2::MmapMut;
use memory::chunked_utils::{UniversalMmapChunk, chunk_name, create_chunk, read_mmaps};
use memory::madvise::{Advice, AdviceSetting};
use memory::madvise::{Advice, AdviceSetting, clear_disk_cache};
use memory::mmap_ops::{create_and_ensure_length, open_write_mmap};
use memory::mmap_type::MmapType;
use num_traits::AsPrimitive;
@@ -423,6 +423,21 @@ impl<T: Sized + Copy + 'static> ChunkedVectorStorage<T> for ChunkedMmapVectors<T
fn is_on_disk(&self) -> bool {
true
}
fn populate(&self) -> OperationResult<()> {
for chunk in &self.chunks {
chunk.populate()?;
}
Ok(())
}
fn clear_cache(&self) -> OperationResult<()> {
for chunk_idx in 0..self.chunks.len() {
let file_path = chunk_name(&self.directory, chunk_idx);
clear_disk_cache(&file_path)?;
}
Ok(())
}
}
#[cfg(test)]

7
lib/segment/src/vector_storage/chunked_vector_storage.rs
View File

@@ -60,4 +60,11 @@ pub trait ChunkedVectorStorage<T> {
/// True, if this storage is on-disk by default.
fn is_on_disk(&self) -> bool;
/// Populate all pages in the mmap.
/// Block until all pages are populated.
fn populate(&self) -> OperationResult<()>;
/// Drop disk cache.
fn clear_cache(&self) -> OperationResult<()>;
}

15
lib/segment/src/vector_storage/dense/appendable_dense_vector_storage.rs
View File

@@ -55,6 +55,21 @@ impl<T: PrimitiveVectorElement, S: ChunkedVectorStorage<T>> AppendableMmapDenseV
}
Ok(previous)
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
self.deleted.populate()?;
self.vectors.populate()?;
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
self.deleted.clear_cache()?;
self.vectors.clear_cache()?;
Ok(())
}
}
impl<T: PrimitiveVectorElement, S: ChunkedVectorStorage<T>> DenseVectorStorage<T>

16
lib/segment/src/vector_storage/dense/dynamic_mmap_flags.rs
View File

@@ -8,7 +8,7 @@ use bitvec::prelude::BitSlice;
use common::counter::referenced_counter::HwMetricRefCounter;
use common::types::PointOffsetType;
use memmap2::MmapMut;
use memory::madvise::{self, AdviceSetting, Madviseable as _};
use memory::madvise::{self, AdviceSetting, Madviseable as _, clear_disk_cache};
use memory::mmap_ops::{create_and_ensure_length, open_write_mmap};
use memory::mmap_type::{MmapBitSlice, MmapFlusher, MmapType};
use parking_lot::Mutex;
@@ -284,6 +284,20 @@ impl DynamicMmapFlags {
pub fn iter_trues(&self) -> impl Iterator<Item = PointOffsetType> + '_ {
self.flags.iter_ones().map(|x| x as PointOffsetType)
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
self.flags.populate()?;
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
let flags_file = self.directory.join(FLAGS_FILE);
clear_disk_cache(&flags_file)?;
Ok(())
}
}
#[cfg(test)]

27
lib/segment/src/vector_storage/dense/memmap_dense_vector_storage.rs
View File

@@ -9,6 +9,7 @@ use std::sync::atomic::AtomicBool;
use bitvec::prelude::BitSlice;
use common::counter::hardware_counter::HardwareCounterCell;
use common::types::PointOffsetType;
use memory::madvise::clear_disk_cache;
use memory::mmap_ops;
use crate::common::Flusher;
@@ -38,6 +39,24 @@ pub struct MemmapDenseVectorStorage<T: PrimitiveVectorElement> {
distance: Distance,
}
impl<T: PrimitiveVectorElement> MemmapDenseVectorStorage<T> {
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
if let Some(mmap_store) = &self.mmap_store {
mmap_store.populate()?;
}
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
clear_disk_cache(&self.vectors_path)?;
clear_disk_cache(&self.deleted_path)?;
Ok(())
}
}
pub fn open_memmap_vector_storage(
path: &Path,
dim: usize,
@@ -108,14 +127,6 @@ fn open_memmap_vector_storage_with_async_io_impl<T: PrimitiveVectorElement>(
}
impl<T: PrimitiveVectorElement> MemmapDenseVectorStorage<T> {
pub fn prefault_mmap_pages(&self) -> Option<mmap_ops::PrefaultMmapPages> {
Some(
self.mmap_store
.as_ref()?
.prefault_mmap_pages(&self.vectors_path),
)
}
pub fn get_mmap_vectors(&self) -> &MmapDenseVectors<T> {
self.mmap_store.as_ref().unwrap()
}

11
lib/segment/src/vector_storage/dense/mmap_dense_vectors.rs
View File

@@ -9,7 +9,7 @@ use common::ext::BitSliceExt as _;
use common::maybe_uninit::maybe_uninit_fill_from;
use common::types::PointOffsetType;
use memmap2::Mmap;
use memory::madvise::{Advice, AdviceSetting};
use memory::madvise::{Advice, AdviceSetting, Madviseable};
use memory::mmap_ops;
use memory::mmap_type::{MmapBitSlice, MmapFlusher};
use parking_lot::Mutex;
@@ -199,10 +199,6 @@ impl<T: PrimitiveVectorElement> MmapDenseVectors<T> {
&self.deleted
}
pub fn prefault_mmap_pages(&self, path: &Path) -> mmap_ops::PrefaultMmapPages {
mmap_ops::PrefaultMmapPages::new(self.mmap.clone(), Some(path))
}
#[cfg(target_os = "linux")]
fn process_points_uring(
&self,
@@ -247,6 +243,11 @@ impl<T: PrimitiveVectorElement> MmapDenseVectors<T> {
Ok(())
}
}
pub fn populate(&self) -> OperationResult<()> {
self.mmap.populate();
Ok(())
}
}
/// Ensure the given mmap file exists and is the given size

8
lib/segment/src/vector_storage/in_ram_persisted_vectors.rs
View File

@@ -112,4 +112,12 @@ impl<T: Sized + Copy + Clone + Default + 'static> ChunkedVectorStorage<T>
fn is_on_disk(&self) -> bool {
false
}
fn populate(&self) -> OperationResult<()> {
self.mmap_storage.populate()
}
fn clear_cache(&self) -> OperationResult<()> {
self.mmap_storage.clear_cache()
}
}

15
lib/segment/src/vector_storage/multi_dense/appendable_mmap_multi_dense_vector_storage.rs
View File

@@ -70,6 +70,21 @@ impl<
}
Ok(previous)
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
self.vectors.populate()?;
self.offsets.populate()?;
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
self.vectors.clear_cache()?;
self.offsets.clear_cache()?;
Ok(())
}
}
impl<

7
lib/segment/src/vector_storage/quantized/quantized_mmap_storage.rs
View File

@@ -2,12 +2,19 @@ use std::path::Path;
use memmap2::{Mmap, MmapMut};
use memory::madvise;
use memory::madvise::Madviseable;
#[derive(Debug)]
pub struct QuantizedMmapStorage {
mmap: Mmap,
}
impl QuantizedMmapStorage {
pub fn populate(&self) {
self.mmap.populate();
}
}
pub struct QuantizedMmapStorageBuilder {
mmap: MmapMut,
cursor_pos: usize,

4
lib/segment/src/vector_storage/quantized/quantized_multivector_storage.rs
View File

@@ -118,6 +118,10 @@ where
QuantizedStorage: EncodedVectors<TEncodedQuery>,
TMultivectorOffsetsStorage: MultivectorOffsetsStorage,
{
pub fn storage(&self) -> &QuantizedStorage {
&self.quantized_storage
}
pub fn new(
dim: usize,
quantized_storage: QuantizedStorage,

30
lib/segment/src/vector_storage/quantized/quantized_vectors.rs
View File

@@ -6,6 +6,7 @@ use bitvec::slice::BitSlice;
use common::counter::hardware_counter::HardwareCounterCell;
use common::types::PointOffsetType;
use io::file_operations::{atomic_save_json, read_json};
use memory::madvise::clear_disk_cache;
use quantization::encoded_vectors_binary::{EncodedBinVector, EncodedVectorsBin};
use quantization::{
EncodedQueryPQ, EncodedQueryU8, EncodedVectors, EncodedVectorsPQ, EncodedVectorsU8,
@@ -936,4 +937,33 @@ impl QuantizedVectors {
pub fn get_storage(&self) -> &QuantizedVectorStorage {
&self.storage_impl
}
pub fn populate(&self) -> OperationResult<()> {
match &self.storage_impl {
QuantizedVectorStorage::ScalarRam(_) => {} // not mmap
QuantizedVectorStorage::ScalarMmap(storage) => storage.storage().populate(),
QuantizedVectorStorage::PQRam(_) => {}
QuantizedVectorStorage::PQMmap(storage) => storage.storage().populate(),
QuantizedVectorStorage::BinaryRam(_) => {}
QuantizedVectorStorage::BinaryMmap(storage) => storage.storage().populate(),
QuantizedVectorStorage::ScalarRamMulti(_) => {}
QuantizedVectorStorage::ScalarMmapMulti(storage) => {
storage.storage().storage().populate()
}
QuantizedVectorStorage::PQRamMulti(_) => {}
QuantizedVectorStorage::PQMmapMulti(storage) => storage.storage().storage().populate(),
QuantizedVectorStorage::BinaryRamMulti(_) => {}
QuantizedVectorStorage::BinaryMmapMulti(storage) => {
storage.storage().storage().populate()
}
}
Ok(())
}
pub fn clear_cache(&self) -> OperationResult<()> {
for file in self.files() {
clear_disk_cache(&file)?;
}
Ok(())
}
}

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

@@ -174,6 +174,21 @@ impl MmapSparseVectorStorage {
Ok(())
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> OperationResult<()> {
self.deleted.populate()?;
self.storage.read().populate()?;
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> OperationResult<()> {
self.deleted.clear_cache()?;
self.storage.read().clear_cache()?;
Ok(())
}
}
impl SparseVectorStorage for MmapSparseVectorStorage {

60
lib/segment/src/vector_storage/vector_storage_base.rs
View File

@@ -429,6 +429,64 @@ impl VectorStorageEnum {
}
}
}
pub fn populate(&self) -> OperationResult<()> {
match self {
VectorStorageEnum::DenseSimple(_) => {} // Can't populate as it is not mmap
VectorStorageEnum::DenseSimpleByte(_) => {} // Can't populate as it is not mmap
VectorStorageEnum::DenseSimpleHalf(_) => {} // Can't populate as it is not mmap
VectorStorageEnum::DenseMemmap(vs) => vs.populate()?,
VectorStorageEnum::DenseMemmapByte(vs) => vs.populate()?,
VectorStorageEnum::DenseMemmapHalf(vs) => vs.populate()?,
VectorStorageEnum::DenseAppendableMemmap(vs) => vs.populate()?,
VectorStorageEnum::DenseAppendableMemmapByte(vs) => vs.populate()?,
VectorStorageEnum::DenseAppendableMemmapHalf(vs) => vs.populate()?,
VectorStorageEnum::DenseAppendableInRam(vs) => vs.populate()?,
VectorStorageEnum::DenseAppendableInRamByte(vs) => vs.populate()?,
VectorStorageEnum::DenseAppendableInRamHalf(vs) => vs.populate()?,
VectorStorageEnum::SparseSimple(_) => {} // Can't populate as it is not mmap
VectorStorageEnum::SparseMmap(vs) => vs.populate()?,
VectorStorageEnum::MultiDenseSimple(_) => {} // Can't populate as it is not mmap
VectorStorageEnum::MultiDenseSimpleByte(_) => {} // Can't populate as it is not mmap
VectorStorageEnum::MultiDenseSimpleHalf(_) => {} // Can't populate as it is not mmap
VectorStorageEnum::MultiDenseAppendableMemmap(vs) => vs.populate()?,
VectorStorageEnum::MultiDenseAppendableMemmapByte(vs) => vs.populate()?,
VectorStorageEnum::MultiDenseAppendableMemmapHalf(vs) => vs.populate()?,
VectorStorageEnum::MultiDenseAppendableInRam(vs) => vs.populate()?,
VectorStorageEnum::MultiDenseAppendableInRamByte(vs) => vs.populate()?,
VectorStorageEnum::MultiDenseAppendableInRamHalf(vs) => vs.populate()?,
}
Ok(())
}
pub fn clear_cache(&self) -> OperationResult<()> {
match self {
VectorStorageEnum::DenseSimple(_) => {} // Can't populate as it is not mmap
VectorStorageEnum::DenseSimpleByte(_) => {} // Can't populate as it is not mmap
VectorStorageEnum::DenseSimpleHalf(_) => {} // Can't populate as it is not mmap
VectorStorageEnum::DenseMemmap(vs) => vs.clear_cache()?,
VectorStorageEnum::DenseMemmapByte(vs) => vs.clear_cache()?,
VectorStorageEnum::DenseMemmapHalf(vs) => vs.clear_cache()?,
VectorStorageEnum::DenseAppendableMemmap(vs) => vs.clear_cache()?,
VectorStorageEnum::DenseAppendableMemmapByte(vs) => vs.clear_cache()?,
VectorStorageEnum::DenseAppendableMemmapHalf(vs) => vs.clear_cache()?,
VectorStorageEnum::DenseAppendableInRam(vs) => vs.clear_cache()?,
VectorStorageEnum::DenseAppendableInRamByte(vs) => vs.clear_cache()?,
VectorStorageEnum::DenseAppendableInRamHalf(vs) => vs.clear_cache()?,
VectorStorageEnum::SparseSimple(_) => {} // Can't populate as it is not mmap
VectorStorageEnum::SparseMmap(vs) => vs.clear_cache()?,
VectorStorageEnum::MultiDenseSimple(_) => {} // Can't populate as it is not mmap
VectorStorageEnum::MultiDenseSimpleByte(_) => {} // Can't populate as it is not mmap
VectorStorageEnum::MultiDenseSimpleHalf(_) => {} // Can't populate as it is not mmap
VectorStorageEnum::MultiDenseAppendableMemmap(vs) => vs.clear_cache()?,
VectorStorageEnum::MultiDenseAppendableMemmapByte(vs) => vs.clear_cache()?,
VectorStorageEnum::MultiDenseAppendableMemmapHalf(vs) => vs.clear_cache()?,
VectorStorageEnum::MultiDenseAppendableInRam(vs) => vs.clear_cache()?,
VectorStorageEnum::MultiDenseAppendableInRamByte(vs) => vs.clear_cache()?,
VectorStorageEnum::MultiDenseAppendableInRamHalf(vs) => vs.clear_cache()?,
}
Ok(())
}
}
impl VectorStorage for VectorStorageEnum {
@@ -488,6 +546,8 @@ impl VectorStorage for VectorStorageEnum {
}
}
/// If false - data is stored in RAM (and persisted on disk)
/// If true - data is stored on disk, and is not forced to be in RAM
fn is_on_disk(&self) -> bool {
match self {
VectorStorageEnum::DenseSimple(v) => v.is_on_disk(),

14
lib/sparse/src/index/inverted_index/inverted_index_compressed_mmap.rs
View File

@@ -10,7 +10,7 @@ use common::types::PointOffsetType;
use io::file_operations::{atomic_save_json, read_json};
use io::storage_version::StorageVersion;
use memmap2::Mmap;
use memory::madvise::{Advice, AdviceSetting};
use memory::madvise::{Advice, AdviceSetting, Madviseable, clear_disk_cache};
use memory::mmap_ops::{
create_and_ensure_length, open_read_mmap, transmute_from_u8_to_slice, transmute_to_u8,
transmute_to_u8_slice,
@@ -346,6 +346,18 @@ impl<W: Weight> InvertedIndexCompressedMmap<W> {
})
.sum()
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> std::io::Result<()> {
self.mmap.populate();
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> std::io::Result<()> {
clear_disk_cache(&self.path)
}
}
#[cfg(test)]

14
lib/sparse/src/index/inverted_index/inverted_index_mmap.rs
View File

@@ -8,7 +8,7 @@ use common::types::PointOffsetType;
use io::file_operations::{atomic_save_json, read_json};
use io::storage_version::StorageVersion;
use memmap2::{Mmap, MmapMut};
use memory::madvise::{Advice, AdviceSetting};
use memory::madvise::{Advice, AdviceSetting, Madviseable, clear_disk_cache};
use memory::mmap_ops::{
create_and_ensure_length, open_read_mmap, open_write_mmap, transmute_from_u8,
transmute_from_u8_to_slice, transmute_to_u8, transmute_to_u8_slice,
@@ -276,6 +276,18 @@ impl InvertedIndexMmap {
offset += posting_elements_bytes.len();
}
}
/// Populate all pages in the mmap.
/// Block until all pages are populated.
pub fn populate(&self) -> std::io::Result<()> {
self.mmap.populate();
Ok(())
}
/// Drop disk cache.
pub fn clear_cache(&self) -> std::io::Result<()> {
clear_disk_cache(&self.path)
}
}
#[cfg(test)]