From a76870eb94541b9ecbdd34cc65fc0fd5a0cbbb80 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Luis=20Coss=C3=ADo?= Date: Wed, 6 May 2026 18:19:18 -0400 Subject: [PATCH] claude wip --- lib/common/common/benches/universal_io.rs | 12 +-- lib/common/common/src/stored_bitslice.rs | 23 ++--- .../universal_io/disk_cache/cached_slice.rs | 40 ++++++-- .../common/src/universal_io/disk_cache/mod.rs | 52 ++++++---- lib/common/common/src/universal_io/family.rs | 29 ------ .../common/src/universal_io/io_uring/mod.rs | 47 +++++---- .../common/src/universal_io/io_uring/tests.rs | 18 ++-- lib/common/common/src/universal_io/mmap.rs | 40 +++++--- lib/common/common/src/universal_io/mod.rs | 6 +- lib/common/common/src/universal_io/read.rs | 51 ++++++---- .../universal_io/wrappers/buffered_update.rs | 16 ++-- .../src/universal_io/wrappers/read_only.rs | 54 +++++++---- .../universal_io/wrappers/stored_struct.rs | 6 +- .../common/src/universal_io/wrappers/typed.rs | 95 ++++++++++--------- .../universal_io/wrappers/wrapped_pipeline.rs | 2 +- lib/common/common/src/universal_io/write.rs | 19 ++-- lib/gridstore/src/bitmask/gaps.rs | 12 +-- lib/gridstore/src/bitmask/mod.rs | 4 +- lib/gridstore/src/gridstore/view.rs | 6 +- lib/gridstore/src/pages.rs | 10 +- lib/gridstore/src/tracker.rs | 76 +++++++-------- .../src/common/buffered_update_bitslice.rs | 2 +- .../src/common/flags/dynamic_stored_flags.rs | 10 +- .../id_tracker/immutable_id_tracker/mod.rs | 4 +- .../mmap_inverted_index/types.rs | 26 ++++- .../mmap_inverted_index/uio_postings.rs | 26 +++-- .../geo_index/immutable_geo_index.rs | 2 +- .../field_index/geo_index/mmap_geo_index.rs | 24 ++--- .../numeric_index/mmap_numeric_index.rs | 8 +- .../field_index/stored_point_to_values.rs | 57 +++++++---- .../vector_storage/chunked_vectors/chunks.rs | 4 +- .../vector_storage/chunked_vectors/read.rs | 6 +- .../vector_storage/chunked_vectors/write.rs | 23 +---- .../dense/dense_vector_storage.rs | 12 +-- .../dense/immutable_dense_vectors.rs | 6 +- .../dense/read_only/chucked_vector_storage.rs | 4 +- ...endable_mmap_multi_dense_vector_storage.rs | 4 +- .../read_only/chunked_vector_storage.rs | 11 +-- .../src/vector_storage/read_only/mod.rs | 34 ++----- src/tonic/api/storage_read_api/helpers.rs | 2 +- src/tonic/api/storage_read_api/mod.rs | 18 ++-- 41 files changed, 469 insertions(+), 432 deletions(-) delete mode 100644 lib/common/common/src/universal_io/family.rs diff --git a/lib/common/common/benches/universal_io.rs b/lib/common/common/benches/universal_io.rs index cffa7b6d71..a1a730cd64 100644 --- a/lib/common/common/benches/universal_io.rs +++ b/lib/common/common/benches/universal_io.rs @@ -61,7 +61,7 @@ fn benches(c: &mut Criterion) { } } -fn read_benches>( +fn read_benches( c: &mut Criterion, impl_name: &str, // Corresponds to `C` elem_size: &str, // Corresponds to `T` @@ -78,7 +78,7 @@ fn read_benches>( let storage = C::open(path, options).unwrap(); let len = FILE_SIZE_BYTES / size_of::() as u64; let mut rng = rand::rng(); - assert_eq!(storage.len().unwrap(), len); + assert_eq!(storage.len::().unwrap(), len); let low_mem = std::env::var_os(LIMIT_MEMORY_ENV_INTERNAL).is_some(); @@ -96,7 +96,7 @@ fn read_benches>( let mut sum = 0u64; let offset = rng.random_range(0..len) * size_of::() as u64; let data = storage - .read::(ReadRange { + .read::(ReadRange { byte_offset: offset, length: 1, }) @@ -119,7 +119,7 @@ fn read_benches>( }) .map(|range| ((), range)); storage - .read_batch::(ranges, |(), chunk| { + .read_batch::(ranges, |(), chunk| { for &item in bytemuck::cast_slice::(chunk) { sum = sum.wrapping_add(item); } @@ -142,7 +142,7 @@ fn read_benches>( }) .map(|range| ((), range)); storage - .read_batch::(ranges, |(), chunk| { + .read_batch::(ranges, |(), chunk| { for &item in bytemuck::cast_slice::(chunk) { sum = sum.wrapping_add(item); } @@ -158,7 +158,7 @@ fn read_benches>( let read_batch_full = || { let mut sum = 0u64; storage - .read_batch::(ranges_full_file::(), |(), chunk| { + .read_batch::(ranges_full_file::(), |(), chunk| { for &item in bytemuck::cast_slice::(chunk) { sum = sum.wrapping_add(item); } diff --git a/lib/common/common/src/stored_bitslice.rs b/lib/common/common/src/stored_bitslice.rs index d54c14281c..0ed0fddd14 100644 --- a/lib/common/common/src/stored_bitslice.rs +++ b/lib/common/common/src/stored_bitslice.rs @@ -1,5 +1,5 @@ -//! Storage-agnostic bitslice backed by any [`UniversalRead`] / -//! [`UniversalWrite`] backend. +//! Storage-agnostic bitslice backed by any [`UniversalRead`] / +//! [`UniversalWrite`] backend, fixed to the `u64` element type. //! //! Provides [`BitSliceStorage`], a wrapper that interprets the underlying //! `u64`-element storage as a sequence of bits, supporting both read and write @@ -15,7 +15,8 @@ use itertools::Itertools; use crate::bitvec::BitVec; use crate::generic_consts::Random; use crate::universal_io::{ - Flusher, OpenOptions, ReadRange, Result, UniversalIoError, UniversalRead, UniversalWrite, + Flusher, OpenOptions, ReadRange, Result, TypedStorage, UniversalIoError, UniversalRead, + UniversalWrite, }; /// Number of bits per `BitStore` element. @@ -29,21 +30,21 @@ pub type MmapBitSlice = StoredBitSlice; /// A storage-agnostic bitslice that supports both reading and writing bits. /// -/// Wraps any [`UniversalRead`] / [`UniversalWrite`] backend and -/// interprets the underlying `u64` elements as a sequence of bits. -/// Bit-level operations are translated to element-level reads and writes -/// on the backend. +/// Wraps any [`UniversalRead`] / [`UniversalWrite`] backend, fixed to the +/// `u64` element type, and interprets the underlying `u64` elements as a +/// sequence of bits. Bit-level operations are translated to element-level +/// reads and writes on the backend. #[derive(Debug)] pub struct StoredBitSlice { - storage: S, + storage: TypedStorage, /// Total number of `BitStore` elements in the underlying storage. element_len: u64, } -impl> StoredBitSlice { +impl StoredBitSlice { /// Open a bitslice storage from the given path using backend `S`. pub fn open(path: impl AsRef, options: OpenOptions) -> Result { - let storage = S::open(path, options)?; + let storage = TypedStorage::::open(path, options)?; let element_len = storage.len()?; Ok(Self { storage, @@ -171,7 +172,7 @@ impl> StoredBitSlice { } } -impl> StoredBitSlice { +impl StoredBitSlice { /// Set multiple individual bits in a batch. /// /// Each `(bit_index, value)` pair sets a single bit. Bits within the same diff --git a/lib/common/common/src/universal_io/disk_cache/cached_slice.rs b/lib/common/common/src/universal_io/disk_cache/cached_slice.rs index cbf4099706..d19cea65b3 100644 --- a/lib/common/common/src/universal_io/disk_cache/cached_slice.rs +++ b/lib/common/common/src/universal_io/disk_cache/cached_slice.rs @@ -49,22 +49,40 @@ impl CachedSlice { /// guarantees correct alignment for any `T`. pub fn get_range(&self, range: Range) -> io::Result> { let t_size = size_of::(); - debug_assert!(t_size != 0, "cannot use zero-sized type"); + let byte_range = range.start * t_size..range.end * t_size; + self.read_byte_range_as::(byte_range) + } - let total_elements = range.end - range.start; - if total_elements == 0 { + /// Read a byte range and reinterpret it as a slice of `T2`. + /// + /// The byte range must be aligned with `size_of::()`. Single-block + /// reads return a zero-copy borrow when possible; multi-block reads + /// allocate a `Vec` for correct alignment. + pub(crate) fn read_byte_range_as( + &self, + byte_range: Range, + ) -> io::Result> { + let t2_size = size_of::(); + debug_assert!(t2_size != 0, "cannot use zero-sized type"); + debug_assert_eq!( + (byte_range.end - byte_range.start) % t2_size, + 0, + "byte range length must be a multiple of size_of::()", + ); + + if byte_range.is_empty() { return Ok(Cow::Borrowed(&[])); } - let byte_range = range.start * t_size..range.end * t_size; + let total_elements = (byte_range.end - byte_range.start) / t2_size; let mut blocks_iter = self.blocks_for(byte_range); // TODO(perf): if blocks are consecutive in the big cache file, we can still return without allocating. if blocks_iter.len() == 1 { let req = blocks_iter.next().expect("We just checked len() == 1"); let result = self.controller.get_from_cache(req, |bytes| { - let mut vec_t = vec![T::zeroed(); bytes.len() / t_size]; - bytemuck::cast_slice_mut::(&mut vec_t).copy_from_slice(bytes); + let mut vec_t = vec![T2::zeroed(); bytes.len() / t2_size]; + bytemuck::cast_slice_mut::(&mut vec_t).copy_from_slice(bytes); vec_t })?; @@ -74,9 +92,9 @@ impl CachedSlice { }); } - // Multi-block: allocate Vec directly for correct alignment. - let mut result = vec![T::zeroed(); total_elements]; - let result_bytes = bytemuck::cast_slice_mut::(&mut result); + // Multi-block: allocate Vec directly for correct alignment. + let mut result = vec![T2::zeroed(); total_elements]; + let result_bytes = bytemuck::cast_slice_mut::(&mut result); let mut copied = 0; let mut copy_block = |slice: &[u8]| { let end = copied + slice.len(); @@ -145,6 +163,10 @@ impl CachedSlice { self.len_bytes / size_of::() } + pub(crate) fn byte_len(&self) -> usize { + self.len_bytes + } + /// Returns the block descriptor for the provided bytes range. fn blocks_for(&self, bytes_range: Range) -> impl ExactSizeIterator { debug_assert!(bytes_range.start <= bytes_range.end); diff --git a/lib/common/common/src/universal_io/disk_cache/mod.rs b/lib/common/common/src/universal_io/disk_cache/mod.rs index a0a264d197..d5341931d5 100644 --- a/lib/common/common/src/universal_io/disk_cache/mod.rs +++ b/lib/common/common/src/universal_io/disk_cache/mod.rs @@ -66,14 +66,15 @@ impl UniversalReadFileOps for CachedSlice { } } -impl UniversalRead for CachedSlice +impl UniversalRead for CachedSlice where T: bytemuck::Pod, { - type ReadPipeline<'a, Meta> - = DiskCacheReadPipeline<'a, T, Meta> + type ReadPipeline<'a, U, Meta> + = DiskCacheReadPipeline<'a, T, U, Meta> where - Self: 'a; + Self: 'a, + U: bytemuck::Pod; fn open(path: impl AsRef, options: OpenOptions) -> Result { let Some(controller) = CacheController::global() else { @@ -97,18 +98,22 @@ where Ok(CachedSlice::open(controller, path.as_ref())?) } - fn read(&self, range: ReadRange) -> Result> { - let elem_start = usize::try_from(range.byte_offset).expect("range.start is within usize") - / size_of::(); - let elem_length = usize::try_from(range.length).expect("range.length is within usize"); - - let range = elem_start..elem_start + elem_length; - - Ok(self.get_range(range)?) + fn read(&self, range: ReadRange) -> Result> + where + U: bytemuck::Pod, + P: AccessPattern, + { + let byte_start = usize::try_from(range.byte_offset).expect("range.start is within usize"); + let byte_length = + usize::try_from(range.length).expect("range.length is within usize") * size_of::(); + Ok(self.read_byte_range_as::(byte_start..byte_start + byte_length)?) } - fn len(&self) -> Result { - Ok(Self::len(self) as u64) + fn len(&self) -> Result + where + U: bytemuck::Pod, + { + Ok((self.byte_len() / size_of::()) as u64) } fn populate(&self) -> Result<()> { @@ -125,21 +130,28 @@ where } } -pub struct DiskCacheReadPipeline<'file, T, Meta> +pub struct DiskCacheReadPipeline<'file, T, U, Meta> where T: bytemuck::Pod, + U: bytemuck::Pod, { - result: Option<(Meta, Cow<'file, [T]>)>, + result: Option<(Meta, Cow<'file, [U]>)>, + _phantom: std::marker::PhantomData<&'file CachedSlice>, } -impl<'file, T, Meta> UniversalReadPipeline<'file, T, Meta> for DiskCacheReadPipeline<'file, T, Meta> +impl<'file, T, U, Meta> UniversalReadPipeline<'file, U, Meta> + for DiskCacheReadPipeline<'file, T, U, Meta> where T: bytemuck::Pod, + U: bytemuck::Pod, { type File = CachedSlice; fn new() -> Result { - Ok(Self { result: None }) + Ok(Self { + result: None, + _phantom: std::marker::PhantomData, + }) } fn can_schedule(&mut self) -> bool { @@ -159,11 +171,11 @@ where return Err(UniversalIoError::QueueIsFull); } - self.result = Some((meta, file.read::(range)?)); + self.result = Some((meta, file.read::(range)?)); Ok(()) } - fn wait(&mut self) -> Result)>> { + fn wait(&mut self) -> Result)>> { Ok(self.result.take()) } } diff --git a/lib/common/common/src/universal_io/family.rs b/lib/common/common/src/universal_io/family.rs deleted file mode 100644 index 94b8a77c6e..0000000000 --- a/lib/common/common/src/universal_io/family.rs +++ /dev/null @@ -1,29 +0,0 @@ -use crate::universal_io::UniversalRead; - -/// Type-family marker for [`UniversalRead`] backends. -/// -/// Rust has no higher-kinded types, so we cannot write `S` where `S` is a -/// generic parameter. This trait stands in: `S: UniversalReadFamily` then -/// `S::Read` yields the concrete `UniversalRead` for element type `T`. -pub trait UniversalReadFamily { - type Read: UniversalRead; -} - -pub struct MmapFamily; -impl UniversalReadFamily for MmapFamily { - type Read = crate::universal_io::mmap::MmapFile; -} - -#[cfg(target_os = "linux")] -pub struct IoUringFamily; -#[cfg(target_os = "linux")] -impl UniversalReadFamily for IoUringFamily { - type Read = crate::universal_io::io_uring::IoUringFile; -} - -#[cfg(not(target_os = "windows"))] -pub struct CachedSliceFamily; -#[cfg(not(target_os = "windows"))] -impl UniversalReadFamily for CachedSliceFamily { - type Read = crate::universal_io::disk_cache::CachedSlice; -} diff --git a/lib/common/common/src/universal_io/io_uring/mod.rs b/lib/common/common/src/universal_io/io_uring/mod.rs index 728bac39ab..955deb4c02 100644 --- a/lib/common/common/src/universal_io/io_uring/mod.rs +++ b/lib/common/common/src/universal_io/io_uring/mod.rs @@ -48,11 +48,11 @@ impl UniversalReadFileOps for IoUringFile { } } -impl UniversalRead for IoUringFile -where - T: bytemuck::Pod, -{ - type ReadPipeline<'a, Meta> = IoUringPipeline<'a, T, Meta>; +impl UniversalRead for IoUringFile { + type ReadPipeline<'a, T, Meta> + = IoUringPipeline<'a, T, Meta> + where + T: bytemuck::Pod; fn open(path: impl AsRef, options: OpenOptions) -> Result { // Check that io_uring is supported on this system. @@ -86,11 +86,15 @@ where Ok(file) } - fn read(&self, range: ReadRange) -> Result> { + fn read(&self, range: ReadRange) -> Result> + where + T: bytemuck::Pod, + P: AccessPattern, + { if self.direct_io { // direct_io needs special handling return self - .read_iter::([((), range)])? + .read_iter::([((), range)])? .next() .expect("there's exactly one read") .map(|(_, data)| data); @@ -102,7 +106,10 @@ where Ok(Cow::Owned(items)) } - fn len(&self) -> Result { + fn len(&self) -> Result + where + T: bytemuck::Pod, + { let byte_len = self.file.metadata()?.len(); let items_len = byte_len / size_of::() as u64; @@ -207,20 +214,23 @@ where } } -impl UniversalWrite for IoUringFile -where - T: bytemuck::Pod, -{ - fn write(&mut self, byte_offset: ByteOffset, items: &[T]) -> Result<()> { +impl UniversalWrite for IoUringFile { + fn write(&mut self, byte_offset: ByteOffset, items: &[T]) -> Result<()> + where + T: bytemuck::Pod, + { let bytes = bytemuck::cast_slice(items); self.file.write_all_at(bytes, byte_offset)?; Ok(()) } - fn write_batch<'a>( + fn write_batch<'a, T>( &mut self, items: impl IntoIterator, - ) -> Result<()> { + ) -> Result<()> + where + T: bytemuck::Pod + 'a, + { let mut rt = IoUringRuntime::new()?; let mut items = items.into_iter().peekable(); @@ -245,10 +255,13 @@ where Ok(()) } - fn write_multi<'a>( + fn write_multi<'a, T>( files: &mut [Self], writes: impl IntoIterator, - ) -> Result<()> { + ) -> Result<()> + where + T: bytemuck::Pod + 'a, + { let mut rt = IoUringRuntime::new()?; let mut writes = writes.into_iter().peekable(); diff --git a/lib/common/common/src/universal_io/io_uring/tests.rs b/lib/common/common/src/universal_io/io_uring/tests.rs index 86638b4f0c..0755e5e90c 100644 --- a/lib/common/common/src/universal_io/io_uring/tests.rs +++ b/lib/common/common/src/universal_io/io_uring/tests.rs @@ -228,8 +228,8 @@ fn test_io_uring_read_multi_iter_basic(#[case] o_direct: bool) -> Result<()> { prevent_caching: Some(o_direct), ..Default::default() }; - let file_0 = >::open(&path_0, opts)?; - let file_1 = >::open(&path_1, opts)?; + let file_0 = ::open(&path_0, opts)?; + let file_1 = ::open(&path_1, opts)?; let files = [file_0, file_1]; // Interleaved reads across both files. @@ -249,7 +249,7 @@ fn test_io_uring_read_multi_iter_basic(#[case] o_direct: bool) -> Result<()> { ]; let mut results: Vec<(char, Vec)> = Vec::new(); - for record in IoUringFile::read_multi_iter::(reads)? { + for record in IoUringFile::read_multi_iter::(reads)? { let (idx, cow) = record?; results.push((idx, cow.into_owned())); } @@ -287,7 +287,7 @@ fn test_io_uring_read_multi_iter_many_ranges(#[case] o_direct: bool) -> Result<( let path = dir.path().join(format!("f{i}.bin")); fs_err::write(&path, bytemuck::cast_slice(&data)).unwrap(); - let file = >::open(&path, opts)?; + let file = ::open(&path, opts)?; files.push(file); all_data.push(data); } @@ -311,7 +311,7 @@ fn test_io_uring_read_multi_iter_many_ranges(#[case] o_direct: bool) -> Result<( .collect(); let mut results: Vec<((usize, usize), Vec)> = Vec::new(); - for record in IoUringFile::read_multi_iter::(reads)? { + for record in IoUringFile::read_multi_iter::(reads)? { let (idx, cow) = record?; results.push((idx, cow.into_owned())); } @@ -346,8 +346,8 @@ fn test_io_uring_read_multi_callback_matches_iter() -> Result<()> { fs_err::write(&path_b, bytemuck::cast_slice(&data_b)).unwrap(); let opts = OpenOptions::default(); - let file_a: IoUringFile = UniversalRead::::open(&path_a, opts)?; - let file_b: IoUringFile = UniversalRead::::open(&path_b, opts)?; + let file_a: IoUringFile = UniversalRead::open(&path_a, opts)?; + let file_b: IoUringFile = UniversalRead::open(&path_b, opts)?; let files = [file_a, file_b]; #[rustfmt::skip] @@ -361,14 +361,14 @@ fn test_io_uring_read_multi_callback_matches_iter() -> Result<()> { // Collect via callback. let mut callback_results: Vec<(usize, Vec)> = Vec::new(); - IoUringFile::read_multi::(reads.clone(), |idx, data| { + IoUringFile::read_multi::(reads.clone(), |idx, data| { callback_results.push((idx, data.to_vec())); Ok(()) })?; // Collect via iterator. let mut iter_results: Vec<(usize, Vec)> = Vec::new(); - for record in IoUringFile::read_multi_iter::(reads)? { + for record in IoUringFile::read_multi_iter::(reads)? { let (idx, cow) = record?; iter_results.push((idx, cow.into_owned())); } diff --git a/lib/common/common/src/universal_io/mmap.rs b/lib/common/common/src/universal_io/mmap.rs index 00108c20f9..bfd2d0144f 100644 --- a/lib/common/common/src/universal_io/mmap.rs +++ b/lib/common/common/src/universal_io/mmap.rs @@ -43,11 +43,11 @@ impl UniversalReadFileOps for MmapFile { } } -impl UniversalRead for MmapFile -where - T: bytemuck::Pod, -{ - type ReadPipeline<'a, Meta> = MmapReadPipeline<'a, T, Meta>; +impl UniversalRead for MmapFile { + type ReadPipeline<'a, T, Meta> + = MmapReadPipeline<'a, T, Meta> + where + T: bytemuck::Pod; fn open(path: impl AsRef, options: OpenOptions) -> Result { let OpenOptions { @@ -100,13 +100,20 @@ where Ok(mmap) } - fn read(&self, range: ReadRange) -> Result> { + fn read(&self, range: ReadRange) -> Result> + where + T: bytemuck::Pod, + P: AccessPattern, + { let mmap = self.as_bytes::

(); let items = read(mmap, range)?; Ok(Cow::Borrowed(items)) } - fn len(&self) -> Result { + fn len(&self) -> Result + where + T: bytemuck::Pod, + { let len = self.len / size_of::(); Ok(len as u64) } @@ -165,20 +172,23 @@ where } } -impl UniversalWrite for MmapFile -where - T: bytemuck::Pod, -{ - fn write(&mut self, byte_offset: ByteOffset, items: &[T]) -> Result<()> { +impl UniversalWrite for MmapFile { + fn write(&mut self, byte_offset: ByteOffset, items: &[T]) -> Result<()> + where + T: bytemuck::Pod, + { let mmap = self.as_bytes_mut(); write(mmap, byte_offset, items)?; Ok(()) } - fn write_batch<'a>( + fn write_batch<'a, T>( &mut self, offset_data: impl IntoIterator, - ) -> Result<()> { + ) -> Result<()> + where + T: bytemuck::Pod + 'a, + { let mmap = self.as_bytes_mut(); for (byte_offset, items) in offset_data { @@ -273,7 +283,7 @@ impl MmapFile { /// ensuring all measurements go through the same mmap path. #[cfg(unix)] pub fn probe_memory_stats(path: impl AsRef) -> std::io::Result<(u64, u64)> { - let file: Self = >::open( + let file: Self = ::open( path, OpenOptions { writeable: false, diff --git a/lib/common/common/src/universal_io/mod.rs b/lib/common/common/src/universal_io/mod.rs index 35f214392a..410de4f46f 100644 --- a/lib/common/common/src/universal_io/mod.rs +++ b/lib/common/common/src/universal_io/mod.rs @@ -1,7 +1,6 @@ #[cfg(not(target_os = "windows"))] pub mod disk_cache; pub mod error; -pub mod family; pub mod file_ops; #[cfg(target_os = "linux")] pub mod io_uring; @@ -16,7 +15,6 @@ use std::path::Path; use serde::de::DeserializeOwned; pub use self::error::UniversalIoError; -pub use self::family::UniversalReadFamily; pub use self::file_ops::UniversalReadFileOps; #[cfg(target_os = "linux")] pub use self::io_uring::*; @@ -118,7 +116,7 @@ pub type Result = std::result::Result; /// Uses a single logical read when the backend overrides [`UniversalRead::read_whole`]. pub fn read_json_via(path: impl AsRef) -> Result where - S: UniversalRead, + S: UniversalRead, T: DeserializeOwned, { let options = OpenOptions { @@ -131,6 +129,6 @@ where }; let storage = S::open(path, options)?; - let bytes = storage.read_whole()?; + let bytes = storage.read_whole::()?; serde_json::from_slice(&bytes).map_err(UniversalIoError::from) } diff --git a/lib/common/common/src/universal_io/read.rs b/lib/common/common/src/universal_io/read.rs index c448782fe3..6e0375f61f 100644 --- a/lib/common/common/src/universal_io/read.rs +++ b/lib/common/common/src/universal_io/read.rs @@ -8,41 +8,46 @@ use crate::universal_io::file_ops::UniversalReadFileOps; /// Interface for accessing files in a universal way, abstracting away possible /// implementations, such as memory map, io_uring, DIRECTIO, S3, etc. #[expect(clippy::len_without_is_empty)] -pub trait UniversalRead: UniversalReadFileOps -where - T: Copy + 'static, -{ - type ReadPipeline<'file, Meta>: UniversalReadPipeline<'file, T, Meta, File = Self> +pub trait UniversalRead: UniversalReadFileOps + Sized { + type ReadPipeline<'file, T, Meta>: UniversalReadPipeline<'file, T, Meta, File = Self> where - Self: 'file; + Self: 'file, + T: bytemuck::Pod; fn open(path: impl AsRef, options: OpenOptions) -> Result; /// Prefer [`read_batch`] if you need high performance. - fn read(&self, range: ReadRange) -> Result>; + fn read(&self, range: ReadRange) -> Result> + where + T: bytemuck::Pod, + P: AccessPattern; /// Read the entire file in one logical access. /// /// Implementations may override this to avoid the two accesses that would result from /// `len()` followed by `read(0..len())`. Default implementation does exactly that. - fn read_whole(&self) -> Result> { + fn read_whole(&self) -> Result> + where + T: bytemuck::Pod, + { let range = ReadRange { byte_offset: 0, - length: self.len()?, + length: self.len::()?, }; - self.read::(range) + self.read::(range) } - fn read_batch( + fn read_batch( &self, ranges: impl IntoIterator, mut callback: impl FnMut(Meta, &[T]) -> Result<()>, ) -> Result<()> where + T: bytemuck::Pod, P: AccessPattern, { - for record in self.read_iter::(ranges)? { + for record in self.read_iter::(ranges)? { let (meta, data) = record?; callback(meta, &data)?; } @@ -52,21 +57,25 @@ where /// Like [`read_batch`](Self::read_batch), but returns a fallible iterator instead of /// accepting a callback. - fn read_iter( + fn read_iter( &self, ranges: impl IntoIterator, ) -> Result)>>> where + T: bytemuck::Pod, P: AccessPattern, { let reads = ranges .into_iter() .map(move |(meta, range)| (meta, self, range)); - Self::read_multi_iter::(reads) + Self::read_multi_iter::(reads) } - fn len(&self) -> Result; + /// Number of `T`-sized elements in the underlying storage. + fn len(&self) -> Result + where + T: bytemuck::Pod; /// Fill RAM cache with related data, if applicable for this implementation. /// @@ -79,15 +88,16 @@ where fn clear_ram_cache(&self) -> Result<()>; /// Read from multiple files in a single operation. - fn read_multi<'a, P, Meta>( + fn read_multi<'a, T, P, Meta>( reads: impl IntoIterator, mut callback: impl FnMut(Meta, &[T]) -> Result<()>, ) -> Result<()> where + T: bytemuck::Pod, P: AccessPattern, Self: 'a, { - for record in Self::read_multi_iter::(reads)? { + for record in Self::read_multi_iter::(reads)? { let (meta, items) = record?; callback(meta, &items)?; } @@ -97,14 +107,15 @@ where /// Like [`read_multi`](Self::read_multi), but returns a fallible iterator instead of /// accepting a callback. - fn read_multi_iter<'a, P, Meta>( + fn read_multi_iter<'a, T, P, Meta>( reads: impl IntoIterator, ) -> Result)>>> where + T: bytemuck::Pod, P: AccessPattern, Self: 'a, { - let mut pipeline = Self::ReadPipeline::<'a, Meta>::new()?; + let mut pipeline = Self::ReadPipeline::<'a, T, Meta>::new()?; let mut reads = reads.into_iter(); let iter = std::iter::from_fn(move || { @@ -131,7 +142,7 @@ where pub trait UniversalReadPipeline<'file, T, Meta>: Sized where - T: Copy + 'static, + T: bytemuck::Pod, { type File: 'file; diff --git a/lib/common/common/src/universal_io/wrappers/buffered_update.rs b/lib/common/common/src/universal_io/wrappers/buffered_update.rs index c0965d3c53..a9b76bf197 100644 --- a/lib/common/common/src/universal_io/wrappers/buffered_update.rs +++ b/lib/common/common/src/universal_io/wrappers/buffered_update.rs @@ -15,8 +15,8 @@ use crate::universal_io::{Flusher, UniversalIoError, UniversalWrite}; #[derive(Debug)] pub struct SliceBufferedUpdateWrapper where - S: UniversalWrite, - T: Copy + 'static, + S: UniversalWrite, + T: bytemuck::Pod, { slice: Arc>, len: u64, @@ -26,11 +26,11 @@ where impl SliceBufferedUpdateWrapper where - S: UniversalWrite, - T: Copy + 'static, + S: UniversalWrite, + T: bytemuck::Pod, { pub fn new(slice_storage: S) -> Result { - let len = slice_storage.len()?; + let len = slice_storage.len::()?; Ok(Self { slice: Arc::new(RwLock::new(slice_storage)), len, @@ -55,8 +55,8 @@ where impl SliceBufferedUpdateWrapper where - S: UniversalWrite + Send + Sync + 'static, - T: Copy + Clone + PartialEq + Sync + Send + 'static, + S: UniversalWrite + Send + Sync + 'static, + T: bytemuck::Pod + PartialEq + Sync + Send, { pub fn flusher(&self) -> Flusher { let updates = { @@ -99,7 +99,7 @@ where .expect("`chunk.len()` is sourced from the same slice as `remaining_values`"); (byte_start, chunk_values) }); - slice_guard.write_batch(it)?; + slice_guard.write_batch::(it)?; slice_guard.flusher()()?; diff --git a/lib/common/common/src/universal_io/wrappers/read_only.rs b/lib/common/common/src/universal_io/wrappers/read_only.rs index d69805191f..c85991b821 100644 --- a/lib/common/common/src/universal_io/wrappers/read_only.rs +++ b/lib/common/common/src/universal_io/wrappers/read_only.rs @@ -28,15 +28,15 @@ where } } -impl UniversalRead for ReadOnly +impl UniversalRead for ReadOnly where - S: UniversalRead, - T: Copy + 'static, + S: UniversalRead, { - type ReadPipeline<'file, Meta> - = WrappedReadPipeline<'file, Self, S::ReadPipeline<'file, Meta>> + type ReadPipeline<'file, T, Meta> + = WrappedReadPipeline<'file, Self, S::ReadPipeline<'file, T, Meta>> where - Self: 'file; + Self: 'file, + T: bytemuck::Pod; #[inline] fn open(path: impl AsRef, options: OpenOptions) -> Result { @@ -46,41 +46,53 @@ where } #[inline] - fn read(&self, range: ReadRange) -> Result> { - self.0.read::

(range) + fn read(&self, range: ReadRange) -> Result> + where + T: bytemuck::Pod, + P: AccessPattern, + { + self.0.read::(range) } #[inline] - fn read_whole(&self) -> Result> { - self.0.read_whole() + fn read_whole(&self) -> Result> + where + T: bytemuck::Pod, + { + self.0.read_whole::() } #[inline] - fn read_batch( + fn read_batch( &self, ranges: impl IntoIterator, callback: impl FnMut(Meta, &[T]) -> Result<()>, ) -> Result<()> where + T: bytemuck::Pod, P: AccessPattern, { - self.0.read_batch::(ranges, callback) + self.0.read_batch::(ranges, callback) } #[inline] - fn read_iter( + fn read_iter( &self, ranges: impl IntoIterator, ) -> Result)>>> where + T: bytemuck::Pod, P: AccessPattern, { - self.0.read_iter::(ranges) + self.0.read_iter::(ranges) } #[inline] - fn len(&self) -> Result { - self.0.len() + fn len(&self) -> Result + where + T: bytemuck::Pod, + { + self.0.len::() } #[inline] @@ -94,11 +106,12 @@ where } #[inline] - fn read_multi<'a, P, Meta>( + fn read_multi<'a, T, P, Meta>( reads: impl IntoIterator, callback: impl FnMut(Meta, &[T]) -> Result<()>, ) -> Result<()> where + T: bytemuck::Pod, P: AccessPattern, Self: 'a, { @@ -106,14 +119,15 @@ where .into_iter() .map(|(meta, file, range)| (meta, &file.0, range)); - S::read_multi::(reads, callback) + S::read_multi::(reads, callback) } #[inline] - fn read_multi_iter<'a, P, Meta>( + fn read_multi_iter<'a, T, P, Meta>( reads: impl IntoIterator, ) -> Result)>>> where + T: bytemuck::Pod, P: AccessPattern, Self: 'a, { @@ -121,7 +135,7 @@ where .into_iter() .map(|(meta, file, range)| (meta, &file.0, range)); - S::read_multi_iter::(it) + S::read_multi_iter::(it) } fn kind() -> UniversalKind { diff --git a/lib/common/common/src/universal_io/wrappers/stored_struct.rs b/lib/common/common/src/universal_io/wrappers/stored_struct.rs index 143246d576..b744e3f303 100644 --- a/lib/common/common/src/universal_io/wrappers/stored_struct.rs +++ b/lib/common/common/src/universal_io/wrappers/stored_struct.rs @@ -4,9 +4,7 @@ use std::sync::Arc; use parking_lot::Mutex; use crate::is_alive_lock::IsAliveLock; -use crate::universal_io::{ - self, Flusher, OpenOptions, TypedStorage, UniversalRead, UniversalWrite, -}; +use crate::universal_io::{self, Flusher, OpenOptions, TypedStorage, UniversalWrite}; /// A generic-storage wrapper for a type that should be possible to `read_whole` fairly cheaply. /// @@ -22,7 +20,7 @@ pub struct StoredStruct { impl StoredStruct where T: bytemuck::Pod + Send, - S: UniversalWrite + Send + 'static, + S: UniversalWrite + Send + 'static, { pub fn open(path: impl AsRef, options: OpenOptions) -> universal_io::Result { let storage = TypedStorage::open(path, options)?; diff --git a/lib/common/common/src/universal_io/wrappers/typed.rs b/lib/common/common/src/universal_io/wrappers/typed.rs index 212c8dee3c..75ec6b7d6a 100644 --- a/lib/common/common/src/universal_io/wrappers/typed.rs +++ b/lib/common/common/src/universal_io/wrappers/typed.rs @@ -8,16 +8,15 @@ use super::super::{ ByteOffset, FileIndex, Flusher, OpenOptions, ReadRange, Result, UniversalKind, UniversalRead, UniversalReadFileOps, UniversalWrite, }; -use super::WrappedReadPipeline; use crate::generic_consts::AccessPattern; -/// A wrapper around [`UniversalRead`]/[`UniversalWrite`] that binds `T` to a -/// specific type. +/// A wrapper around [`UniversalRead`]/[`UniversalWrite`] that fixes the element +/// type at the type level, providing a typed API that does not require a +/// turbofish for `T` at every call site. /// -/// This wrapper is not needed for code with a single universal io trait bound, -/// (e.g. `where S: UniversalRead`), but it helps the compiler to -/// distinguish when more than one bound is used, e.g. -/// `where S: UniversalRead + UniversalRead + …`. +/// `TypedStorage` itself does not implement [`UniversalRead`] — it exposes the +/// same surface via inherent methods, with `T` already chosen by the type +/// parameter. #[derive(Debug, TransparentWrapper)] #[repr(transparent)] #[transparent(S)] @@ -34,33 +33,28 @@ impl TypedStorage { } } -impl UniversalReadFileOps for TypedStorage +impl TypedStorage where S: UniversalReadFileOps, { #[inline] - fn list_files(prefix_path: &Path) -> Result> { + pub fn list_files(prefix_path: &Path) -> Result> { S::list_files(prefix_path) } #[inline] - fn exists(path: &Path) -> Result { + pub fn exists(path: &Path) -> Result { S::exists(path) } } -impl UniversalRead for TypedStorage +impl TypedStorage where - S: UniversalRead, - T: Copy + 'static, + S: UniversalRead, + T: bytemuck::Pod, { - type ReadPipeline<'file, Meta> - = WrappedReadPipeline<'file, Self, S::ReadPipeline<'file, Meta>> - where - Self: 'file; - #[inline] - fn open(path: impl AsRef, options: OpenOptions) -> Result { + pub fn open(path: impl AsRef, options: OpenOptions) -> Result { S::open(path, options).map(|inner| TypedStorage { inner, _phantom: PhantomData, @@ -68,17 +62,17 @@ where } #[inline] - fn read(&self, range: ReadRange) -> Result> { - self.inner.read::

(range) + pub fn read(&self, range: ReadRange) -> Result> { + self.inner.read::(range) } #[inline] - fn read_whole(&self) -> Result> { - self.inner.read_whole() + pub fn read_whole(&self) -> Result> { + self.inner.read_whole::() } #[inline] - fn read_batch( + pub fn read_batch( &self, ranges: impl IntoIterator, callback: impl FnMut(Meta, &[T]) -> Result<()>, @@ -86,37 +80,38 @@ where where P: AccessPattern, { - self.inner.read_batch::(ranges, callback) + self.inner.read_batch::(ranges, callback) } #[inline] - fn read_iter( + pub fn read_iter( &self, ranges: impl IntoIterator, ) -> Result)>>> where P: AccessPattern, { - self.inner.read_iter::(ranges) + self.inner.read_iter::(ranges) } #[inline] - fn len(&self) -> Result { - self.inner.len() + #[expect(clippy::len_without_is_empty)] + pub fn len(&self) -> Result { + self.inner.len::() } #[inline] - fn populate(&self) -> Result<()> { + pub fn populate(&self) -> Result<()> { self.inner.populate() } #[inline] - fn clear_ram_cache(&self) -> Result<()> { + pub fn clear_ram_cache(&self) -> Result<()> { self.inner.clear_ram_cache() } #[inline] - fn read_multi<'a, P, Meta>( + pub fn read_multi<'a, P, Meta>( reads: impl IntoIterator, callback: impl FnMut(Meta, &[T]) -> Result<()>, ) -> Result<()> @@ -128,11 +123,11 @@ where .into_iter() .map(|(meta, file, range)| (meta, &file.inner, range)); - S::read_multi::<'a, P, Meta>(reads, callback) + S::read_multi::(reads, callback) } #[inline] - fn read_multi_iter<'a, P, Meta>( + pub fn read_multi_iter<'a, P, Meta>( reads: impl IntoIterator, ) -> Result)>>> where @@ -143,42 +138,48 @@ where .into_iter() .map(|(meta, file, range)| (meta, &file.inner, range)); - S::read_multi_iter::(reads) + S::read_multi_iter::(reads) } - fn kind() -> UniversalKind { + pub fn kind() -> UniversalKind { S::kind() } } -impl UniversalWrite for TypedStorage +impl TypedStorage where - S: UniversalWrite, - T: Copy + 'static, + S: UniversalWrite, + T: bytemuck::Pod, { #[inline] - fn write(&mut self, byte_offset: ByteOffset, data: &[T]) -> Result<()> { + pub fn write(&mut self, byte_offset: ByteOffset, data: &[T]) -> Result<()> { self.inner.write(byte_offset, data) } #[inline] - fn write_batch<'a>( + pub fn write_batch<'a>( &mut self, offset_data: impl IntoIterator, - ) -> Result<()> { - self.inner.write_batch(offset_data) + ) -> Result<()> + where + T: 'a, + { + self.inner.write_batch::(offset_data) } #[inline] - fn flusher(&self) -> Flusher { + pub fn flusher(&self) -> Flusher { self.inner.flusher() } #[inline] - fn write_multi<'a>( + pub fn write_multi<'a>( files: &mut [Self], writes: impl IntoIterator, - ) -> Result<()> { - S::write_multi(Self::peel_slice_mut(files), writes) + ) -> Result<()> + where + T: 'a, + { + S::write_multi::(Self::peel_slice_mut(files), writes) } } diff --git a/lib/common/common/src/universal_io/wrappers/wrapped_pipeline.rs b/lib/common/common/src/universal_io/wrappers/wrapped_pipeline.rs index 050f6d664e..bbcf114e9b 100644 --- a/lib/common/common/src/universal_io/wrappers/wrapped_pipeline.rs +++ b/lib/common/common/src/universal_io/wrappers/wrapped_pipeline.rs @@ -18,7 +18,7 @@ impl<'a, File, Inner, T, Meta> UniversalReadPipeline<'a, T, Meta> where File: TransparentWrapper, Inner: UniversalReadPipeline<'a, T, Meta>, - T: Copy + 'static, + T: bytemuck::Pod, { type File = File; diff --git a/lib/common/common/src/universal_io/write.rs b/lib/common/common/src/universal_io/write.rs index 71ba2a2171..de89b05b1f 100644 --- a/lib/common/common/src/universal_io/write.rs +++ b/lib/common/common/src/universal_io/write.rs @@ -1,21 +1,28 @@ use super::read::UniversalRead; use super::*; -pub trait UniversalWrite: UniversalRead { - fn write(&mut self, byte_offset: ByteOffset, data: &[T]) -> Result<()>; +pub trait UniversalWrite: UniversalRead { + fn write(&mut self, byte_offset: ByteOffset, data: &[T]) -> Result<()> + where + T: bytemuck::Pod; - fn write_batch<'a>( + fn write_batch<'a, T>( &mut self, offset_data: impl IntoIterator, - ) -> Result<()>; + ) -> Result<()> + where + T: bytemuck::Pod + 'a; fn flusher(&self) -> Flusher; /// Write to multiple files in a single operation. - fn write_multi<'a>( + fn write_multi<'a, T>( files: &mut [Self], writes: impl IntoIterator, - ) -> Result<()> { + ) -> Result<()> + where + T: bytemuck::Pod + 'a, + { let files_len = files.len(); for (file_index, offset, data) in writes { diff --git a/lib/gridstore/src/bitmask/gaps.rs b/lib/gridstore/src/bitmask/gaps.rs index 3ca2f4de3d..c8cc92ecf4 100644 --- a/lib/gridstore/src/bitmask/gaps.rs +++ b/lib/gridstore/src/bitmask/gaps.rs @@ -3,7 +3,7 @@ use std::ops::Range; use std::path::{Path, PathBuf}; use common::mmap::{Advice, AdviceSetting, create_and_ensure_length}; -use common::universal_io::{Flusher, OpenOptions, UniversalWrite}; +use common::universal_io::{Flusher, OpenOptions, TypedStorage, UniversalWrite}; use itertools::Itertools; use super::{RegionId, StorageConfig}; @@ -83,10 +83,10 @@ fn gaps_file_path(dir: &Path) -> PathBuf { pub(super) struct BitmaskGaps { path: PathBuf, config: StorageConfig, - slice_store: S, + slice_store: TypedStorage, } -impl> BitmaskGaps { +impl BitmaskGaps { pub fn path(&self) -> PathBuf { self.path.clone() } @@ -110,7 +110,7 @@ impl> BitmaskGaps { advice: None, prevent_caching: None, }; - let mut slice_store = S::open(&path, options)?; + let mut slice_store = TypedStorage::::open(&path, options)?; debug_assert_eq!(slice_store.len()? as usize, data.len()); @@ -133,7 +133,7 @@ impl> BitmaskGaps { advice: Some(AdviceSetting::Advice(Advice::Normal)), prevent_caching: None, }; - let slice_store = S::open(&path, options)?; + let slice_store = TypedStorage::::open(&path, options)?; Ok(Self { path, @@ -168,7 +168,7 @@ impl> BitmaskGaps { advice: Some(AdviceSetting::Advice(Advice::Normal)), prevent_caching: None, }; - self.slice_store = S::open(&self.path, options)?; + self.slice_store = TypedStorage::::open(&self.path, options)?; debug_assert_eq!(self.len()? - prev_len, data.len()); diff --git a/lib/gridstore/src/bitmask/mod.rs b/lib/gridstore/src/bitmask/mod.rs index 776f3495fd..632b701f74 100644 --- a/lib/gridstore/src/bitmask/mod.rs +++ b/lib/gridstore/src/bitmask/mod.rs @@ -32,8 +32,8 @@ type RegionId = u32; /// Concrete bitmask type using memory-mapped storage. pub type MmapBitmask = Bitmask; -pub trait BitmaskStorage: UniversalWrite + UniversalWrite {} -impl BitmaskStorage for T where T: UniversalWrite + UniversalWrite {} +pub trait BitmaskStorage: UniversalWrite {} +impl BitmaskStorage for T where T: UniversalWrite {} #[derive(Debug)] pub struct Bitmask { diff --git a/lib/gridstore/src/gridstore/view.rs b/lib/gridstore/src/gridstore/view.rs index d88b76e7d8..8ee2c5f534 100644 --- a/lib/gridstore/src/gridstore/view.rs +++ b/lib/gridstore/src/gridstore/view.rs @@ -31,14 +31,14 @@ pub(super) fn decompress_lz4(value: &[u8]) -> Vec { /// [`Tracker`]). /// /// Constructed from either [`super::Gridstore`] or [`super::GridstoreReader`]. -pub struct GridstoreView<'a, V, S: UniversalRead> { +pub struct GridstoreView<'a, V, S: UniversalRead> { pub(super) config: &'a StorageConfig, pub(super) tracker: &'a Tracker, pub(super) pages: &'a Pages, pub(super) _value_type: std::marker::PhantomData, } -impl<'a, V, S: UniversalRead> GridstoreView<'a, V, S> { +impl<'a, V, S: UniversalRead> GridstoreView<'a, V, S> { pub(crate) fn new( config: &'a StorageConfig, tracker: &'a Tracker, @@ -74,7 +74,7 @@ impl<'a, V, S: UniversalRead> GridstoreView<'a, V, S> { } } -impl<'a, V: Blob, S: UniversalRead> GridstoreView<'a, V, S> { +impl<'a, V: Blob, S: UniversalRead> GridstoreView<'a, V, S> { pub(super) fn compress(&self, value: Vec) -> Vec { match self.config.compression { Compression::None => value, diff --git a/lib/gridstore/src/pages.rs b/lib/gridstore/src/pages.rs index 6fcd1aa5ae..c14ed6bf95 100644 --- a/lib/gridstore/src/pages.rs +++ b/lib/gridstore/src/pages.rs @@ -31,7 +31,7 @@ impl Pages { } } -impl> Pages { +impl Pages { pub fn new(base_path: PathBuf) -> Self { Self { base_path, @@ -213,7 +213,7 @@ impl> Pages { vec![MaybeUninit::uninit(); pointer.length as _] }; - for result in S::read_multi_iter::(reads)? { + for result in S::read_multi_iter::(reads)? { let (offset, bytes) = result?; if pages == 1 { @@ -282,7 +282,7 @@ impl> Pages { }) }); - let chunks = S::read_multi_iter::(reads).map_err(GridstoreError::from)?; + let chunks = S::read_multi_iter::(reads).map_err(GridstoreError::from)?; // Multi-page values need buffering since chunks for the same value may // arrive interleaved with chunks for other values. Single-page reads @@ -384,7 +384,7 @@ impl> Pages { } } -impl> Pages { +impl Pages { pub fn write_to_pages( &mut self, pointer: ValuePointer, @@ -398,7 +398,7 @@ impl> Pages { }); // Execute writes (mutable borrow of self.pages) - S::write_multi(self.pages.as_mut_slice(), writes)?; + S::write_multi::(self.pages.as_mut_slice(), writes)?; Ok(()) } diff --git a/lib/gridstore/src/tracker.rs b/lib/gridstore/src/tracker.rs index ae2b44342c..3fff1fa5a8 100644 --- a/lib/gridstore/src/tracker.rs +++ b/lib/gridstore/src/tracker.rs @@ -3,8 +3,6 @@ use std::path::{Path, PathBuf}; use ahash::{AHashMap, AHashSet}; use common::generic_consts::Random; use common::mmap::{Advice, AdviceSetting, create_and_ensure_length}; -#[expect(deprecated, reason = "legacy code")] -use common::mmap::{transmute_from_u8, transmute_to_u8}; use common::universal_io::{ OpenOptions, ReadRange, UniversalIoError, UniversalRead, UniversalWrite, }; @@ -43,6 +41,13 @@ struct Optional { value: T, } +// SAFETY: `Optional` is `#[repr(C)]` with a `u32` followed by `T`. With +// `T: bytemuck::Pod`, both fields are Pod, the layout has no padding for +// 4-byte-aligned `T`, and any bit pattern is valid (since `T: AnyBitPattern` +// and the discriminant field accepts any `u32`). +unsafe impl bytemuck::Zeroable for Optional {} +unsafe impl bytemuck::Pod for Optional {} + impl From> for Optional { fn from(value: Option) -> Self { match value { @@ -81,7 +86,9 @@ impl Optional { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, zerocopy::FromBytes)] +#[derive( + Debug, Clone, Copy, PartialEq, Eq, Hash, zerocopy::FromBytes, bytemuck::Pod, bytemuck::Zeroable, +)] #[repr(C)] pub struct ValuePointer { /// Which page the value is stored in @@ -227,7 +234,7 @@ impl PointerUpdates { } } -#[derive(Debug, Default, Clone, Copy)] +#[derive(Debug, Default, Clone, Copy, bytemuck::Pod, bytemuck::Zeroable)] #[repr(C)] struct TrackerHeader { next_pointer_offset: u32, @@ -268,7 +275,7 @@ impl Tracker { } // Read operations -- only require UniversalRead -impl> Tracker { +impl Tracker { /// Open an existing PageTracker at the given path /// If the file does not exist, return an error pub fn open(path: &Path) -> Result { @@ -286,12 +293,12 @@ impl> Tracker { } fn read_header(storage: &S) -> Result { - let header_bytes = storage.read::(ReadRange { + let header_bytes = storage.read::(ReadRange { byte_offset: 0, length: std::mem::size_of::() as u64, - })?; - #[expect(deprecated, reason = "legacy code")] - Ok(*unsafe { transmute_from_u8::(header_bytes.as_ref()) }) + })?[0]; + + Ok(header_bytes) } fn open_storage(path: &Path) -> Result { @@ -345,16 +352,16 @@ impl> Tracker { let start_offset = std::mem::size_of::() + point_offset as usize * std::mem::size_of::>(); let end_offset = start_offset + std::mem::size_of::>(); - let storage_len = self.storage.len()?; + let storage_len = self.storage.len::()?; if end_offset as u64 > storage_len { return Ok(None); } - let bytes = self.storage.read::(ReadRange { - byte_offset: start_offset as u64, - length: std::mem::size_of::>() as u64, - })?; - #[expect(deprecated, reason = "legacy code")] - let opt: &Optional = unsafe { transmute_from_u8(bytes.as_ref()) }; + let opt = self + .storage + .read::, Random>(ReadRange { + byte_offset: start_offset as u64, + length: 1, + })?[0]; Ok(Some(opt.is_some().copied())) } @@ -382,7 +389,7 @@ impl> Tracker { pub fn get_batch(&self, point_offsets: &[PointOffset]) -> Result>> { let item_size = std::mem::size_of::>(); let header_size = std::mem::size_of::(); - let storage_len = self.storage.len()?; + let storage_len = self.storage.len::()?; let mut result: Vec> = vec![None; point_offsets.len()]; let mut storage_reads: Vec<(usize, ReadRange)> = Vec::with_capacity(point_offsets.len()); @@ -409,7 +416,7 @@ impl> Tracker { i, ReadRange { byte_offset: start_offset as u64, - length: item_size as u64, + length: 1, }, )); } @@ -418,11 +425,9 @@ impl> Tracker { .iter() .map(|(i, range)| (*i, &self.storage, *range)); - for read_result in S::read_multi_iter::(reads)? { - let (i, bytes) = read_result?; - #[expect(deprecated, reason = "legacy code")] - let opt: &Optional = unsafe { transmute_from_u8(bytes.as_ref()) }; - result[i] = opt.is_some().copied(); + for read_result in S::read_multi_iter::, Random, _>(reads)? { + let (i, opts) = read_result?; + result[i] = opts[0].is_some().copied(); } Ok(result) @@ -470,14 +475,17 @@ impl> Tracker { /// Return the size of the underlying file #[cfg(test)] pub fn mmap_file_size(&self) -> Result { - self.storage.len().map(|u| u as usize).map_err(Into::into) + self.storage + .len::() + .map(|u| u as usize) + .map_err(Into::into) } } // Write operations and constructors -- require UniversalWrite impl Tracker where - S: UniversalRead + UniversalWrite, + S: UniversalWrite, { const DEFAULT_SIZE: usize = 1024 * 1024; // 1MB @@ -575,9 +583,7 @@ where /// Write the current page header to the storage fn write_header(&mut self) -> Result<()> { - #[expect(deprecated, reason = "legacy code")] - let header_bytes = unsafe { transmute_to_u8(&self.header) }; - self.storage.write(0, header_bytes)?; + self.storage.write(0, &[self.header])?; Ok(()) } @@ -588,7 +594,7 @@ where point_offset: PointOffset, pointer: Option, ) -> Result<()> { - let storage_len = self.storage.len()? as usize; + let storage_len = self.storage.len::()? as usize; if pointer.is_none() && point_offset as usize >= storage_len { return Ok(()); } @@ -607,9 +613,7 @@ where } let pointer: Optional<_> = pointer.into(); - #[expect(deprecated, reason = "legacy code")] - let pointer_bytes = unsafe { transmute_to_u8(&pointer) }; - self.storage.write(start_offset as u64, pointer_bytes)?; + self.storage.write(start_offset as u64, &[pointer])?; Ok(()) } @@ -646,8 +650,6 @@ where mod tests { use std::path::PathBuf; - #[expect(deprecated, reason = "legacy code")] - use common::mmap::transmute_from_u8; use common::universal_io::MmapFile; use rstest::rstest; use tempfile::Builder; @@ -974,8 +976,7 @@ mod tests { ); let none_data = AlignedData([0; _]); - #[expect(deprecated, reason = "legacy code")] - let none_val: &Optional = unsafe { transmute_from_u8(&none_data.0) }; + let none_val: &Optional = bytemuck::from_bytes(&none_data.0); assert!(none_val.is_some().is_none()); let some_data = AlignedData([ @@ -984,8 +985,7 @@ mod tests { 0x88, 0x77, 0x66, 0x55, // block_offset 0xDD, 0xCC, 0xBB, 0xAA, // length ]); - #[expect(deprecated, reason = "legacy code")] - let some_val: &Optional = unsafe { transmute_from_u8(&some_data.0) }; + let some_val: &Optional = bytemuck::from_bytes(&some_data.0); // N.B. fails on a big-endian machine. assert_eq!( some_val.is_some(), diff --git a/lib/segment/src/common/buffered_update_bitslice.rs b/lib/segment/src/common/buffered_update_bitslice.rs index 1beae46516..b5a647caf2 100644 --- a/lib/segment/src/common/buffered_update_bitslice.rs +++ b/lib/segment/src/common/buffered_update_bitslice.rs @@ -21,7 +21,7 @@ pub struct BufferedUpdateBitSlice { is_alive_flush_lock: common::is_alive_lock::IsAliveLock, } -impl + Send + Sync + 'static> BufferedUpdateBitSlice { +impl BufferedUpdateBitSlice { pub fn new(bitslice: StoredBitSlice) -> Self { let len = bitslice.bit_len() as usize; Self { diff --git a/lib/segment/src/common/flags/dynamic_stored_flags.rs b/lib/segment/src/common/flags/dynamic_stored_flags.rs index 00905a9bd8..660287b324 100644 --- a/lib/segment/src/common/flags/dynamic_stored_flags.rs +++ b/lib/segment/src/common/flags/dynamic_stored_flags.rs @@ -28,14 +28,8 @@ fn status_file(directory: &Path) -> PathBuf { directory.join(STATUS_FILE_NAME) } -pub trait UioDynamicFlags: - UniversalWrite + UniversalWrite + Send + 'static -{ -} -impl UioDynamicFlags for T where - T: UniversalWrite + UniversalWrite + Send + 'static -{ -} +pub trait UioDynamicFlags: UniversalWrite + Send + 'static {} +impl UioDynamicFlags for T where T: UniversalWrite + Send + 'static {} #[derive(Debug, Copy, Clone, bytemuck::Pod, bytemuck::Zeroable)] #[repr(C)] diff --git a/lib/segment/src/id_tracker/immutable_id_tracker/mod.rs b/lib/segment/src/id_tracker/immutable_id_tracker/mod.rs index 6f30d0a2b3..699effd1b2 100644 --- a/lib/segment/src/id_tracker/immutable_id_tracker/mod.rs +++ b/lib/segment/src/id_tracker/immutable_id_tracker/mod.rs @@ -16,9 +16,7 @@ use common::bitvec::{BitSlice, BitVec}; use common::mmap::create_and_ensure_length; use common::stored_bitslice::MmapBitSlice; use common::types::PointOffsetType; -use common::universal_io::{ - MmapFile, OpenOptions, SliceBufferedUpdateWrapper, TypedStorage, UniversalRead, UniversalWrite, -}; +use common::universal_io::{MmapFile, OpenOptions, SliceBufferedUpdateWrapper, TypedStorage}; use fs_err::File; pub use self::deleted_storage::DELETED_FILE_NAME; diff --git a/lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/types.rs b/lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/types.rs index 03c29b0803..9ddcd5586b 100644 --- a/lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/types.rs +++ b/lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/types.rs @@ -20,7 +20,18 @@ impl ZerocopyPostingValue for () {} impl ZerocopyPostingValue for Positions {} -#[derive(Debug, Default, Clone, FromBytes, Immutable, IntoBytes, KnownLayout)] +#[derive( + Debug, + Default, + Copy, + Clone, + FromBytes, + Immutable, + IntoBytes, + KnownLayout, + bytemuck::Pod, + bytemuck::Zeroable, +)] #[repr(C)] pub(in crate::index::field_index::full_text_index) struct PostingsHeader { /// Number of posting lists. One posting list per term @@ -30,7 +41,18 @@ pub(in crate::index::field_index::full_text_index) struct PostingsHeader { /// This data structure should contain all the necessary information to /// construct `PostingListView` from the mmap file. -#[derive(Debug, Default, Clone, FromBytes, Immutable, IntoBytes, KnownLayout)] +#[derive( + Debug, + Default, + Copy, + Clone, + FromBytes, + Immutable, + IntoBytes, + KnownLayout, + bytemuck::Pod, + bytemuck::Zeroable, +)] #[repr(C)] pub(in crate::index::field_index::full_text_index) struct PostingListHeader { /// Offset in bytes from the start of the mmap file diff --git a/lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/uio_postings.rs b/lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/uio_postings.rs index 3195bbb1b6..0cb92035f5 100644 --- a/lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/uio_postings.rs +++ b/lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/uio_postings.rs @@ -5,7 +5,6 @@ use std::path::PathBuf; use common::generic_consts::{Random, Sequential}; use common::universal_io::{OpenOptions, ReadRange, UniversalIoError, UniversalRead}; use posting_list::{PostingList, PostingListView}; -use zerocopy::FromBytes; use crate::common::operation_error::OperationResult; use crate::index::field_index::full_text_index::inverted_index::TokenId; @@ -23,7 +22,7 @@ use crate::index::field_index::full_text_index::inverted_index::mmap_inverted_in /// `size_of::() + token_id * size_of::()`. /// Each [`PostingListHeader`] then points (via absolute `offset`) into the /// posting-data region. -pub struct UniversalPostings> { +pub struct UniversalPostings { _path: PathBuf, storage: S, header: PostingsHeader, @@ -40,18 +39,16 @@ struct HeadersBatch<'a> { missing: Vec, } -impl> UniversalPostings { +impl UniversalPostings { /// Open the postings file at `path` via the `S` storage backend. pub fn open(path: impl Into, options: OpenOptions) -> OperationResult { let path = path.into(); let storage = S::open(&path, options)?; - let header_bytes = storage.read::(ReadRange { + let header = storage.read::(ReadRange { byte_offset: 0, - length: size_of::() as u64, - })?; - - let (header, _) = PostingsHeader::read_from_prefix(header_bytes.as_ref())?; + length: 1, + })?[0]; Ok(Self { _path: path, @@ -100,18 +97,17 @@ impl> UniversalPostings { token_id, ReadRange { byte_offset: header_offset, - length: header_length, + length: 1, }, )); } let valid_iter = self .storage - .read_iter::(valid_ranges)? + .read_iter::(valid_ranges)? .map(|res| { - let (token_id, bytes) = res?; - let (header, _) = PostingListHeader::read_from_prefix(bytes.as_ref())?; - Ok((token_id, header)) + let (token_id, headers) = res?; + Ok((token_id, headers[0])) }); Ok(HeadersBatch { @@ -150,7 +146,7 @@ impl> UniversalPostings { byte_offset: header.offset, length: header.posting_size::() as u64, }; - let bytes = self.storage.read::(read_range)?; + let bytes = self.storage.read::(read_range)?; let result = RawPostingList::new(bytes, header); Ok(result) } @@ -200,7 +196,7 @@ impl> UniversalPostings { let mut raw_postings: Vec<(TokenId, RawPostingList<'_>)> = Vec::with_capacity(expected_capacity); - for entry in self.storage.read_iter::(range_iter)? { + for entry in self.storage.read_iter::(range_iter)? { let ((token_id, header), bytes) = entry?; raw_postings.push((token_id, RawPostingList::new(bytes, header))); } diff --git a/lib/segment/src/index/field_index/geo_index/immutable_geo_index.rs b/lib/segment/src/index/field_index/geo_index/immutable_geo_index.rs index 0bb29f3e02..b4660690ed 100644 --- a/lib/segment/src/index/field_index/geo_index/immutable_geo_index.rs +++ b/lib/segment/src/index/field_index/geo_index/immutable_geo_index.rs @@ -4,7 +4,7 @@ use std::path::PathBuf; use common::bitvec::BitSliceExt; use common::generic_consts::Random; use common::types::PointOffsetType; -use common::universal_io::{MmapFile, ReadRange, UniversalRead}; +use common::universal_io::{MmapFile, ReadRange}; use super::mmap_geo_index::StoredGeoMapIndex; use crate::common::Flusher; diff --git a/lib/segment/src/index/field_index/geo_index/mmap_geo_index.rs b/lib/segment/src/index/field_index/geo_index/mmap_geo_index.rs index da5017aa2f..e4bbef8dc1 100644 --- a/lib/segment/src/index/field_index/geo_index/mmap_geo_index.rs +++ b/lib/segment/src/index/field_index/geo_index/mmap_geo_index.rs @@ -48,21 +48,8 @@ pub(super) struct PointKeyValue { } /// An alias to set of traits required by [`StoredGeoMapIndex`]. -#[expect(private_bounds)] -pub trait StoredGeoMapIndexStorage: - UniversalRead - + UniversalRead - + UniversalRead - + UniversalRead -{ -} -impl StoredGeoMapIndexStorage for T where - T: UniversalRead - + UniversalRead - + UniversalRead - + UniversalRead -{ -} +pub trait StoredGeoMapIndexStorage: UniversalRead {} +impl StoredGeoMapIndexStorage for T where T: UniversalRead {} /// /// points_map @@ -286,9 +273,10 @@ impl StoredGeoMapIndex { prevent_caching: None, }; - let counts_per_hash = UniversalRead::open(&counts_per_hash_path, open_options)?; - let points_map = UniversalRead::open(&points_map_path, open_options)?; - let points_map_ids = UniversalRead::open(&points_map_ids_path, open_options)?; + let counts_per_hash = TypedStorage::::open(&counts_per_hash_path, open_options)?; + let points_map = TypedStorage::::open(&points_map_path, open_options)?; + let points_map_ids = + TypedStorage::::open(&points_map_ids_path, open_options)?; let point_to_values = StoredPointToValues::open(path, true)?; let mut deleted = deleted_points.to_owned(); diff --git a/lib/segment/src/index/field_index/numeric_index/mmap_numeric_index.rs b/lib/segment/src/index/field_index/numeric_index/mmap_numeric_index.rs index f08aaf11b0..1aa4c25975 100644 --- a/lib/segment/src/index/field_index/numeric_index/mmap_numeric_index.rs +++ b/lib/segment/src/index/field_index/numeric_index/mmap_numeric_index.rs @@ -51,7 +51,7 @@ pub struct MmapNumericIndex> + UniversalRead, + S: UniversalRead, > { deleted: BitVec, // sorted pairs (id + value), sorted by value (by id if values are equal) @@ -59,11 +59,7 @@ pub(super) struct Storage< pub(super) point_to_values: StoredPointToValues, } -impl< - T: Encodable + Numericable + Default + StoredValue + 'static, - S: UniversalRead> + UniversalRead, -> Storage -{ +impl Storage { pub(crate) fn ram_usage_bytes(&self) -> usize { let Self { deleted, diff --git a/lib/segment/src/index/field_index/stored_point_to_values.rs b/lib/segment/src/index/field_index/stored_point_to_values.rs index a88dd56458..4ce47fb440 100644 --- a/lib/segment/src/index/field_index/stored_point_to_values.rs +++ b/lib/segment/src/index/field_index/stored_point_to_values.rs @@ -68,7 +68,7 @@ impl StoredValue for str { /// This structure is immutable. /// It's used in mmap field indices like `MmapMapIndex`, `MmapNumericIndex`, etc to store points-to-values map. /// This structure is not generic to avoid boxing lifetimes for `&str` values. -pub struct StoredPointToValues> { +pub struct StoredPointToValues { file_name: PathBuf, store: ReadOnly, header: Header, @@ -79,14 +79,35 @@ pub struct StoredPointToValues> { pub const MMAP_PTV_ACCESS_OVERHEAD: usize = size_of::(); #[repr(C)] -#[derive(Copy, Clone, Debug, Default, FromBytes, Immutable, IntoBytes, KnownLayout)] +#[derive( + Copy, + Clone, + Debug, + Default, + FromBytes, + Immutable, + IntoBytes, + KnownLayout, + bytemuck::Pod, + bytemuck::Zeroable, +)] struct MmapRange { start: u64, count: u64, } #[repr(C)] -#[derive(Copy, Clone, Debug, FromBytes, Immutable, IntoBytes, KnownLayout)] +#[derive( + Copy, + Clone, + Debug, + FromBytes, + Immutable, + IntoBytes, + KnownLayout, + bytemuck::Pod, + bytemuck::Zeroable, +)] struct Header { ranges_start: u64, points_count: u64, @@ -95,7 +116,7 @@ struct Header { impl StoredPointToValues where T: StoredValue + ?Sized, - S: UniversalRead, + S: UniversalRead, { pub fn from_iter<'a>( path: &Path, @@ -124,8 +145,7 @@ where ranges_start: PADDING_SIZE as u64, points_count: points_count as u64, }; - header - .write_to_prefix(mmap.as_mut()) + IntoBytes::write_to_prefix(&header, mmap.as_mut()) .map_err(|_| OperationError::service_error(NOT_ENOUGH_BYTES_ERROR_MESSAGE))?; // counter for values offset @@ -152,7 +172,7 @@ where header.ranges_start as usize + point_id as usize * std::mem::size_of::().., ) - .and_then(|bytes| range.write_to_prefix(bytes).ok()) + .and_then(|bytes| IntoBytes::write_to_prefix(&range, bytes).ok()) .ok_or_else(|| OperationError::service_error(NOT_ENOUGH_BYTES_ERROR_MESSAGE))?; } @@ -176,13 +196,10 @@ where let store = ReadOnly::open(&file_name, open_options)?; - let header_bytes = store.read::(ReadRange { + let header = store.read::(ReadRange { byte_offset: 0, - length: std::mem::size_of::

() as u64, - })?; - - let (header, _) = Header::read_from_prefix(&header_bytes) - .map_err(|_| OperationError::inconsistent_storage(NOT_ENOUGH_BYTES_ERROR_MESSAGE))?; + length: 1, + })?[0]; Ok(Self { file_name, @@ -241,7 +258,7 @@ where return Ok(None); }; - let bytes = self.store.read::(bytes_range)?; + let bytes = self.store.read::(bytes_range)?; let count = self.get_values_count(point_id)?.unwrap_or(0); let iter = ValuesIter::new(bytes, count); @@ -275,13 +292,11 @@ where let range_offset = (self.header.ranges_start as usize) + (point_id as usize) * std::mem::size_of::(); - let bytes = self.store.read::(ReadRange { + let range = self.store.read::(ReadRange { byte_offset: range_offset as u64, - length: std::mem::size_of::() as u64, - })?; - Ok(MmapRange::read_from_prefix(&bytes) - .ok() - .map(|(range, _)| range)) + length: 1, + })?[0]; + Ok(Some(range)) } else { Ok(None) } @@ -297,7 +312,7 @@ where Some(end) => end, None => { // if there is no next point, then use end of file - self.store.len()? + self.store.len::()? } }; diff --git a/lib/segment/src/vector_storage/chunked_vectors/chunks.rs b/lib/segment/src/vector_storage/chunked_vectors/chunks.rs index 803cdcabc4..c565de18da 100644 --- a/lib/segment/src/vector_storage/chunked_vectors/chunks.rs +++ b/lib/segment/src/vector_storage/chunked_vectors/chunks.rs @@ -18,7 +18,7 @@ fn check_mmap_file_name_pattern(file_name: &str) -> Option { .and_then(|file_name| file_name.parse::().ok()) } -pub fn read_chunks>( +pub fn read_chunks( directory: &Path, advice: AdviceSetting, populate: bool, @@ -73,7 +73,7 @@ pub fn chunk_name(directory: &Path, chunk_id: usize) -> PathBuf { )) } -pub fn create_chunk>( +pub fn create_chunk( directory: &Path, chunk_id: usize, chunk_length_bytes: usize, diff --git a/lib/segment/src/vector_storage/chunked_vectors/read.rs b/lib/segment/src/vector_storage/chunked_vectors/read.rs index 5f7b4d128f..1900c9cdf4 100644 --- a/lib/segment/src/vector_storage/chunked_vectors/read.rs +++ b/lib/segment/src/vector_storage/chunked_vectors/read.rs @@ -25,7 +25,7 @@ use crate::vector_storage::{VectorOffset, VectorOffsetType}; /// not refreshed afterwards. Mutating storage uses [`super::ChunkedVectors`] /// which wraps this and adds a writable status mmap. #[derive(Debug)] -pub struct ChunkedVectorsRead> { +pub struct ChunkedVectorsRead { pub(super) config: ChunkedVectorsConfig, /// Number of vectors currently stored. Snapshot for read-only mode; for /// [`super::ChunkedVectors`] this is kept in sync with the writable status @@ -35,7 +35,7 @@ pub struct ChunkedVectorsRead> { pub(super) directory: PathBuf, } -impl> ChunkedVectorsRead { +impl ChunkedVectorsRead { pub(super) fn config_file(directory: &Path) -> PathBuf { directory.join(CONFIG_FILE_NAME) } @@ -222,7 +222,7 @@ impl> ChunkedVectorsRead { }); // access pattern does not matter for io_uring - UniversalRead::read_multi_iter::(reads) + TypedStorage::::read_multi_iter::(reads) .expect("iterator initialized") .map(|result| result.expect("vector read")) } diff --git a/lib/segment/src/vector_storage/chunked_vectors/write.rs b/lib/segment/src/vector_storage/chunked_vectors/write.rs index 2a6e59cee1..c606d86168 100644 --- a/lib/segment/src/vector_storage/chunked_vectors/write.rs +++ b/lib/segment/src/vector_storage/chunked_vectors/write.rs @@ -17,30 +17,17 @@ use crate::common::operation_error::{OperationError, OperationResult}; use crate::vector_storage::VectorOffsetType; use crate::vector_storage::common::CHUNK_SIZE; -pub trait UioChunkedVectors: - UniversalWrite + UniversalWrite + Send + 'static -where - T: Copy + 'static, -{ -} -impl UioChunkedVectors for S -where - T: Copy + 'static, - S: UniversalWrite + UniversalWrite + Send + 'static, -{ -} - #[derive(Debug)] pub struct ChunkedVectors where - T: Copy + 'static, - S: UioChunkedVectors, + T: bytemuck::Pod, + S: UniversalWrite + Send + 'static, { inner: ChunkedVectorsRead, status: StoredStruct, } -impl> Deref for ChunkedVectors { +impl Deref for ChunkedVectors { type Target = ChunkedVectorsRead; fn deref(&self) -> &Self::Target { @@ -50,8 +37,8 @@ impl> Deref for ChunkedVectors impl ChunkedVectors where - T: Copy + 'static, - S: UioChunkedVectors, + T: bytemuck::Pod, + S: UniversalWrite + Send + 'static, { pub fn ensure_status_file(directory: &Path) -> OperationResult { let status_file = ChunkedVectorsRead::::status_file(directory); diff --git a/lib/segment/src/vector_storage/dense/dense_vector_storage.rs b/lib/segment/src/vector_storage/dense/dense_vector_storage.rs index 3591dc5492..9bfa84248e 100644 --- a/lib/segment/src/vector_storage/dense/dense_vector_storage.rs +++ b/lib/segment/src/vector_storage/dense/dense_vector_storage.rs @@ -39,7 +39,7 @@ const DELETED_PATH: &str = "deleted.dat"; pub struct DenseVectorStorageImpl where T: PrimitiveVectorElement, - S: UniversalRead, + S: UniversalRead, { vectors_path: PathBuf, deleted_path: PathBuf, @@ -51,7 +51,7 @@ where impl DenseVectorStorageImpl where T: PrimitiveVectorElement, - S: UniversalRead, + S: UniversalRead, { /// Populate all pages in the mmap. /// Block until all pages are populated. @@ -170,7 +170,7 @@ fn open_dense_vector_storage_impl( ) -> OperationResult> where T: PrimitiveVectorElement, - S: UniversalRead, + S: UniversalRead, { fs::create_dir_all(path)?; @@ -192,7 +192,7 @@ where impl DenseVectorStorage for DenseVectorStorageImpl where T: PrimitiveVectorElement, - S: UniversalRead, + S: UniversalRead, { fn vector_dim(&self) -> usize { self.vectors.as_ref().unwrap().dim @@ -215,7 +215,7 @@ where impl VectorStorageRead for DenseVectorStorageImpl where T: PrimitiveVectorElement, - S: UniversalRead, + S: UniversalRead, { fn distance(&self) -> Distance { self.distance @@ -284,7 +284,7 @@ where impl VectorStorage for DenseVectorStorageImpl where T: PrimitiveVectorElement, - S: UniversalRead, + S: UniversalRead, { fn insert_vector( &mut self, diff --git a/lib/segment/src/vector_storage/dense/immutable_dense_vectors.rs b/lib/segment/src/vector_storage/dense/immutable_dense_vectors.rs index f6d587a041..a2bc3bdc0e 100644 --- a/lib/segment/src/vector_storage/dense/immutable_dense_vectors.rs +++ b/lib/segment/src/vector_storage/dense/immutable_dense_vectors.rs @@ -29,11 +29,11 @@ const DELETED_HEADER: &[u8; HEADER_SIZE] = b"drop"; pub struct ImmutableDenseVectors where T: PrimitiveVectorElement, - S: UniversalRead, + S: UniversalRead, { pub dim: usize, pub num_vectors: usize, - /// Vector data storage, providing read access via [`UniversalRead`]. + /// Vector data storage, providing read access via [`UniversalRead`]. storage: TypedStorage, T>, /// Memory mapped deletion flags deleted: MmapBitSlice, @@ -41,7 +41,7 @@ where pub deleted_count: usize, } -impl> ImmutableDenseVectors { +impl ImmutableDenseVectors { pub fn open( vectors_path: &Path, deleted_path: &Path, diff --git a/lib/segment/src/vector_storage/dense/read_only/chucked_vector_storage.rs b/lib/segment/src/vector_storage/dense/read_only/chucked_vector_storage.rs index f3ce1c7871..9efeb5c1b3 100644 --- a/lib/segment/src/vector_storage/dense/read_only/chucked_vector_storage.rs +++ b/lib/segment/src/vector_storage/dense/read_only/chucked_vector_storage.rs @@ -10,7 +10,7 @@ use crate::vector_storage::chunked_vectors::ChunkedVectorsRead; use crate::vector_storage::{VectorOffsetType, VectorStorageRead}; #[derive(Debug)] -pub struct ReadOnlyChunkedDenseVectorStorage> { +pub struct ReadOnlyChunkedDenseVectorStorage { vectors: ChunkedVectorsRead, /// Flags marking deleted vectors /// @@ -21,7 +21,7 @@ pub struct ReadOnlyChunkedDenseVectorStorage> VectorStorageRead +impl VectorStorageRead for ReadOnlyChunkedDenseVectorStorage { fn distance(&self) -> Distance { diff --git a/lib/segment/src/vector_storage/multi_dense/appendable_mmap_multi_dense_vector_storage.rs b/lib/segment/src/vector_storage/multi_dense/appendable_mmap_multi_dense_vector_storage.rs index a6428de5aa..61acd21205 100644 --- a/lib/segment/src/vector_storage/multi_dense/appendable_mmap_multi_dense_vector_storage.rs +++ b/lib/segment/src/vector_storage/multi_dense/appendable_mmap_multi_dense_vector_storage.rs @@ -68,8 +68,8 @@ pub(crate) fn read_multi_vector<'a, T, P, So, Sv>( where T: PrimitiveVectorElement, P: AccessPattern, - So: UniversalRead, - Sv: UniversalRead, + So: UniversalRead, + Sv: UniversalRead, { let mmap_offset = *offsets .get::

(key as VectorOffsetType)? diff --git a/lib/segment/src/vector_storage/multi_dense/read_only/chunked_vector_storage.rs b/lib/segment/src/vector_storage/multi_dense/read_only/chunked_vector_storage.rs index fa1df75a56..84a04084ea 100644 --- a/lib/segment/src/vector_storage/multi_dense/read_only/chunked_vector_storage.rs +++ b/lib/segment/src/vector_storage/multi_dense/read_only/chunked_vector_storage.rs @@ -1,7 +1,7 @@ use common::bitvec::{BitSlice, BitVec}; use common::generic_consts::AccessPattern; use common::types::PointOffsetType; -use common::universal_io::UniversalReadFamily; +use common::universal_io::UniversalRead; use crate::data_types::named_vectors::CowVector; use crate::data_types::primitive::PrimitiveVectorElement; @@ -13,10 +13,9 @@ use crate::vector_storage::multi_dense::appendable_mmap_multi_dense_vector_stora }; #[derive(Debug)] -pub struct ReadOnlyChunkedMultiDenseVectorStorage -{ - vectors: ChunkedVectorsRead>, - offsets: ChunkedVectorsRead>, +pub struct ReadOnlyChunkedMultiDenseVectorStorage { + vectors: ChunkedVectorsRead, + offsets: ChunkedVectorsRead, /// Flags marking deleted vectors /// /// Structure grows dynamically, but may be smaller than actual number of vectors. Must not @@ -26,7 +25,7 @@ pub struct ReadOnlyChunkedMultiDenseVectorStorage VectorStorageRead +impl VectorStorageRead for ReadOnlyChunkedMultiDenseVectorStorage { fn distance(&self) -> Distance { diff --git a/lib/segment/src/vector_storage/read_only/mod.rs b/lib/segment/src/vector_storage/read_only/mod.rs index 409e873e85..c3b8b4a36a 100644 --- a/lib/segment/src/vector_storage/read_only/mod.rs +++ b/lib/segment/src/vector_storage/read_only/mod.rs @@ -1,7 +1,7 @@ use common::bitvec::BitSlice; use common::generic_consts::AccessPattern; use common::types::PointOffsetType; -use common::universal_io::UniversalReadFamily; +use common::universal_io::UniversalRead; use crate::data_types::named_vectors::CowVector; use crate::data_types::vectors::{VectorElementType, VectorElementTypeByte, VectorElementTypeHalf}; @@ -16,36 +16,20 @@ use crate::vector_storage::sparse::read_only::sparse_vector_storage::ReadOnlySpa /// /// Wraps each on-disk storage type with its [`super`] read-only variant. /// Volatile, empty and test-only variants are intentionally absent. -pub enum VectorStorageReadEnum { - Dense(Box>>), - DenseByte(Box>>), - DenseHalf(Box>>), - DenseChunked( - Box>>, - ), - DenseChunkedByte( - Box< - ReadOnlyChunkedDenseVectorStorage< - VectorElementTypeByte, - S::Read, - >, - >, - ), - DenseChunkedHalf( - Box< - ReadOnlyChunkedDenseVectorStorage< - VectorElementTypeHalf, - S::Read, - >, - >, - ), +pub enum VectorStorageReadEnum { + Dense(Box>), + DenseByte(Box>), + DenseHalf(Box>), + DenseChunked(Box>), + DenseChunkedByte(Box>), + DenseChunkedHalf(Box>), MultiDenseChunked(Box>), MultiDenseChunkedByte(Box>), MultiDenseChunkedHalf(Box>), Sparse(Box), } -impl VectorStorageRead for VectorStorageReadEnum { +impl VectorStorageRead for VectorStorageReadEnum { fn distance(&self) -> Distance { match self { VectorStorageReadEnum::Dense(s) => s.distance(), diff --git a/src/tonic/api/storage_read_api/helpers.rs b/src/tonic/api/storage_read_api/helpers.rs index 616b642358..56533f8bab 100644 --- a/src/tonic/api/storage_read_api/helpers.rs +++ b/src/tonic/api/storage_read_api/helpers.rs @@ -12,7 +12,7 @@ use tonic::Status; use crate::tonic::api::storage_read_api::StorageReadService; -impl + Send + Sync + 'static> StorageReadService { +impl StorageReadService { pub fn new(dispatcher: Arc) -> Self { Self { dispatcher, diff --git a/src/tonic/api/storage_read_api/mod.rs b/src/tonic/api/storage_read_api/mod.rs index b59a19c452..571c7cfb46 100644 --- a/src/tonic/api/storage_read_api/mod.rs +++ b/src/tonic/api/storage_read_api/mod.rs @@ -29,13 +29,13 @@ mod tests; /// Chunk size for streaming reads (~1 MB). const STREAM_CHUNK_SIZE: u64 = 1024 * 1024; -pub struct StorageReadService + Send + Sync + 'static = MmapFile> { +pub struct StorageReadService { dispatcher: Arc, _marker: PhantomData, } #[async_trait] -impl + Send + Sync + 'static> StorageRead for StorageReadService { +impl StorageRead for StorageReadService { // Check if a file exists via UniversalRead::open(), catch NotFound → false. async fn file_exists( &self, @@ -127,7 +127,7 @@ impl + Send + Sync + 'static> StorageRead for StorageReadSe let open_options = OpenOptions::default(); let length = tokio::task::spawn_blocking(move || { let storage = S::open(&path, open_options).map_err(io_error_to_status)?; - storage.len().map_err(io_error_to_status) + storage.len::().map_err(io_error_to_status) }) .await .map_err(|e| Status::internal(format!("Task join error: {e}")))??; @@ -159,7 +159,7 @@ impl + Send + Sync + 'static> StorageRead for StorageReadSe let data = tokio::task::spawn_blocking(move || { let storage = S::open(&path, open_options).map_err(io_error_to_status)?; let cow = storage - .read::(ReadRange::new(byte_offset, length)) + .read::(ReadRange::new(byte_offset, length)) .map_err(io_error_to_status)?; Ok::<_, Status>(cow.into_owned()) }) @@ -196,7 +196,7 @@ impl + Send + Sync + 'static> StorageRead for StorageReadSe let range = ReadRange::new(byte_offset, length); let (storage, range) = tokio::task::spawn_blocking(move || { let s = S::open(&path, open_options).map_err(io_error_to_status)?; - let file_len = s.len().map_err(io_error_to_status)?; + let file_len = s.len::().map_err(io_error_to_status)?; validate_range(range, file_len).map_err(io_error_to_status)?; Ok::<_, Status>((s, range)) }) @@ -217,7 +217,7 @@ impl + Send + Sync + 'static> StorageRead for StorageReadSe let data = tokio::task::spawn_blocking(move || { storage_for_read - .read::(ReadRange::new(current_offset, chunk_size)) + .read::(ReadRange::new(current_offset, chunk_size)) .map(|cow| cow.into_owned()) .map_err(io_error_to_status) }) @@ -254,7 +254,7 @@ impl + Send + Sync + 'static> StorageRead for StorageReadSe let data = tokio::task::spawn_blocking(move || { let storage = S::open(&path, open_options).map_err(io_error_to_status)?; - let cow = storage.read_whole().map_err(io_error_to_status)?; + let cow = storage.read_whole::().map_err(io_error_to_status)?; Ok::<_, Status>(cow.into_owned()) }) .await @@ -291,7 +291,7 @@ impl + Send + Sync + 'static> StorageRead for StorageReadSe let storage = S::open(&path, open_options).map_err(io_error_to_status)?; let mut results = ranges.iter().map(|_| Vec::new()).collect::>(); storage - .read_batch::(ranges.into_iter().enumerate(), |idx, chunk| { + .read_batch::(ranges.into_iter().enumerate(), |idx, chunk| { results[idx].extend_from_slice(chunk); Ok(()) }) @@ -352,7 +352,7 @@ impl + Send + Sync + 'static> StorageRead for StorageReadSe .enumerate() .map(|(op_idx, (file_idx, range))| (op_idx, &files[file_idx], range)); - S::read_multi::(reads, |op_idx, chunk| { + S::read_multi::(reads, |op_idx, chunk| { results[op_idx].extend_from_slice(chunk); Ok(()) })