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

claude wip

Browse Source
This commit is contained in:
Luis Cossío
2026-05-06 18:19:18 -04:00
parent cd8f21b391
commit a76870eb94
41 changed files with 469 additions and 432 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
lib/common/common/benches/universal_io.rs
View File

@@ -61,7 +61,7 @@ fn benches(c: &mut Criterion) {
}
}
fn read_benches<T: bytemuck::Pod, C: UniversalRead<T>>(
fn read_benches<T: bytemuck::Pod, C: UniversalRead>(
c: &mut Criterion,
impl_name: &str, // Corresponds to `C`
elem_size: &str, // Corresponds to `T`
@@ -78,7 +78,7 @@ fn read_benches<T: bytemuck::Pod, C: UniversalRead<T>>(
let storage = C::open(path, options).unwrap();
let len = FILE_SIZE_BYTES / size_of::<T>() as u64;
let mut rng = rand::rng();
assert_eq!(storage.len().unwrap(), len);
assert_eq!(storage.len::<T>().unwrap(), len);
let low_mem = std::env::var_os(LIMIT_MEMORY_ENV_INTERNAL).is_some();
@@ -96,7 +96,7 @@ fn read_benches<T: bytemuck::Pod, C: UniversalRead<T>>(
let mut sum = 0u64;
let offset = rng.random_range(0..len) * size_of::<T>() as u64;
let data = storage
.read::<Random>(ReadRange {
.read::<T, Random>(ReadRange {
byte_offset: offset,
length: 1,
})
@@ -119,7 +119,7 @@ fn read_benches<T: bytemuck::Pod, C: UniversalRead<T>>(
})
.map(|range| ((), range));
storage
.read_batch::<Random, ()>(ranges, |(), chunk| {
.read_batch::<T, Random, ()>(ranges, |(), chunk| {
for &item in bytemuck::cast_slice::<T, u64>(chunk) {
sum = sum.wrapping_add(item);
}
@@ -142,7 +142,7 @@ fn read_benches<T: bytemuck::Pod, C: UniversalRead<T>>(
})
.map(|range| ((), range));
storage
.read_batch::<Sequential, ()>(ranges, |(), chunk| {
.read_batch::<T, Sequential, ()>(ranges, |(), chunk| {
for &item in bytemuck::cast_slice::<T, u64>(chunk) {
sum = sum.wrapping_add(item);
}
@@ -158,7 +158,7 @@ fn read_benches<T: bytemuck::Pod, C: UniversalRead<T>>(
let read_batch_full = || {
let mut sum = 0u64;
storage
.read_batch::<Sequential, ()>(ranges_full_file::<T>(), |(), chunk| {
.read_batch::<T, Sequential, ()>(ranges_full_file::<T>(), |(), chunk| {
for &item in bytemuck::cast_slice::<T, u64>(chunk) {
sum = sum.wrapping_add(item);
}

23
lib/common/common/src/stored_bitslice.rs
View File

@@ -1,5 +1,5 @@
//! Storage-agnostic bitslice backed by any [`UniversalRead<u64>`] /
//! [`UniversalWrite<u64>`] 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<crate::universal_io::MmapFile>;
/// A storage-agnostic bitslice that supports both reading and writing bits.
///
/// Wraps any [`UniversalRead<u64>`] / [`UniversalWrite<u64>`] 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<S> {
storage: S,
storage: TypedStorage<S, BitStore>,
/// Total number of `BitStore` elements in the underlying storage.
element_len: u64,
}
impl<S: UniversalRead<BitStore>> StoredBitSlice<S> {
impl<S: UniversalRead> StoredBitSlice<S> {
/// Open a bitslice storage from the given path using backend `S`.
pub fn open(path: impl AsRef<Path>, options: OpenOptions) -> Result<Self> {
let storage = S::open(path, options)?;
let storage = TypedStorage::<S, BitStore>::open(path, options)?;
let element_len = storage.len()?;
Ok(Self {
storage,
@@ -171,7 +172,7 @@ impl<S: UniversalRead<BitStore>> StoredBitSlice<S> {
}
}
impl<S: UniversalWrite<u64>> StoredBitSlice<S> {
impl<S: UniversalWrite> StoredBitSlice<S> {
/// Set multiple individual bits in a batch.
///
/// Each `(bit_index, value)` pair sets a single bit. Bits within the same

40
lib/common/common/src/universal_io/disk_cache/cached_slice.rs
View File

@@ -49,22 +49,40 @@ impl<T: bytemuck::Pod> CachedSlice<T> {
/// guarantees correct alignment for any `T`.
pub fn get_range(&self, range: Range<usize>) -> io::Result<Cow<'_, [T]>> {
let t_size = size_of::<T>();
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::<T>(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::<T2>()`. Single-block
/// reads return a zero-copy borrow when possible; multi-block reads
/// allocate a `Vec<T2>` for correct alignment.
pub(crate) fn read_byte_range_as<T2: bytemuck::Pod>(
&self,
byte_range: Range<usize>,
) -> io::Result<Cow<'_, [T2]>> {
let t2_size = size_of::<T2>();
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::<T2>()",
);
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::<T, u8>(&mut vec_t).copy_from_slice(bytes);
let mut vec_t = vec![T2::zeroed(); bytes.len() / t2_size];
bytemuck::cast_slice_mut::<T2, u8>(&mut vec_t).copy_from_slice(bytes);
vec_t
})?;
@@ -74,9 +92,9 @@ impl<T: bytemuck::Pod> CachedSlice<T> {
});
}
// Multi-block: allocate Vec<T> directly for correct alignment.
let mut result = vec![T::zeroed(); total_elements];
let result_bytes = bytemuck::cast_slice_mut::<T, u8>(&mut result);
// Multi-block: allocate Vec<T2> directly for correct alignment.
let mut result = vec![T2::zeroed(); total_elements];
let result_bytes = bytemuck::cast_slice_mut::<T2, u8>(&mut result);
let mut copied = 0;
let mut copy_block = |slice: &[u8]| {
let end = copied + slice.len();
@@ -145,6 +163,10 @@ impl<T: bytemuck::Pod> CachedSlice<T> {
self.len_bytes / size_of::<T>()
}
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<usize>) -> impl ExactSizeIterator<Item = BlockRequest> {
debug_assert!(bytes_range.start <= bytes_range.end);

52
lib/common/common/src/universal_io/disk_cache/mod.rs
View File

@@ -66,14 +66,15 @@ impl<T> UniversalReadFileOps for CachedSlice<T> {
}
}
impl<T> UniversalRead<T> for CachedSlice<T>
impl<T> UniversalRead for CachedSlice<T>
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<Path>, options: OpenOptions) -> Result<Self> {
let Some(controller) = CacheController::global() else {
@@ -97,18 +98,22 @@ where
Ok(CachedSlice::open(controller, path.as_ref())?)
}
fn read<P: AccessPattern>(&self, range: ReadRange) -> Result<Cow<'_, [T]>> {
let elem_start = usize::try_from(range.byte_offset).expect("range.start is within usize")
/ size_of::<T>();
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<U, P>(&self, range: ReadRange) -> Result<Cow<'_, [U]>>
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::<U>();
Ok(self.read_byte_range_as::<U>(byte_start..byte_start + byte_length)?)
}
fn len(&self) -> Result<u64> {
Ok(Self::len(self) as u64)
fn len<U>(&self) -> Result<u64>
where
U: bytemuck::Pod,
{
Ok((self.byte_len() / size_of::<U>()) 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<T>>,
}
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<T>;
fn new() -> Result<Self> {
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::<Random>(range)?));
self.result = Some((meta, file.read::<U, Random>(range)?));
Ok(())
}
fn wait(&mut self) -> Result<Option<(Meta, Cow<'file, [T]>)>> {
fn wait(&mut self) -> Result<Option<(Meta, Cow<'file, [U]>)>> {
Ok(self.result.take())
}
}

29
lib/common/common/src/universal_io/family.rs
View File

@@ -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<T>` where `S` is a
/// generic parameter. This trait stands in: `S: UniversalReadFamily` then
/// `S::Read<T>` yields the concrete `UniversalRead<T>` for element type `T`.
pub trait UniversalReadFamily {
type Read<T: bytemuck::Pod + 'static>: UniversalRead<T>;
}
pub struct MmapFamily;
impl UniversalReadFamily for MmapFamily {
type Read<T: bytemuck::Pod + 'static> = crate::universal_io::mmap::MmapFile;
}
#[cfg(target_os = "linux")]
pub struct IoUringFamily;
#[cfg(target_os = "linux")]
impl UniversalReadFamily for IoUringFamily {
type Read<T: bytemuck::Pod + 'static> = 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<T: bytemuck::Pod + 'static> = crate::universal_io::disk_cache::CachedSlice<T>;
}

47
lib/common/common/src/universal_io/io_uring/mod.rs
View File

@@ -48,11 +48,11 @@ impl UniversalReadFileOps for IoUringFile {
}
}
impl<T> UniversalRead<T> 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<Path>, options: OpenOptions) -> Result<Self> {
// Check that io_uring is supported on this system.
@@ -86,11 +86,15 @@ where
Ok(file)
}
fn read<P: AccessPattern>(&self, range: ReadRange) -> Result<Cow<'_, [T]>> {
fn read<T, P>(&self, range: ReadRange) -> Result<Cow<'_, [T]>>
where
T: bytemuck::Pod,
P: AccessPattern,
{
if self.direct_io {
// direct_io needs special handling
return self
.read_iter::<P, _>([((), range)])?
.read_iter::<T, P, _>([((), range)])?
.next()
.expect("there's exactly one read")
.map(|(_, data)| data);
@@ -102,7 +106,10 @@ where
Ok(Cow::Owned(items))
}
fn len(&self) -> Result<u64> {
fn len<T>(&self) -> Result<u64>
where
T: bytemuck::Pod,
{
let byte_len = self.file.metadata()?.len();
let items_len = byte_len / size_of::<T>() as u64;
@@ -207,20 +214,23 @@ where
}
}
impl<T> UniversalWrite<T> for IoUringFile
where
T: bytemuck::Pod,
{
fn write(&mut self, byte_offset: ByteOffset, items: &[T]) -> Result<()> {
impl UniversalWrite for IoUringFile {
fn write<T>(&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<Item = (ByteOffset, &'a [T])>,
) -> 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<Item = (FileIndex, ByteOffset, &'a [T])>,
) -> Result<()> {
) -> Result<()>
where
T: bytemuck::Pod + 'a,
{
let mut rt = IoUringRuntime::new()?;
let mut writes = writes.into_iter().peekable();

18
lib/common/common/src/universal_io/io_uring/tests.rs
View File

@@ -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 = <IoUringFile as UniversalRead<u64>>::open(&path_0, opts)?;
let file_1 = <IoUringFile as UniversalRead<u64>>::open(&path_1, opts)?;
let file_0 = <IoUringFile as UniversalRead>::open(&path_0, opts)?;
let file_1 = <IoUringFile as UniversalRead>::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<u64>)> = Vec::new();
for record in IoUringFile::read_multi_iter::<Sequential, _>(reads)? {
for record in IoUringFile::read_multi_iter::<u64, Sequential, _>(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 = <IoUringFile as UniversalRead<u64>>::open(&path, opts)?;
let file = <IoUringFile as UniversalRead>::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<u64>)> = Vec::new();
for record in IoUringFile::read_multi_iter::<Sequential, _>(reads)? {
for record in IoUringFile::read_multi_iter::<u64, Sequential, _>(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::<u64>::open(&path_a, opts)?;
let file_b: IoUringFile = UniversalRead::<u64>::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<u64>)> = Vec::new();
IoUringFile::read_multi::<Sequential, _>(reads.clone(), |idx, data| {
IoUringFile::read_multi::<u64, Sequential, _>(reads.clone(), |idx, data| {
callback_results.push((idx, data.to_vec()));
Ok(())
})?;
// Collect via iterator.
let mut iter_results: Vec<(usize, Vec<u64>)> = Vec::new();
for record in IoUringFile::read_multi_iter::<Sequential, _>(reads)? {
for record in IoUringFile::read_multi_iter::<u64, Sequential, _>(reads)? {
let (idx, cow) = record?;
iter_results.push((idx, cow.into_owned()));
}

40
lib/common/common/src/universal_io/mmap.rs
View File

@@ -43,11 +43,11 @@ impl UniversalReadFileOps for MmapFile {
}
}
impl<T> UniversalRead<T> 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<Path>, options: OpenOptions) -> Result<Self> {
let OpenOptions {
@@ -100,13 +100,20 @@ where
Ok(mmap)
}
fn read<P: AccessPattern>(&self, range: ReadRange) -> Result<Cow<'_, [T]>> {
fn read<T, P>(&self, range: ReadRange) -> Result<Cow<'_, [T]>>
where
T: bytemuck::Pod,
P: AccessPattern,
{
let mmap = self.as_bytes::<P>();
let items = read(mmap, range)?;
Ok(Cow::Borrowed(items))
}
fn len(&self) -> Result<u64> {
fn len<T>(&self) -> Result<u64>
where
T: bytemuck::Pod,
{
let len = self.len / size_of::<T>();
Ok(len as u64)
}
@@ -165,20 +172,23 @@ where
}
}
impl<T> UniversalWrite<T> for MmapFile
where
T: bytemuck::Pod,
{
fn write(&mut self, byte_offset: ByteOffset, items: &[T]) -> Result<()> {
impl UniversalWrite for MmapFile {
fn write<T>(&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<Item = (ByteOffset, &'a [T])>,
) -> 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<Path>) -> std::io::Result<(u64, u64)> {
let file: Self = <Self as UniversalRead<u8>>::open(
let file: Self = <Self as UniversalRead>::open(
path,
OpenOptions {
writeable: false,

6
lib/common/common/src/universal_io/mod.rs
View File

@@ -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<T, E = UniversalIoError> = std::result::Result<T, E>;
/// Uses a single logical read when the backend overrides [`UniversalRead::read_whole`].
pub fn read_json_via<S, T>(path: impl AsRef<Path>) -> Result<T>
where
S: UniversalRead<u8>,
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::<u8>()?;
serde_json::from_slice(&bytes).map_err(UniversalIoError::from)
}

51
lib/common/common/src/universal_io/read.rs
View File

@@ -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<T>: 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<Path>, options: OpenOptions) -> Result<Self>;
/// Prefer [`read_batch`] if you need high performance.
fn read<P: AccessPattern>(&self, range: ReadRange) -> Result<Cow<'_, [T]>>;
fn read<T, P>(&self, range: ReadRange) -> Result<Cow<'_, [T]>>
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<Cow<'_, [T]>> {
fn read_whole<T>(&self) -> Result<Cow<'_, [T]>>
where
T: bytemuck::Pod,
{
let range = ReadRange {
byte_offset: 0,
length: self.len()?,
length: self.len::<T>()?,
};
self.read::<Sequential>(range)
self.read::<T, Sequential>(range)
}
fn read_batch<P, Meta>(
fn read_batch<T, P, Meta>(
&self,
ranges: impl IntoIterator<Item = (Meta, ReadRange)>,
mut callback: impl FnMut(Meta, &[T]) -> Result<()>,
) -> Result<()>
where
T: bytemuck::Pod,
P: AccessPattern,
{
for record in self.read_iter::<P, Meta>(ranges)? {
for record in self.read_iter::<T, P, Meta>(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<P, Meta>(
fn read_iter<T, P, Meta>(
&self,
ranges: impl IntoIterator<Item = (Meta, ReadRange)>,
) -> Result<impl Iterator<Item = Result<(Meta, Cow<'_, [T]>)>>>
where
T: bytemuck::Pod,
P: AccessPattern,
{
let reads = ranges
.into_iter()
.map(move |(meta, range)| (meta, self, range));
Self::read_multi_iter::<P, Meta>(reads)
Self::read_multi_iter::<T, P, Meta>(reads)
}
fn len(&self) -> Result<u64>;
/// Number of `T`-sized elements in the underlying storage.
fn len<T>(&self) -> Result<u64>
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<Item = (Meta, &'a Self, ReadRange)>,
mut callback: impl FnMut(Meta, &[T]) -> Result<()>,
) -> Result<()>
where
T: bytemuck::Pod,
P: AccessPattern,
Self: 'a,
{
for record in Self::read_multi_iter::<P, Meta>(reads)? {
for record in Self::read_multi_iter::<T, P, Meta>(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<Item = (Meta, &'a Self, ReadRange)>,
) -> Result<impl Iterator<Item = Result<(Meta, Cow<'a, [T]>)>>>
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;

16
lib/common/common/src/universal_io/wrappers/buffered_update.rs
View File

@@ -15,8 +15,8 @@ use crate::universal_io::{Flusher, UniversalIoError, UniversalWrite};
#[derive(Debug)]
pub struct SliceBufferedUpdateWrapper<S, T>
where
S: UniversalWrite<T>,
T: Copy + 'static,
S: UniversalWrite,
T: bytemuck::Pod,
{
slice: Arc<RwLock<S>>,
len: u64,
@@ -26,11 +26,11 @@ where
impl<S, T> SliceBufferedUpdateWrapper<S, T>
where
S: UniversalWrite<T>,
T: Copy + 'static,
S: UniversalWrite,
T: bytemuck::Pod,
{
pub fn new(slice_storage: S) -> Result<Self, UniversalIoError> {
let len = slice_storage.len()?;
let len = slice_storage.len::<T>()?;
Ok(Self {
slice: Arc::new(RwLock::new(slice_storage)),
len,
@@ -55,8 +55,8 @@ where
impl<S, T> SliceBufferedUpdateWrapper<S, T>
where
S: UniversalWrite<T> + 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::<T>(it)?;
slice_guard.flusher()()?;

54
lib/common/common/src/universal_io/wrappers/read_only.rs
View File

@@ -28,15 +28,15 @@ where
}
}
impl<S, T> UniversalRead<T> for ReadOnly<S>
impl<S> UniversalRead for ReadOnly<S>
where
S: UniversalRead<T>,
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<Path>, options: OpenOptions) -> Result<Self> {
@@ -46,41 +46,53 @@ where
}
#[inline]
fn read<P: AccessPattern>(&self, range: ReadRange) -> Result<Cow<'_, [T]>> {
self.0.read::<P>(range)
fn read<T, P>(&self, range: ReadRange) -> Result<Cow<'_, [T]>>
where
T: bytemuck::Pod,
P: AccessPattern,
{
self.0.read::<T, P>(range)
}
#[inline]
fn read_whole(&self) -> Result<Cow<'_, [T]>> {
self.0.read_whole()
fn read_whole<T>(&self) -> Result<Cow<'_, [T]>>
where
T: bytemuck::Pod,
{
self.0.read_whole::<T>()
}
#[inline]
fn read_batch<P, Meta>(
fn read_batch<T, P, Meta>(
&self,
ranges: impl IntoIterator<Item = (Meta, ReadRange)>,
callback: impl FnMut(Meta, &[T]) -> Result<()>,
) -> Result<()>
where
T: bytemuck::Pod,
P: AccessPattern,
{
self.0.read_batch::<P, Meta>(ranges, callback)
self.0.read_batch::<T, P, Meta>(ranges, callback)
}
#[inline]
fn read_iter<P, Meta>(
fn read_iter<T, P, Meta>(
&self,
ranges: impl IntoIterator<Item = (Meta, ReadRange)>,
) -> Result<impl Iterator<Item = Result<(Meta, Cow<'_, [T]>)>>>
where
T: bytemuck::Pod,
P: AccessPattern,
{
self.0.read_iter::<P, Meta>(ranges)
self.0.read_iter::<T, P, Meta>(ranges)
}
#[inline]
fn len(&self) -> Result<u64> {
self.0.len()
fn len<T>(&self) -> Result<u64>
where
T: bytemuck::Pod,
{
self.0.len::<T>()
}
#[inline]
@@ -94,11 +106,12 @@ where
}
#[inline]
fn read_multi<'a, P, Meta>(
fn read_multi<'a, T, P, Meta>(
reads: impl IntoIterator<Item = (Meta, &'a Self, ReadRange)>,
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::<P, _>(reads, callback)
S::read_multi::<T, P, _>(reads, callback)
}
#[inline]
fn read_multi_iter<'a, P, Meta>(
fn read_multi_iter<'a, T, P, Meta>(
reads: impl IntoIterator<Item = (Meta, &'a Self, ReadRange)>,
) -> Result<impl Iterator<Item = Result<(Meta, Cow<'a, [T]>)>>>
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::<P, _>(it)
S::read_multi_iter::<T, P, _>(it)
}
fn kind() -> UniversalKind {

6
lib/common/common/src/universal_io/wrappers/stored_struct.rs
View File

@@ -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<S, T> {
impl<S, T> StoredStruct<S, T>
where
T: bytemuck::Pod + Send,
S: UniversalWrite<T> + Send + 'static,
S: UniversalWrite + Send + 'static,
{
pub fn open(path: impl AsRef<Path>, options: OpenOptions) -> universal_io::Result<Self> {
let storage = TypedStorage::open(path, options)?;

95
lib/common/common/src/universal_io/wrappers/typed.rs
View File

@@ -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<f32>`), but it helps the compiler to
/// distinguish when more than one bound is used, e.g.
/// `where S: UniversalRead<f32> + UniversalRead<PointOffsetType> + …`.
/// `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<S, T> TypedStorage<S, T> {
}
}
impl<S, T> UniversalReadFileOps for TypedStorage<S, T>
impl<S, T> TypedStorage<S, T>
where
S: UniversalReadFileOps,
{
#[inline]
fn list_files(prefix_path: &Path) -> Result<Vec<PathBuf>> {
pub fn list_files(prefix_path: &Path) -> Result<Vec<PathBuf>> {
S::list_files(prefix_path)
}
#[inline]
fn exists(path: &Path) -> Result<bool> {
pub fn exists(path: &Path) -> Result<bool> {
S::exists(path)
}
}
impl<S, T> UniversalRead<T> for TypedStorage<S, T>
impl<S, T> TypedStorage<S, T>
where
S: UniversalRead<T>,
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<Path>, options: OpenOptions) -> Result<Self> {
pub fn open(path: impl AsRef<Path>, options: OpenOptions) -> Result<Self> {
S::open(path, options).map(|inner| TypedStorage {
inner,
_phantom: PhantomData,
@@ -68,17 +62,17 @@ where
}
#[inline]
fn read<P: AccessPattern>(&self, range: ReadRange) -> Result<Cow<'_, [T]>> {
self.inner.read::<P>(range)
pub fn read<P: AccessPattern>(&self, range: ReadRange) -> Result<Cow<'_, [T]>> {
self.inner.read::<T, P>(range)
}
#[inline]
fn read_whole(&self) -> Result<Cow<'_, [T]>> {
self.inner.read_whole()
pub fn read_whole(&self) -> Result<Cow<'_, [T]>> {
self.inner.read_whole::<T>()
}
#[inline]
fn read_batch<P, Meta>(
pub fn read_batch<P, Meta>(
&self,
ranges: impl IntoIterator<Item = (Meta, ReadRange)>,
callback: impl FnMut(Meta, &[T]) -> Result<()>,
@@ -86,37 +80,38 @@ where
where
P: AccessPattern,
{
self.inner.read_batch::<P, Meta>(ranges, callback)
self.inner.read_batch::<T, P, Meta>(ranges, callback)
}
#[inline]
fn read_iter<P, Meta>(
pub fn read_iter<P, Meta>(
&self,
ranges: impl IntoIterator<Item = (Meta, ReadRange)>,
) -> Result<impl Iterator<Item = Result<(Meta, Cow<'_, [T]>)>>>
where
P: AccessPattern,
{
self.inner.read_iter::<P, Meta>(ranges)
self.inner.read_iter::<T, P, Meta>(ranges)
}
#[inline]
fn len(&self) -> Result<u64> {
self.inner.len()
#[expect(clippy::len_without_is_empty)]
pub fn len(&self) -> Result<u64> {
self.inner.len::<T>()
}
#[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<Item = (Meta, &'a Self, ReadRange)>,
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::<T, P, _>(reads, callback)
}
#[inline]
fn read_multi_iter<'a, P, Meta>(
pub fn read_multi_iter<'a, P, Meta>(
reads: impl IntoIterator<Item = (Meta, &'a Self, ReadRange)>,
) -> Result<impl Iterator<Item = Result<(Meta, Cow<'a, [T]>)>>>
where
@@ -143,42 +138,48 @@ where
.into_iter()
.map(|(meta, file, range)| (meta, &file.inner, range));
S::read_multi_iter::<P, _>(reads)
S::read_multi_iter::<T, P, _>(reads)
}
fn kind() -> UniversalKind {
pub fn kind() -> UniversalKind {
S::kind()
}
}
impl<S, T> UniversalWrite<T> for TypedStorage<S, T>
impl<S, T> TypedStorage<S, T>
where
S: UniversalWrite<T>,
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<Item = (ByteOffset, &'a [T])>,
) -> Result<()> {
self.inner.write_batch(offset_data)
) -> Result<()>
where
T: 'a,
{
self.inner.write_batch::<T>(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<Item = (FileIndex, ByteOffset, &'a [T])>,
) -> Result<()> {
S::write_multi(Self::peel_slice_mut(files), writes)
) -> Result<()>
where
T: 'a,
{
S::write_multi::<T>(Self::peel_slice_mut(files), writes)
}
}

2
lib/common/common/src/universal_io/wrappers/wrapped_pipeline.rs
View File

@@ -18,7 +18,7 @@ impl<'a, File, Inner, T, Meta> UniversalReadPipeline<'a, T, Meta>
where
File: TransparentWrapper<Inner::File>,
Inner: UniversalReadPipeline<'a, T, Meta>,
T: Copy + 'static,
T: bytemuck::Pod,
{
type File = File;

19
lib/common/common/src/universal_io/write.rs
View File

@@ -1,21 +1,28 @@
use super::read::UniversalRead;
use super::*;
pub trait UniversalWrite<T: Copy + 'static>: UniversalRead<T> {
fn write(&mut self, byte_offset: ByteOffset, data: &[T]) -> Result<()>;
pub trait UniversalWrite: UniversalRead {
fn write<T>(&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<Item = (ByteOffset, &'a [T])>,
) -> 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<Item = (FileIndex, ByteOffset, &'a [T])>,
) -> Result<()> {
) -> Result<()>
where
T: bytemuck::Pod + 'a,
{
let files_len = files.len();
for (file_index, offset, data) in writes {

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

@@ -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<S> {
path: PathBuf,
config: StorageConfig,
slice_store: S,
slice_store: TypedStorage<S, RegionGaps>,
}
impl<S: UniversalWrite<RegionGaps>> BitmaskGaps<S> {
impl<S: UniversalWrite> BitmaskGaps<S> {
pub fn path(&self) -> PathBuf {
self.path.clone()
}
@@ -110,7 +110,7 @@ impl<S: UniversalWrite<RegionGaps>> BitmaskGaps<S> {
advice: None,
prevent_caching: None,
};
let mut slice_store = S::open(&path, options)?;
let mut slice_store = TypedStorage::<S, RegionGaps>::open(&path, options)?;
debug_assert_eq!(slice_store.len()? as usize, data.len());
@@ -133,7 +133,7 @@ impl<S: UniversalWrite<RegionGaps>> BitmaskGaps<S> {
advice: Some(AdviceSetting::Advice(Advice::Normal)),
prevent_caching: None,
};
let slice_store = S::open(&path, options)?;
let slice_store = TypedStorage::<S, RegionGaps>::open(&path, options)?;
Ok(Self {
path,
@@ -168,7 +168,7 @@ impl<S: UniversalWrite<RegionGaps>> BitmaskGaps<S> {
advice: Some(AdviceSetting::Advice(Advice::Normal)),
prevent_caching: None,
};
self.slice_store = S::open(&self.path, options)?;
self.slice_store = TypedStorage::<S, RegionGaps>::open(&self.path, options)?;
debug_assert_eq!(self.len()? - prev_len, data.len());

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

@@ -32,8 +32,8 @@ type RegionId = u32;
/// Concrete bitmask type using memory-mapped storage.
pub type MmapBitmask = Bitmask<MmapFile>;
pub trait BitmaskStorage: UniversalWrite<RegionGaps> + UniversalWrite<u64> {}
impl<T> BitmaskStorage for T where T: UniversalWrite<RegionGaps> + UniversalWrite<u64> {}
pub trait BitmaskStorage: UniversalWrite {}
impl<T> BitmaskStorage for T where T: UniversalWrite {}
#[derive(Debug)]
pub struct Bitmask<S> {

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

@@ -31,14 +31,14 @@ pub(super) fn decompress_lz4(value: &[u8]) -> Vec<u8> {
/// [`Tracker<S>`]).
///
/// Constructed from either [`super::Gridstore`] or [`super::GridstoreReader`].
pub struct GridstoreView<'a, V, S: UniversalRead<u8>> {
pub struct GridstoreView<'a, V, S: UniversalRead> {
pub(super) config: &'a StorageConfig,
pub(super) tracker: &'a Tracker<S>,
pub(super) pages: &'a Pages<S>,
pub(super) _value_type: std::marker::PhantomData<V>,
}
impl<'a, V, S: UniversalRead<u8>> GridstoreView<'a, V, S> {
impl<'a, V, S: UniversalRead> GridstoreView<'a, V, S> {
pub(crate) fn new(
config: &'a StorageConfig,
tracker: &'a Tracker<S>,
@@ -74,7 +74,7 @@ impl<'a, V, S: UniversalRead<u8>> GridstoreView<'a, V, S> {
}
}
impl<'a, V: Blob, S: UniversalRead<u8>> GridstoreView<'a, V, S> {
impl<'a, V: Blob, S: UniversalRead> GridstoreView<'a, V, S> {
pub(super) fn compress(&self, value: Vec<u8>) -> Vec<u8> {
match self.config.compression {
Compression::None => value,

10
lib/gridstore/src/pages.rs
View File

@@ -31,7 +31,7 @@ impl<S> Pages<S> {
}
}
impl<S: UniversalRead<u8>> Pages<S> {
impl<S: UniversalRead> Pages<S> {
pub fn new(base_path: PathBuf) -> Self {
Self {
base_path,
@@ -213,7 +213,7 @@ impl<S: UniversalRead<u8>> Pages<S> {
vec![MaybeUninit::uninit(); pointer.length as _]
};
for result in S::read_multi_iter::<P, _>(reads)? {
for result in S::read_multi_iter::<u8, P, _>(reads)? {
let (offset, bytes) = result?;
if pages == 1 {
@@ -282,7 +282,7 @@ impl<S: UniversalRead<u8>> Pages<S> {
})
});
let chunks = S::read_multi_iter::<P, _>(reads).map_err(GridstoreError::from)?;
let chunks = S::read_multi_iter::<u8, P, _>(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<S: UniversalRead<u8>> Pages<S> {
}
}
impl<S: UniversalWrite<u8>> Pages<S> {
impl<S: UniversalWrite> Pages<S> {
pub fn write_to_pages(
&mut self,
pointer: ValuePointer,
@@ -398,7 +398,7 @@ impl<S: UniversalWrite<u8>> Pages<S> {
});
// Execute writes (mutable borrow of self.pages)
S::write_multi(self.pages.as_mut_slice(), writes)?;
S::write_multi::<u8>(self.pages.as_mut_slice(), writes)?;
Ok(())
}

76
lib/gridstore/src/tracker.rs
View File

@@ -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<T> {
value: T,
}
// SAFETY: `Optional<T>` 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<T: bytemuck::Pod> bytemuck::Zeroable for Optional<T> {}
unsafe impl<T: bytemuck::Pod> bytemuck::Pod for Optional<T> {}
impl<T: FromZeros> From<Option<T>> for Optional<T> {
fn from(value: Option<T>) -> Self {
match value {
@@ -81,7 +86,9 @@ impl<T: FromZeros> Optional<T> {
}
}
#[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<S> Tracker<S> {
}
// Read operations -- only require UniversalRead
impl<S: UniversalRead<u8>> Tracker<S> {
impl<S: UniversalRead> Tracker<S> {
/// Open an existing PageTracker at the given path
/// If the file does not exist, return an error
pub fn open(path: &Path) -> Result<Self> {
@@ -286,12 +293,12 @@ impl<S: UniversalRead<u8>> Tracker<S> {
}
fn read_header(storage: &S) -> Result<TrackerHeader> {
let header_bytes = storage.read::<Random>(ReadRange {
let header_bytes = storage.read::<TrackerHeader, Random>(ReadRange {
byte_offset: 0,
length: std::mem::size_of::<TrackerHeader>() as u64,
})?;
#[expect(deprecated, reason = "legacy code")]
Ok(*unsafe { transmute_from_u8::<TrackerHeader>(header_bytes.as_ref()) })
})?[0];
Ok(header_bytes)
}
fn open_storage(path: &Path) -> Result<S> {
@@ -345,16 +352,16 @@ impl<S: UniversalRead<u8>> Tracker<S> {
let start_offset = std::mem::size_of::<TrackerHeader>()
+ point_offset as usize * std::mem::size_of::<Optional<ValuePointer>>();
let end_offset = start_offset + std::mem::size_of::<Optional<ValuePointer>>();
let storage_len = self.storage.len()?;
let storage_len = self.storage.len::<u8>()?;
if end_offset as u64 > storage_len {
return Ok(None);
}
let bytes = self.storage.read::<Random>(ReadRange {
byte_offset: start_offset as u64,
length: std::mem::size_of::<Optional<ValuePointer>>() as u64,
})?;
#[expect(deprecated, reason = "legacy code")]
let opt: &Optional<ValuePointer> = unsafe { transmute_from_u8(bytes.as_ref()) };
let opt = self
.storage
.read::<Optional<ValuePointer>, Random>(ReadRange {
byte_offset: start_offset as u64,
length: 1,
})?[0];
Ok(Some(opt.is_some().copied()))
}
@@ -382,7 +389,7 @@ impl<S: UniversalRead<u8>> Tracker<S> {
pub fn get_batch(&self, point_offsets: &[PointOffset]) -> Result<Vec<Option<ValuePointer>>> {
let item_size = std::mem::size_of::<Optional<ValuePointer>>();
let header_size = std::mem::size_of::<TrackerHeader>();
let storage_len = self.storage.len()?;
let storage_len = self.storage.len::<u8>()?;
let mut result: Vec<Option<ValuePointer>> = vec![None; point_offsets.len()];
let mut storage_reads: Vec<(usize, ReadRange)> = Vec::with_capacity(point_offsets.len());
@@ -409,7 +416,7 @@ impl<S: UniversalRead<u8>> Tracker<S> {
i,
ReadRange {
byte_offset: start_offset as u64,
length: item_size as u64,
length: 1,
},
));
}
@@ -418,11 +425,9 @@ impl<S: UniversalRead<u8>> Tracker<S> {
.iter()
.map(|(i, range)| (*i, &self.storage, *range));
for read_result in S::read_multi_iter::<Random, _>(reads)? {
let (i, bytes) = read_result?;
#[expect(deprecated, reason = "legacy code")]
let opt: &Optional<ValuePointer> = unsafe { transmute_from_u8(bytes.as_ref()) };
result[i] = opt.is_some().copied();
for read_result in S::read_multi_iter::<Optional<ValuePointer>, Random, _>(reads)? {
let (i, opts) = read_result?;
result[i] = opts[0].is_some().copied();
}
Ok(result)
@@ -470,14 +475,17 @@ impl<S: UniversalRead<u8>> Tracker<S> {
/// Return the size of the underlying file
#[cfg(test)]
pub fn mmap_file_size(&self) -> Result<usize> {
self.storage.len().map(|u| u as usize).map_err(Into::into)
self.storage
.len::<u8>()
.map(|u| u as usize)
.map_err(Into::into)
}
}
// Write operations and constructors -- require UniversalWrite
impl<S> Tracker<S>
where
S: UniversalRead<u8> + UniversalWrite<u8>,
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<ValuePointer>,
) -> Result<()> {
let storage_len = self.storage.len()? as usize;
let storage_len = self.storage.len::<u8>()? 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<ValuePointer> = unsafe { transmute_from_u8(&none_data.0) };
let none_val: &Optional<ValuePointer> = 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<ValuePointer> = unsafe { transmute_from_u8(&some_data.0) };
let some_val: &Optional<ValuePointer> = bytemuck::from_bytes(&some_data.0);
// N.B. fails on a big-endian machine.
assert_eq!(
some_val.is_some(),

2
lib/segment/src/common/buffered_update_bitslice.rs
View File

@@ -21,7 +21,7 @@ pub struct BufferedUpdateBitSlice<S> {
is_alive_flush_lock: common::is_alive_lock::IsAliveLock,
}
impl<S: UniversalWrite<u64> + Send + Sync + 'static> BufferedUpdateBitSlice<S> {
impl<S: UniversalWrite + Send + Sync + 'static> BufferedUpdateBitSlice<S> {
pub fn new(bitslice: StoredBitSlice<S>) -> Self {
let len = bitslice.bit_len() as usize;
Self {

10
lib/segment/src/common/flags/dynamic_stored_flags.rs
View File

@@ -28,14 +28,8 @@ fn status_file(directory: &Path) -> PathBuf {
directory.join(STATUS_FILE_NAME)
}
pub trait UioDynamicFlags:
UniversalWrite<DynamicFlagsStatus> + UniversalWrite<u64> + Send + 'static
{
}
impl<T> UioDynamicFlags for T where
T: UniversalWrite<DynamicFlagsStatus> + UniversalWrite<u64> + Send + 'static
{
}
pub trait UioDynamicFlags: UniversalWrite + Send + 'static {}
impl<T> UioDynamicFlags for T where T: UniversalWrite + Send + 'static {}
#[derive(Debug, Copy, Clone, bytemuck::Pod, bytemuck::Zeroable)]
#[repr(C)]

4
lib/segment/src/id_tracker/immutable_id_tracker/mod.rs
View File

@@ -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;

26
lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/types.rs
View File

@@ -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<V>` 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

26
lib/segment/src/index/field_index/full_text_index/inverted_index/mmap_inverted_index/uio_postings.rs
View File

@@ -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::<PostingsHeader>() + token_id * size_of::<PostingListHeader>()`.
/// Each [`PostingListHeader`] then points (via absolute `offset`) into the
/// posting-data region.
pub struct UniversalPostings<V: ZerocopyPostingValue, S: UniversalRead<u8>> {
pub struct UniversalPostings<V: ZerocopyPostingValue, S: UniversalRead> {
_path: PathBuf,
storage: S,
header: PostingsHeader,
@@ -40,18 +39,16 @@ struct HeadersBatch<'a> {
missing: Vec<TokenId>,
}
impl<V: ZerocopyPostingValue, S: UniversalRead<u8>> UniversalPostings<V, S> {
impl<V: ZerocopyPostingValue, S: UniversalRead> UniversalPostings<V, S> {
/// Open the postings file at `path` via the `S` storage backend.
pub fn open(path: impl Into<PathBuf>, options: OpenOptions) -> OperationResult<Self> {
let path = path.into();
let storage = S::open(&path, options)?;
let header_bytes = storage.read::<Sequential>(ReadRange {
let header = storage.read::<PostingsHeader, Sequential>(ReadRange {
byte_offset: 0,
length: size_of::<PostingsHeader>() as u64,
})?;
let (header, _) = PostingsHeader::read_from_prefix(header_bytes.as_ref())?;
length: 1,
})?[0];
Ok(Self {
_path: path,
@@ -100,18 +97,17 @@ impl<V: ZerocopyPostingValue, S: UniversalRead<u8>> UniversalPostings<V, S> {
token_id,
ReadRange {
byte_offset: header_offset,
length: header_length,
length: 1,
},
));
}
let valid_iter = self
.storage
.read_iter::<Random, _>(valid_ranges)?
.read_iter::<PostingListHeader, Random, _>(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<V: ZerocopyPostingValue, S: UniversalRead<u8>> UniversalPostings<V, S> {
byte_offset: header.offset,
length: header.posting_size::<V>() as u64,
};
let bytes = self.storage.read::<Sequential>(read_range)?;
let bytes = self.storage.read::<u8, Sequential>(read_range)?;
let result = RawPostingList::new(bytes, header);
Ok(result)
}
@@ -200,7 +196,7 @@ impl<V: ZerocopyPostingValue, S: UniversalRead<u8>> UniversalPostings<V, S> {
let mut raw_postings: Vec<(TokenId, RawPostingList<'_>)> =
Vec::with_capacity(expected_capacity);
for entry in self.storage.read_iter::<Sequential, _>(range_iter)? {
for entry in self.storage.read_iter::<u8, Sequential, _>(range_iter)? {
let ((token_id, header), bytes) = entry?;
raw_postings.push((token_id, RawPostingList::new(bytes, header)));
}

2
lib/segment/src/index/field_index/geo_index/immutable_geo_index.rs
View File

@@ -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;

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

@@ -48,21 +48,8 @@ pub(super) struct PointKeyValue {
}
/// An alias to set of traits required by [`StoredGeoMapIndex`].
#[expect(private_bounds)]
pub trait StoredGeoMapIndexStorage:
UniversalRead<u8>
+ UniversalRead<Counts>
+ UniversalRead<PointKeyValue>
+ UniversalRead<PointOffsetType>
{
}
impl<T> StoredGeoMapIndexStorage for T where
T: UniversalRead<u8>
+ UniversalRead<Counts>
+ UniversalRead<PointKeyValue>
+ UniversalRead<PointOffsetType>
{
}
pub trait StoredGeoMapIndexStorage: UniversalRead {}
impl<T> StoredGeoMapIndexStorage for T where T: UniversalRead {}
///
/// points_map
@@ -286,9 +273,10 @@ impl<S: StoredGeoMapIndexStorage> StoredGeoMapIndex<S> {
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::<S, Counts>::open(&counts_per_hash_path, open_options)?;
let points_map = TypedStorage::<S, PointKeyValue>::open(&points_map_path, open_options)?;
let points_map_ids =
TypedStorage::<S, PointOffsetType>::open(&points_map_ids_path, open_options)?;
let point_to_values = StoredPointToValues::open(path, true)?;
let mut deleted = deleted_points.to_owned();

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

@@ -51,7 +51,7 @@ pub struct MmapNumericIndex<T: Encodable + Numericable + Default + StoredValue +
pub(super) struct Storage<
T: Encodable + Numericable + Default + StoredValue + 'static,
S: UniversalRead<Point<T>> + UniversalRead<u8>,
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<T, S>,
}
impl<
T: Encodable + Numericable + Default + StoredValue + 'static,
S: UniversalRead<Point<T>> + UniversalRead<u8>,
> Storage<T, S>
{
impl<T: Encodable + Numericable + Default + StoredValue + 'static, S: UniversalRead> Storage<T, S> {
pub(crate) fn ram_usage_bytes(&self) -> usize {
let Self {
deleted,

57
lib/segment/src/index/field_index/stored_point_to_values.rs
View File

@@ -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<T: StoredValue + ?Sized, S: UniversalRead<u8>> {
pub struct StoredPointToValues<T: StoredValue + ?Sized, S: UniversalRead> {
file_name: PathBuf,
store: ReadOnly<S>,
header: Header,
@@ -79,14 +79,35 @@ pub struct StoredPointToValues<T: StoredValue + ?Sized, S: UniversalRead<u8>> {
pub const MMAP_PTV_ACCESS_OVERHEAD: usize = size_of::<MmapRange>();
#[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<T, S> StoredPointToValues<T, S>
where
T: StoredValue + ?Sized,
S: UniversalRead<u8>,
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::<MmapRange>()..,
)
.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::<Random>(ReadRange {
let header = store.read::<Header, Random>(ReadRange {
byte_offset: 0,
length: std::mem::size_of::<Header>() 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::<Random>(bytes_range)?;
let bytes = self.store.read::<u8, Random>(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::<MmapRange>();
let bytes = self.store.read::<Random>(ReadRange {
let range = self.store.read::<MmapRange, Random>(ReadRange {
byte_offset: range_offset as u64,
length: std::mem::size_of::<MmapRange>() 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::<u8>()?
}
};

4
lib/segment/src/vector_storage/chunked_vectors/chunks.rs
View File

@@ -18,7 +18,7 @@ fn check_mmap_file_name_pattern(file_name: &str) -> Option<usize> {
.and_then(|file_name| file_name.parse::<usize>().ok())
}
pub fn read_chunks<T: Copy + 'static, S: UniversalRead<T>>(
pub fn read_chunks<T: bytemuck::Pod, S: UniversalRead>(
directory: &Path,
advice: AdviceSetting,
populate: bool,
@@ -73,7 +73,7 @@ pub fn chunk_name(directory: &Path, chunk_id: usize) -> PathBuf {
))
}
pub fn create_chunk<T: Sized + Copy + 'static, S: UniversalWrite<T>>(
pub fn create_chunk<T: bytemuck::Pod, S: UniversalWrite>(
directory: &Path,
chunk_id: usize,
chunk_length_bytes: usize,

6
lib/segment/src/vector_storage/chunked_vectors/read.rs
View File

@@ -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<T: Copy + 'static, S: UniversalRead<T>> {
pub struct ChunkedVectorsRead<T: bytemuck::Pod, S: UniversalRead> {
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<T: Copy + 'static, S: UniversalRead<T>> {
pub(super) directory: PathBuf,
}
impl<T: Copy + 'static, S: UniversalRead<T>> ChunkedVectorsRead<T, S> {
impl<T: bytemuck::Pod, S: UniversalRead> ChunkedVectorsRead<T, S> {
pub(super) fn config_file(directory: &Path) -> PathBuf {
directory.join(CONFIG_FILE_NAME)
}
@@ -222,7 +222,7 @@ impl<T: Copy + 'static, S: UniversalRead<T>> ChunkedVectorsRead<T, S> {
});
// access pattern does not matter for io_uring
UniversalRead::read_multi_iter::<Random, _>(reads)
TypedStorage::<S, T>::read_multi_iter::<Random, _>(reads)
.expect("iterator initialized")
.map(|result| result.expect("vector read"))
}

23
lib/segment/src/vector_storage/chunked_vectors/write.rs
View File

@@ -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<T>:
UniversalWrite<T> + UniversalWrite<Status> + Send + 'static
where
T: Copy + 'static,
{
}
impl<T, S> UioChunkedVectors<T> for S
where
T: Copy + 'static,
S: UniversalWrite<T> + UniversalWrite<Status> + Send + 'static,
{
}
#[derive(Debug)]
pub struct ChunkedVectors<T, S>
where
T: Copy + 'static,
S: UioChunkedVectors<T>,
T: bytemuck::Pod,
S: UniversalWrite + Send + 'static,
{
inner: ChunkedVectorsRead<T, S>,
status: StoredStruct<S, Status>,
}
impl<T: Copy + 'static, S: UioChunkedVectors<T>> Deref for ChunkedVectors<T, S> {
impl<T: bytemuck::Pod, S: UniversalWrite + Send + 'static> Deref for ChunkedVectors<T, S> {
type Target = ChunkedVectorsRead<T, S>;
fn deref(&self) -> &Self::Target {
@@ -50,8 +37,8 @@ impl<T: Copy + 'static, S: UioChunkedVectors<T>> Deref for ChunkedVectors<T, S>
impl<T, S> ChunkedVectors<T, S>
where
T: Copy + 'static,
S: UioChunkedVectors<T>,
T: bytemuck::Pod,
S: UniversalWrite + Send + 'static,
{
pub fn ensure_status_file(directory: &Path) -> OperationResult<PathBuf> {
let status_file = ChunkedVectorsRead::<T, S>::status_file(directory);

12
lib/segment/src/vector_storage/dense/dense_vector_storage.rs
View File

@@ -39,7 +39,7 @@ const DELETED_PATH: &str = "deleted.dat";
pub struct DenseVectorStorageImpl<T, S = MmapFile>
where
T: PrimitiveVectorElement,
S: UniversalRead<T>,
S: UniversalRead,
{
vectors_path: PathBuf,
deleted_path: PathBuf,
@@ -51,7 +51,7 @@ where
impl<T, S> DenseVectorStorageImpl<T, S>
where
T: PrimitiveVectorElement,
S: UniversalRead<T>,
S: UniversalRead,
{
/// Populate all pages in the mmap.
/// Block until all pages are populated.
@@ -170,7 +170,7 @@ fn open_dense_vector_storage_impl<T, S>(
) -> OperationResult<DenseVectorStorageImpl<T, S>>
where
T: PrimitiveVectorElement,
S: UniversalRead<T>,
S: UniversalRead,
{
fs::create_dir_all(path)?;
@@ -192,7 +192,7 @@ where
impl<T, S> DenseVectorStorage<T> for DenseVectorStorageImpl<T, S>
where
T: PrimitiveVectorElement,
S: UniversalRead<T>,
S: UniversalRead,
{
fn vector_dim(&self) -> usize {
self.vectors.as_ref().unwrap().dim
@@ -215,7 +215,7 @@ where
impl<T, S> VectorStorageRead for DenseVectorStorageImpl<T, S>
where
T: PrimitiveVectorElement,
S: UniversalRead<T>,
S: UniversalRead,
{
fn distance(&self) -> Distance {
self.distance
@@ -284,7 +284,7 @@ where
impl<T, S> VectorStorage for DenseVectorStorageImpl<T, S>
where
T: PrimitiveVectorElement,
S: UniversalRead<T>,
S: UniversalRead,
{
fn insert_vector(
&mut self,

6
lib/segment/src/vector_storage/dense/immutable_dense_vectors.rs
View File

@@ -29,11 +29,11 @@ const DELETED_HEADER: &[u8; HEADER_SIZE] = b"drop";
pub struct ImmutableDenseVectors<T, S = MmapFile>
where
T: PrimitiveVectorElement,
S: UniversalRead<T>,
S: UniversalRead,
{
pub dim: usize,
pub num_vectors: usize,
/// Vector data storage, providing read access via [`UniversalRead<T>`].
/// Vector data storage, providing read access via [`UniversalRead`].
storage: TypedStorage<ReadOnly<S>, T>,
/// Memory mapped deletion flags
deleted: MmapBitSlice,
@@ -41,7 +41,7 @@ where
pub deleted_count: usize,
}
impl<T: PrimitiveVectorElement, S: UniversalRead<T>> ImmutableDenseVectors<T, S> {
impl<T: PrimitiveVectorElement, S: UniversalRead> ImmutableDenseVectors<T, S> {
pub fn open(
vectors_path: &Path,
deleted_path: &Path,

4
lib/segment/src/vector_storage/dense/read_only/chucked_vector_storage.rs
View File

@@ -10,7 +10,7 @@ use crate::vector_storage::chunked_vectors::ChunkedVectorsRead;
use crate::vector_storage::{VectorOffsetType, VectorStorageRead};
#[derive(Debug)]
pub struct ReadOnlyChunkedDenseVectorStorage<T: PrimitiveVectorElement, S: UniversalRead<T>> {
pub struct ReadOnlyChunkedDenseVectorStorage<T: PrimitiveVectorElement, S: UniversalRead> {
vectors: ChunkedVectorsRead<T, S>,
/// Flags marking deleted vectors
///
@@ -21,7 +21,7 @@ pub struct ReadOnlyChunkedDenseVectorStorage<T: PrimitiveVectorElement, S: Unive
deleted_count: usize,
}
impl<T: PrimitiveVectorElement, S: UniversalRead<T>> VectorStorageRead
impl<T: PrimitiveVectorElement, S: UniversalRead> VectorStorageRead
for ReadOnlyChunkedDenseVectorStorage<T, S>
{
fn distance(&self) -> Distance {

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

@@ -68,8 +68,8 @@ pub(crate) fn read_multi_vector<'a, T, P, So, Sv>(
where
T: PrimitiveVectorElement,
P: AccessPattern,
So: UniversalRead<MultivectorMmapOffset>,
Sv: UniversalRead<T>,
So: UniversalRead,
Sv: UniversalRead,
{
let mmap_offset = *offsets
.get::<P>(key as VectorOffsetType)?

11
lib/segment/src/vector_storage/multi_dense/read_only/chunked_vector_storage.rs
View File

@@ -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<T: PrimitiveVectorElement, S: UniversalReadFamily>
{
vectors: ChunkedVectorsRead<T, S::Read<T>>,
offsets: ChunkedVectorsRead<MultivectorMmapOffset, S::Read<MultivectorMmapOffset>>,
pub struct ReadOnlyChunkedMultiDenseVectorStorage<T: PrimitiveVectorElement, S: UniversalRead> {
vectors: ChunkedVectorsRead<T, S>,
offsets: ChunkedVectorsRead<MultivectorMmapOffset, S>,
/// 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<T: PrimitiveVectorElement, S:
deleted_count: usize,
}
impl<T: PrimitiveVectorElement, S: UniversalReadFamily> VectorStorageRead
impl<T: PrimitiveVectorElement, S: UniversalRead> VectorStorageRead
for ReadOnlyChunkedMultiDenseVectorStorage<T, S>
{
fn distance(&self) -> Distance {

34
lib/segment/src/vector_storage/read_only/mod.rs
View File

@@ -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<S: UniversalReadFamily> {
Dense(Box<DenseVectorStorageImpl<VectorElementType, S::Read<VectorElementType>>>),
DenseByte(Box<DenseVectorStorageImpl<VectorElementTypeByte, S::Read<VectorElementTypeByte>>>),
DenseHalf(Box<DenseVectorStorageImpl<VectorElementTypeHalf, S::Read<VectorElementTypeHalf>>>),
DenseChunked(
Box<ReadOnlyChunkedDenseVectorStorage<VectorElementType, S::Read<VectorElementType>>>,
),
DenseChunkedByte(
Box<
ReadOnlyChunkedDenseVectorStorage<
VectorElementTypeByte,
S::Read<VectorElementTypeByte>,
>,
>,
),
DenseChunkedHalf(
Box<
ReadOnlyChunkedDenseVectorStorage<
VectorElementTypeHalf,
S::Read<VectorElementTypeHalf>,
>,
>,
),
pub enum VectorStorageReadEnum<S: UniversalRead> {
Dense(Box<DenseVectorStorageImpl<VectorElementType, S>>),
DenseByte(Box<DenseVectorStorageImpl<VectorElementTypeByte, S>>),
DenseHalf(Box<DenseVectorStorageImpl<VectorElementTypeHalf, S>>),
DenseChunked(Box<ReadOnlyChunkedDenseVectorStorage<VectorElementType, S>>),
DenseChunkedByte(Box<ReadOnlyChunkedDenseVectorStorage<VectorElementTypeByte, S>>),
DenseChunkedHalf(Box<ReadOnlyChunkedDenseVectorStorage<VectorElementTypeHalf, S>>),
MultiDenseChunked(Box<ReadOnlyChunkedMultiDenseVectorStorage<VectorElementType, S>>),
MultiDenseChunkedByte(Box<ReadOnlyChunkedMultiDenseVectorStorage<VectorElementTypeByte, S>>),
MultiDenseChunkedHalf(Box<ReadOnlyChunkedMultiDenseVectorStorage<VectorElementTypeHalf, S>>),
Sparse(Box<ReadOnlySparseVectorStorage>),
}
impl<S: UniversalReadFamily> VectorStorageRead for VectorStorageReadEnum<S> {
impl<S: UniversalRead> VectorStorageRead for VectorStorageReadEnum<S> {
fn distance(&self) -> Distance {
match self {
VectorStorageReadEnum::Dense(s) => s.distance(),

2
src/tonic/api/storage_read_api/helpers.rs
View File

@@ -12,7 +12,7 @@ use tonic::Status;
use crate::tonic::api::storage_read_api::StorageReadService;
impl<S: UniversalRead<u8> + Send + Sync + 'static> StorageReadService<S> {
impl<S: UniversalRead + Send + Sync + 'static> StorageReadService<S> {
pub fn new(dispatcher: Arc<Dispatcher>) -> Self {
Self {
dispatcher,

18
src/tonic/api/storage_read_api/mod.rs
View File

@@ -29,13 +29,13 @@ mod tests;
/// Chunk size for streaming reads (~1 MB).
const STREAM_CHUNK_SIZE: u64 = 1024 * 1024;
pub struct StorageReadService<S: UniversalRead<u8> + Send + Sync + 'static = MmapFile> {
pub struct StorageReadService<S: UniversalRead + Send + Sync + 'static = MmapFile> {
dispatcher: Arc<Dispatcher>,
_marker: PhantomData<S>,
}
#[async_trait]
impl<S: UniversalRead<u8> + Send + Sync + 'static> StorageRead for StorageReadService<S> {
impl<S: UniversalRead + Send + Sync + 'static> StorageRead for StorageReadService<S> {
// Check if a file exists via UniversalRead::open(), catch NotFound → false.
async fn file_exists(
&self,
@@ -127,7 +127,7 @@ impl<S: UniversalRead<u8> + 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::<u8>().map_err(io_error_to_status)
})
.await
.map_err(|e| Status::internal(format!("Task join error: {e}")))??;
@@ -159,7 +159,7 @@ impl<S: UniversalRead<u8> + 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::<Random>(ReadRange::new(byte_offset, length))
.read::<u8, Random>(ReadRange::new(byte_offset, length))
.map_err(io_error_to_status)?;
Ok::<_, Status>(cow.into_owned())
})
@@ -196,7 +196,7 @@ impl<S: UniversalRead<u8> + 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::<u8>().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<S: UniversalRead<u8> + Send + Sync + 'static> StorageRead for StorageReadSe
let data = tokio::task::spawn_blocking(move || {
storage_for_read
.read::<Random>(ReadRange::new(current_offset, chunk_size))
.read::<u8, Random>(ReadRange::new(current_offset, chunk_size))
.map(|cow| cow.into_owned())
.map_err(io_error_to_status)
})
@@ -254,7 +254,7 @@ impl<S: UniversalRead<u8> + 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::<u8>().map_err(io_error_to_status)?;
Ok::<_, Status>(cow.into_owned())
})
.await
@@ -291,7 +291,7 @@ impl<S: UniversalRead<u8> + 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::<Vec<_>>();
storage
.read_batch::<Random, _>(ranges.into_iter().enumerate(), |idx, chunk| {
.read_batch::<u8, Random, _>(ranges.into_iter().enumerate(), |idx, chunk| {
results[idx].extend_from_slice(chunk);
Ok(())
})
@@ -352,7 +352,7 @@ impl<S: UniversalRead<u8> + Send + Sync + 'static> StorageRead for StorageReadSe
.enumerate()
.map(|(op_idx, (file_idx, range))| (op_idx, &files[file_idx], range));
S::read_multi::<Random, _>(reads, |op_idx, chunk| {
S::read_multi::<u8, Random, _>(reads, |op_idx, chunk| {
results[op_idx].extend_from_slice(chunk);
Ok(())
})