diff --git a/lib/segment/src/index/field_index/field_index_base/read_only/lifecycle.rs b/lib/segment/src/index/field_index/field_index_base/read_only/lifecycle.rs index ff40bc64d1..54495feb9a 100644 --- a/lib/segment/src/index/field_index/field_index_base/read_only/lifecycle.rs +++ b/lib/segment/src/index/field_index/field_index_base/read_only/lifecycle.rs @@ -1,7 +1,7 @@ use std::path::Path; use common::bitvec::BitSlice; -use common::universal_io::UniversalReadFs; +use common::universal_io::{CachedReadFs, UniversalReadFs}; use super::ReadOnlyFieldIndex; use crate::common::operation_error::OperationResult; @@ -38,6 +38,86 @@ enum ReadMode { } impl ReadOnlyFieldIndex { + pub fn preopen( + fs: &impl CachedReadFs, + dir: &Path, + field: &JsonPath, + index_type: &FullPayloadIndexType, + ) -> OperationResult { + let mode = match index_type.storage_type { + StorageType::Gridstore => ReadMode::Appendable, + StorageType::Mmap { is_on_disk } => ReadMode::Immutable { is_on_disk }, + }; + + let preopened = match index_type.index_type { + PayloadIndexType::KeywordIndex => match mode { + ReadMode::Appendable => { + ReadOnlyMapIndex::::preopen_appendable(fs, map_dir(dir, field))? + } + ReadMode::Immutable { is_on_disk } => { + ReadOnlyMapIndex::::preopen_immutable( + fs, + &map_dir(dir, field), + is_on_disk, + )? + } + }, + PayloadIndexType::IntMapIndex => match mode { + ReadMode::Appendable => ReadOnlyMapIndex::::preopen_appendable( + fs, + map_dir(dir, field), + )?, + ReadMode::Immutable { is_on_disk } => { + ReadOnlyMapIndex::::preopen_immutable( + fs, + &map_dir(dir, field), + is_on_disk, + )? + } + }, + PayloadIndexType::UuidIndex | PayloadIndexType::UuidMapIndex => match mode { + ReadMode::Appendable => { + ReadOnlyMapIndex::::preopen_appendable(fs, map_dir(dir, field))? + } + ReadMode::Immutable { is_on_disk } => { + ReadOnlyMapIndex::::preopen_immutable( + fs, + &map_dir(dir, field), + is_on_disk, + )? + } + }, + PayloadIndexType::IntIndex => match mode { + ReadMode::Appendable => false, + ReadMode::Immutable { is_on_disk: _ } => false, + }, + PayloadIndexType::DatetimeIndex => match mode { + ReadMode::Appendable => false, + ReadMode::Immutable { is_on_disk: _ } => false, + }, + PayloadIndexType::FloatIndex => match mode { + ReadMode::Appendable => false, + ReadMode::Immutable { is_on_disk: _ } => false, + }, + // Geo reuses the writable selector's `map_dir` (`-map` suffix). + PayloadIndexType::GeoIndex => match mode { + ReadMode::Appendable => false, + ReadMode::Immutable { is_on_disk: _ } => false, + }, + PayloadIndexType::FullTextIndex => match mode { + ReadMode::Appendable => false, + ReadMode::Immutable { is_on_disk: _ } => false, + }, + // Bool and null are roaring-flag backed: a single read-only `open` + // serves both modes (neither consumes the immutable-only + // `is_on_disk` / `deleted_points`). + PayloadIndexType::BoolIndex => false, + PayloadIndexType::NullIndex => false, + }; + + Ok(preopened) + } + /// Read-only mirror of [`IndexSelector::new_index_with_type`][1]: dispatches /// on [`FullPayloadIndexType::index_type`] and forwards to each per-index /// parent's open, wrapping the leaf in the matching variant. diff --git a/lib/segment/src/index/field_index/map_index/mutable_map_index/read_only/lifecycle.rs b/lib/segment/src/index/field_index/map_index/mutable_map_index/read_only/lifecycle.rs index d891c9fd56..a494711ff0 100644 --- a/lib/segment/src/index/field_index/map_index/mutable_map_index/read_only/lifecycle.rs +++ b/lib/segment/src/index/field_index/map_index/mutable_map_index/read_only/lifecycle.rs @@ -1,7 +1,7 @@ use std::path::PathBuf; use common::counter::hardware_counter::HardwareCounterCell; -use common::universal_io::{OkNotFound, Populate, UniversalRead, UniversalReadFs}; +use common::universal_io::{CachedReadFs, OkNotFound, Populate, UniversalRead, UniversalReadFs}; use gridstore::error::GridstoreError; use gridstore::{Blob, GridstoreReader}; @@ -14,6 +14,19 @@ impl ReadOnlyAppendableMapIndex where Vec<::Owned>: Blob + Send + Sync, { + pub fn preopen(fs: &impl CachedReadFs, dir: PathBuf) -> OperationResult { + // Gridstore reader + Ok( + GridstoreReader::::Owned>, S>::preopen( + fs, + dir, + Populate::PreferBackground, + ) + .ok_not_found()? + .is_some(), + ) + } + /// Open the appendable (Gridstore) map index read-only, threading every /// file open through the filesystem handle `fs`. /// diff --git a/lib/segment/src/index/field_index/map_index/on_disk_map_index/lifecycle.rs b/lib/segment/src/index/field_index/map_index/on_disk_map_index/lifecycle.rs index ddb732f2b4..a0b28589a5 100644 --- a/lib/segment/src/index/field_index/map_index/on_disk_map_index/lifecycle.rs +++ b/lib/segment/src/index/field_index/map_index/on_disk_map_index/lifecycle.rs @@ -10,8 +10,8 @@ use common::persisted_hashmap::{Key, UniversalHashMap, serialize_hashmap}; use common::stored_bitslice::StoredBitSlice; use common::types::PointOffsetType; use common::universal_io::{ - MmapFile, OkNotFound, OpenOptions, Populate, UniversalRead, UniversalReadFs, UniversalWrite, - read_json_via, + CachedReadFs, MmapFile, OkNotFound, OpenOptions, Populate, UniversalRead, UniversalReadFs, + UniversalWrite, read_json_via, }; use fs_err as fs; @@ -29,6 +29,52 @@ where N: MapIndexKey + Key + ?Sized, S: UniversalRead, { + fn open_options(populate: Populate) -> OpenOptions { + OpenOptions { + writeable: false, + need_sequential: false, + populate, + advice: AdviceSetting::Global, + } + } + + pub fn preopen( + fs: &impl CachedReadFs, + path: &Path, + populate: Populate, + ) -> OperationResult { + // Config + let config_path = path.join(CONFIG_PATH); + if fs + .schedule_prefetch(&config_path, None, None) + .ok_not_found()? + .is_none() + { + // If config doesn't exist, assume the index doesn't exist on disk + return Ok(false); + } + + // Value to points + let hashmap_path = path.join(HASHMAP_PATH); + fs.schedule_prefetch(&hashmap_path, Some(Self::open_options(populate)), None)?; + + // Point to values + OnDiskPointToValues::::preopen(fs, path, populate)?; + + // Prefix index + PrefixIndex::preopen(fs, path, populate)?; + + // Deleted bitslice + let deleted_path = path.join(DELETED_PATH); + fs.schedule_prefetch( + &deleted_path, + Some(Self::open_options(Populate::PreferBackground)), + None, + )?; + + Ok(true) + } + /// Open and load mmap map index from the given path pub fn open( fs: &impl UniversalReadFs, @@ -50,12 +96,7 @@ where let value_to_points = UniversalHashMap::open( fs, &hashmap_path, - OpenOptions { - writeable: false, - need_sequential: false, - populate, - advice: AdviceSetting::Global, - }, + Self::open_options(populate), Default::default(), )?; let point_to_values = OnDiskPointToValues::open(fs, path, populate)?; @@ -66,12 +107,7 @@ where let deleted_payload_mmap = StoredBitSlice::::open( fs, &deleted_path, - OpenOptions { - writeable: false, - need_sequential: false, - populate, - advice: AdviceSetting::Global, - }, + Self::open_options(Populate::No), Default::default(), )?; diff --git a/lib/segment/src/index/field_index/map_index/prefix_index/reader.rs b/lib/segment/src/index/field_index/map_index/prefix_index/reader.rs index cf1362fde8..0cabe15020 100644 --- a/lib/segment/src/index/field_index/map_index/prefix_index/reader.rs +++ b/lib/segment/src/index/field_index/map_index/prefix_index/reader.rs @@ -8,7 +8,7 @@ use common::counter::hardware_counter::HardwareCounterCell; use common::generic_consts::Random; use common::mmap::AdviceSetting; use common::universal_io::{ - MmapFile, OpenOptions, Populate, ReadRange, UniversalRead, UniversalReadFileOps, + CachedReadFs, MmapFile, OpenOptions, Populate, ReadRange, UniversalRead, UniversalReadFileOps, UniversalReadFs, }; @@ -53,28 +53,44 @@ pub struct PrefixIndex { } impl PrefixIndex { + fn open_options(populate: Populate) -> OpenOptions { + OpenOptions { + writeable: false, + need_sequential: false, + populate, + advice: AdviceSetting::Global, + } + } + + pub fn preopen( + fs: &impl CachedReadFs, + dir: &Path, + populate: Populate, + ) -> OperationResult<()> { + let file_path = dir.join(PREFIX_INDEX_PATH); + if !UniversalReadFileOps::exists(fs, &file_path)? { + return Ok(()); + } + + // TODO(uio): Turn Populate::No into Populate::BackgroundPartial(0..header + header.block_index_size) + fs.schedule_prefetch(&file_path, Some(Self::open_options(populate)), None)?; + + Ok(()) + } + /// Open the prefix index if its file exists; `Ok(None)` when the backing /// map index was built without prefix support. pub fn open( fs: &impl UniversalReadFs, - path: &Path, + dir: &Path, populate: Populate, ) -> OperationResult> { - let file_path = path.join(PREFIX_INDEX_PATH); + let file_path = dir.join(PREFIX_INDEX_PATH); if !UniversalReadFileOps::exists(fs, &file_path)? { return Ok(None); } - let storage = fs.open( - &file_path, - OpenOptions { - writeable: false, - need_sequential: false, - populate, - advice: AdviceSetting::Global, - }, - Default::default(), - )?; + let storage = fs.open(&file_path, Self::open_options(populate), Default::default())?; let header_size = size_of::
() as u64; let header_bytes = storage.read_bytes::(0..header_size, align_of::
())?; diff --git a/lib/segment/src/index/field_index/map_index/read_only/lifecycle.rs b/lib/segment/src/index/field_index/map_index/read_only/lifecycle.rs index 52db1f3ca3..760e634cc8 100644 --- a/lib/segment/src/index/field_index/map_index/read_only/lifecycle.rs +++ b/lib/segment/src/index/field_index/map_index/read_only/lifecycle.rs @@ -1,7 +1,7 @@ use std::path::{Path, PathBuf}; use common::bitvec::BitSlice; -use common::universal_io::{Populate, UniversalRead, UniversalReadFs}; +use common::universal_io::{CachedReadFs, Populate, UniversalRead, UniversalReadFs}; use gridstore::Blob; use super::super::MapIndexKey; @@ -16,6 +16,13 @@ impl ReadOnlyMapIndex where Vec<::Owned>: Blob + Send + Sync, { + pub fn preopen_appendable( + fs: &impl CachedReadFs, + dir: PathBuf, + ) -> OperationResult { + ReadOnlyAppendableMapIndex::::preopen(fs, dir) + } + /// Read-only mirror of [`MapIndex::new_gridstore`][1]: open the appendable /// (Gridstore-backed) map index read-only, threading every file open /// through the filesystem handle `fs`. @@ -34,6 +41,22 @@ where Ok(ReadOnlyAppendableMapIndex::open(fs, dir)?.map(Self::Appendable)) } + pub fn preopen_immutable( + fs: &impl CachedReadFs, + dir: &Path, + is_on_disk: bool, + ) -> OperationResult { + let effective_is_on_disk = + is_on_disk || common::low_memory::low_memory_mode().prefer_disk(); + + let populate = match effective_is_on_disk { + true => Populate::No, + false => Populate::PreferBackground, + }; + + OnDiskMapIndex::::preopen(fs, dir, populate) + } + /// Read-only mirror of [`MapIndex::new_mmap`][1]: open the immutable /// (mmap-format) map index read-only through [`UniversalMapIndex::open`], /// threading every file open through the filesystem handle `fs`. @@ -55,7 +78,10 @@ where let effective_is_on_disk = is_on_disk || common::low_memory::low_memory_mode().prefer_disk(); - let populate = Populate::from(!effective_is_on_disk); + let populate = match effective_is_on_disk { + true => Populate::No, + false => Populate::PreferBackground, + }; let Some(on_disk_index) = OnDiskMapIndex::open(fs, path, populate, deleted_points)? else { return Ok(None); }; diff --git a/lib/segment/src/index/field_index/on_disk_point_to_values.rs b/lib/segment/src/index/field_index/on_disk_point_to_values.rs index 6e592cc8f5..1ace32e6f3 100644 --- a/lib/segment/src/index/field_index/on_disk_point_to_values.rs +++ b/lib/segment/src/index/field_index/on_disk_point_to_values.rs @@ -8,7 +8,9 @@ use common::ext::ResultOptionExt; use common::generic_consts::Random; use common::mmap::{AdviceSetting, create_and_ensure_length, open_write_mmap}; use common::types::PointOffsetType; -use common::universal_io::{self, Populate, ReadOnly, ReadRange, UniversalRead, UniversalReadFs}; +use common::universal_io::{ + self, CachedReadFs, OpenOptions, Populate, ReadOnly, ReadRange, UniversalRead, UniversalReadFs, +}; use zerocopy::IntoBytes; use crate::common::operation_error::{OperationError, OperationResult}; @@ -97,6 +99,15 @@ where T: StoredValue + ?Sized, S: UniversalRead, { + fn open_options(populate: Populate) -> OpenOptions { + OpenOptions { + writeable: false, + need_sequential: false, + populate, + advice: AdviceSetting::Global, + } + } + pub fn build_from_iter<'a>( path: &Path, iter: impl Iterator)> + Clone, @@ -161,12 +172,25 @@ where Ok(()) } + pub fn preopen( + fs: &impl CachedReadFs, + dir: &Path, + populate: Populate, + ) -> OperationResult<()> { + let file_name = dir.join(POINT_TO_VALUES_PATH); + + // TODO(uio): Turn Populate::No into Populate::BackgroundPartial(0..header_size) + fs.schedule_prefetch(&file_name, Some(Self::open_options(populate)), None)?; + + Ok(()) + } + pub fn open( fs: &impl UniversalReadFs, - path: &Path, + dir: &Path, populate: Populate, ) -> OperationResult { - let file_name = path.join(POINT_TO_VALUES_PATH); + let file_name = dir.join(POINT_TO_VALUES_PATH); let open_options = common::universal_io::OpenOptions { writeable: false, diff --git a/lib/segment/src/index/struct_payload_index/read_only/lifecycle.rs b/lib/segment/src/index/struct_payload_index/read_only/lifecycle.rs index d6161fc96e..fdf96d1326 100644 --- a/lib/segment/src/index/struct_payload_index/read_only/lifecycle.rs +++ b/lib/segment/src/index/struct_payload_index/read_only/lifecycle.rs @@ -3,7 +3,7 @@ use std::path::Path; use std::sync::Arc; use atomic_refcell::AtomicRefCell; -use common::universal_io::UniversalReadFs; +use common::universal_io::{CachedReadFs, UniversalReadFs}; use super::{ReadOnlyIndexesMap, ReadOnlyStructPayloadIndex}; use crate::common::operation_error::{OperationError, OperationResult}; @@ -17,6 +17,29 @@ use crate::types::VectorNameBuf; use crate::vector_storage::read_only::VectorStorageReadEnum; impl ReadOnlyStructPayloadIndex { + pub fn preopen( + fs: &impl CachedReadFs, + path: &Path, + ) -> OperationResult { + // Config + let config_path = PayloadConfig::get_config_path(path); + let config = PayloadConfig::load_universal(fs, &config_path)?.ok_or_else(|| { + OperationError::service_error(format!( + "Read-only payload index missing config at {}", + config_path.display() + )) + })?; + + // Payload indexes + for (field, indexed) in config.indices.iter() { + for index_type in &indexed.types { + ReadOnlyFieldIndex::preopen(fs, path, field, index_type)?; + } + } + + Ok(config) + } + /// Read-only mirror of `StructPayloadIndex::open`: loads the payload config /// and each persisted field index through `fs` (never builds/migrates/writes). pub fn open( diff --git a/lib/segment/src/segment/read_only/lifecycle.rs b/lib/segment/src/segment/read_only/lifecycle.rs index 6bb23768cc..c37aa92098 100644 --- a/lib/segment/src/segment/read_only/lifecycle.rs +++ b/lib/segment/src/segment/read_only/lifecycle.rs @@ -14,6 +14,7 @@ use super::{ReadOnlySegment, ReadOnlyVectorData}; use crate::common::operation_error::{OperationError, OperationResult}; use crate::id_tracker::read_only_tracker_enum::ReadOnlyIdTrackerEnum; use crate::index::UniversalReadExt; +use crate::index::payload_config::PayloadConfig; use crate::index::read_only::{ReadOnlyVectorIndexOpenArgs, VectorIndexReadEnum}; use crate::index::struct_payload_index::read_only::ReadOnlyStructPayloadIndex; use crate::payload_storage::read_only::ReadOnlyPayloadStorage; @@ -48,6 +49,15 @@ fn build_cached_fs( .ok_not_found()?; } + // Payload index config + cached_fs + .schedule_prefetch( + &PayloadConfig::get_config_path(&get_payload_index_path(segment_path)), + None, + None, + ) + .ok_not_found()?; + cached_fs.cache_file_info()?; Ok(cached_fs) @@ -69,22 +79,31 @@ impl ReadOnlySegment { Self::open_via(&cached_fs, fs, segment_path, uuid, deferred_internal_id) } - fn first_preopen(fs: &impl CachedReadFs, segment_path: &Path) -> OperationResult<()> { + fn first_preopen( + fs: &impl CachedReadFs, + segment_path: &Path, + ) -> OperationResult { let SegmentState { initial_version: _, version: _, config, } = read_json_via(fs, segment_path.join(SEGMENT_STATE_FILE))?; + // Payload storage let payload_populate = match config.payload_storage_type { PayloadStorageType::InRamMmap => Populate::PreferBackground, PayloadStorageType::Mmap => Populate::No, }; ReadOnlyPayloadStorage::preopen(fs, segment_path.to_path_buf(), payload_populate)?; + // Id tracker ReadOnlyIdTrackerEnum::preopen(fs, segment_path)?; - Ok(()) + // Payload indexes + let payload_index_config = + ReadOnlyStructPayloadIndex::preopen(fs, &get_payload_index_path(segment_path))?; + + Ok(payload_index_config) } /// Read-only mirror of `load_segment`: assembles every read-only component