upstream-archive/qdrant
1
0
Fork 0
mirror of https://github.com/qdrant/qdrant.git synced 2026-08-06 01:50:57 -05:00
Code Issues Packages Projects Releases Wiki Activity

blobstore with raw bytes api (#10024)

Browse Source
This commit is contained in:
Ivan Pleshkov
2026-07-30 11:39:37 +02:00
committed by GitHub
parent ec1aec5f4c
commit af4e9e5d77
9 changed files with 273 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
lib/blobstore/src/blobstore/gridstore/mod.rs
View File

@@ -3,6 +3,7 @@ pub(super) mod pages;
mod reader;
mod view;
use std::borrow::Cow;
use std::cmp;
use std::path::PathBuf;
use std::sync::Arc;
@@ -236,6 +237,21 @@ where
point_offset: PointOffset,
value: &V,
hw_counter: HwMetricRefCounter,
) -> Result<bool> {
self.put_value_bytes(point_offset, value.to_bytes(), hw_counter)
}
/// Put an already serialized value in the storage.
///
/// `value_bytes` must be the value in its [`Blob`] encoding, uncompressed. Compression is
/// applied here with the local storage config.
///
/// Returns true if the value existed previously and was updated, false if it was newly inserted.
pub(super) fn put_value_bytes(
&mut self,
point_offset: PointOffset,
value_bytes: Vec<u8>,
hw_counter: HwMetricRefCounter,
) -> Result<bool> {
// This function needs to NOT corrupt data in case of a crash.
//
@@ -285,7 +301,6 @@ where
// that should happen is that we mark more cells as used than they actually are,
// so will never reuse such space, but data will not be corrupted.
let value_bytes = value.to_bytes();
let comp_value = self.with_view(|view| view.compress(value_bytes));
let value_size = comp_value.len();
@@ -380,6 +395,20 @@ where
self.with_view(|view| view.get_value::<P>(point_offset, hw_counter))
}
/// Get the serialized value for a given point offset.
///
/// The returned bytes are the value in its [`Blob`] encoding, always decompressed.
pub(super) fn get_value_bytes<P: AccessPattern>(
&self,
point_offset: PointOffset,
hw_counter: &HardwareCounterCell,
) -> Result<Option<Vec<u8>>> {
self.with_view(|view| {
let bytes = view.get_value_bytes::<P>(point_offset, hw_counter)?;
Ok(bytes.map(Cow::into_owned))
})
}
/// Iterate over all given values and execute callback for each one.
///
/// Return `false` from the callback to stop iteration early.

13
lib/blobstore/src/blobstore/gridstore/reader.rs
View File

@@ -115,6 +115,19 @@ impl<V: Blob, S: UniversalRead> GridstoreReader<V, S> {
self.view().get_value::<P>(point_offset, hw_counter)
}
/// Get the serialized value for a given point offset.
///
/// The returned bytes are the value in its [`Blob`] encoding, always decompressed.
pub(crate) fn get_value_bytes<P: AccessPattern>(
&self,
point_offset: PointOffset,
hw_counter: &HardwareCounterCell,
) -> Result<Option<Vec<u8>>> {
let view = self.view();
let bytes = view.get_value_bytes::<P>(point_offset, hw_counter)?;
Ok(bytes.map(std::borrow::Cow::into_owned))
}
/// Iterate over all values with point offsets below `max_id` and execute callback for each one.
/// Missing values are skipped.
///

17
lib/blobstore/src/blobstore/gridstore/view.rs
View File

@@ -77,6 +77,18 @@ impl<'a, V: Blob, S: UniversalRead, T: TrackerRead<S>> GridstoreView<'a, V, S, T
point_offset: PointOffset,
hw_counter: &HardwareCounterCell,
) -> Result<Option<V>> {
let bytes = self.get_value_bytes::<P>(point_offset, hw_counter)?;
Ok(bytes.map(|bytes| V::from_bytes(&bytes)))
}
/// Get the serialized value for a given point offset.
///
/// The returned bytes are the value in its [`Blob`] encoding, always decompressed.
pub fn get_value_bytes<P: AccessPattern>(
&self,
point_offset: PointOffset,
hw_counter: &HardwareCounterCell,
) -> Result<Option<Cow<'_, [u8]>>> {
let Some(pointer) = self.get_pointer(point_offset)? else {
return Ok(None);
};
@@ -84,10 +96,7 @@ impl<'a, V: Blob, S: UniversalRead, T: TrackerRead<S>> GridstoreView<'a, V, S, T
let raw = self.read_from_pages::<P>(pointer)?;
hw_counter.payload_io_read_counter().incr_delta(raw.len());
let decompressed = self.decompress(raw);
let value = V::from_bytes(&decompressed);
Ok(Some(value))
Ok(Some(self.decompress(raw)))
}
pub fn read_values<P, U, E>(

35
lib/blobstore/src/blobstore/logstore/mod.rs
View File

@@ -4,6 +4,7 @@ mod reader;
mod tests;
mod view;
use std::borrow::Cow;
use std::marker::PhantomData;
use std::path::PathBuf;
use std::sync::Arc;
@@ -181,13 +182,28 @@ where
/// rejected, the storage is append-only.
///
/// Always returns false on success, as values can never be updated.
// Takes &mut self for signature parity with the mutable variant
#[allow(clippy::needless_pass_by_ref_mut)]
pub(super) fn put_value(
&mut self,
point_offset: PointOffset,
value: &V,
hw_counter: HwMetricRefCounter,
) -> Result<bool> {
self.put_value_bytes(point_offset, value.to_bytes(), hw_counter)
}
/// Put an already serialized value in the storage.
///
/// `value_bytes` must be the value in its [`Blob`] encoding, uncompressed. Compression is
/// applied here with the local storage config.
///
/// See [`put_value`](Self::put_value) for buffering and append-only semantics.
// Takes &mut self for signature parity with the mutable variant
#[allow(clippy::needless_pass_by_ref_mut)]
pub(super) fn put_value_bytes(
&mut self,
point_offset: PointOffset,
value_bytes: Vec<u8>,
hw_counter: HwMetricRefCounter,
) -> Result<bool> {
// Validate before buffering anything, a rejected put must not leave data behind
let next = self.tracker.read().pointer_count();
@@ -199,7 +215,6 @@ where
)));
}
let value_bytes = value.to_bytes();
let comp_value = self.config.compression.compress(value_bytes);
let value_size = comp_value.len();
@@ -281,6 +296,20 @@ where
self.with_view(|view| view.get_value::<P>(point_offset, hw_counter))
}
/// Get the serialized value for a given point offset.
///
/// The returned bytes are the value in its [`Blob`] encoding, always decompressed.
pub(super) fn get_value_bytes<P: AccessPattern>(
&self,
point_offset: PointOffset,
hw_counter: &HardwareCounterCell,
) -> Result<Option<Vec<u8>>> {
self.with_view(|view| {
let bytes = view.get_value_bytes::<P>(point_offset, hw_counter)?;
Ok(bytes.map(Cow::into_owned))
})
}
/// Iterate over all given values and execute callback for each one.
pub(super) fn read_values<P, U, E>(
&self,

13
lib/blobstore/src/blobstore/logstore/reader.rs
View File

@@ -96,6 +96,19 @@ impl<V: Blob, S: UniversalRead> LogstoreReader<V, S> {
self.view().get_value::<P>(point_offset, hw_counter)
}
/// Get the serialized value for a given point offset.
///
/// The returned bytes are the value in its [`Blob`] encoding, always decompressed.
pub(crate) fn get_value_bytes<P: AccessPattern>(
&self,
point_offset: PointOffset,
hw_counter: &HardwareCounterCell,
) -> Result<Option<Vec<u8>>> {
let view = self.view();
let bytes = view.get_value_bytes::<P>(point_offset, hw_counter)?;
Ok(bytes.map(std::borrow::Cow::into_owned))
}
/// Iterate over all values with point offsets below `max_id` and execute callback for each
/// one. Missing values are skipped.
///

15
lib/blobstore/src/blobstore/logstore/view.rs
View File

@@ -66,6 +66,18 @@ impl<'a, V: Blob, S: UniversalRead> LogstoreView<'a, V, S> {
point_offset: PointOffset,
hw_counter: &HardwareCounterCell,
) -> Result<Option<V>> {
let bytes = self.get_value_bytes::<P>(point_offset, hw_counter)?;
Ok(bytes.map(|bytes| V::from_bytes(&bytes)))
}
/// Get the serialized value for a given point offset.
///
/// The returned bytes are the value in its [`Blob`] encoding, always decompressed.
pub(crate) fn get_value_bytes<P: AccessPattern>(
&self,
point_offset: PointOffset,
hw_counter: &HardwareCounterCell,
) -> Result<Option<Cow<'_, [u8]>>> {
let Some(pointer) = self.tracker.get::<P>(point_offset)? else {
return Ok(None);
};
@@ -73,8 +85,7 @@ impl<'a, V: Blob, S: UniversalRead> LogstoreView<'a, V, S> {
let raw = self.read_from_pages::<P>(pointer)?;
hw_counter.payload_io_read_counter().incr_delta(raw.len());
let decompressed = self.config.compression.decompress(raw);
Ok(Some(V::from_bytes(&decompressed)))
Ok(Some(self.config.compression.decompress(raw)))
}
/// Iterate over all given values and execute callback for each one.

42
lib/blobstore/src/blobstore/mod.rs
View File

@@ -145,6 +145,32 @@ where
}
}
/// Put an already serialized value in the storage.
///
/// `value_bytes` must be the value in its [`Blob`] encoding, uncompressed — the same bytes
/// `V::to_bytes` would produce. Compression is applied here with the local storage config,
/// so callers can exchange these bytes between storages with different compression settings.
///
/// Returns true if the value existed previously and was updated, false if it was newly inserted.
///
/// In append-only mode values must be put at monotonically increasing point offsets, and
/// cannot be overwritten.
pub fn put_value_bytes(
&mut self,
point_offset: PointOffset,
value_bytes: Vec<u8>,
hw_counter: HwMetricRefCounter,
) -> Result<bool> {
match self {
Blobstore::Gridstore(storage) => {
storage.put_value_bytes(point_offset, value_bytes, hw_counter)
}
Blobstore::Logstore(storage) => {
storage.put_value_bytes(point_offset, value_bytes, hw_counter)
}
}
}
/// Delete a value from the storage.
///
/// Returns None if the point_offset, page, or value was not found.
@@ -198,6 +224,22 @@ where
}
}
/// Get the serialized value for a given point offset.
///
/// The returned bytes are the value in its [`Blob`] encoding, always decompressed — the same
/// bytes `V::to_bytes` would produce, valid as [`put_value_bytes`](Self::put_value_bytes)
/// input regardless of either storage's compression setting.
pub fn get_value_bytes<P: AccessPattern>(
&self,
point_offset: PointOffset,
hw_counter: &HardwareCounterCell,
) -> Result<Option<Vec<u8>>> {
match self {
Blobstore::Gridstore(storage) => storage.get_value_bytes::<P>(point_offset, hw_counter),
Blobstore::Logstore(storage) => storage.get_value_bytes::<P>(point_offset, hw_counter),
}
}
/// Iterate over all given values and execute callback for each one.
///
/// Return `false` from the callback to stop iteration early.

15
lib/blobstore/src/blobstore/reader.rs
View File

@@ -128,6 +128,21 @@ impl<V: Blob, S: UniversalRead> BlobstoreReader<V, S> {
}
}
/// Get the serialized value for a given point offset.
///
/// The returned bytes are the value in its [`Blob`] encoding, always decompressed — the same
/// bytes `V::to_bytes` would produce.
pub fn get_value_bytes<P: AccessPattern>(
&self,
point_offset: PointOffset,
hw_counter: &HardwareCounterCell,
) -> Result<Option<Vec<u8>>> {
match self {
Self::Gridstore(reader) => reader.get_value_bytes::<P>(point_offset, hw_counter),
Self::Logstore(reader) => reader.get_value_bytes::<P>(point_offset, hw_counter),
}
}
/// Iterate over all values with point offsets below `max_id` and execute callback for each one.
/// Missing values are skipped.
///

102
lib/blobstore/src/blobstore/tests.rs
View File

@@ -110,6 +110,108 @@ fn test_put_single_payload(#[values(Mode::Mutable, Mode::AppendOnly)] mode: Mode
}
}
fn empty_storage_compression(
mode: Mode,
compression: Compression,
) -> (tempfile::TempDir, Blobstore<Payload>) {
let dir = Builder::new().prefix("test-storage").tempdir().unwrap();
let config = match mode {
Mode::Mutable => StorageConfig::Mutable(GridstoreConfig {
compression,
..GridstoreConfig::DEFAULT
}),
Mode::AppendOnly => StorageConfig::AppendOnly(LogstoreConfig {
compression,
..LogstoreConfig::DEFAULT
}),
};
let storage = Blobstore::new(MmapFs, dir.path().to_path_buf(), config).unwrap();
(dir, storage)
}
#[rstest]
fn test_put_get_value_bytes(
#[values(Mode::Mutable, Mode::AppendOnly)] mode: Mode,
#[values(Compression::None, Compression::LZ4)] compression: Compression,
) {
let (_dir, mut storage) = empty_storage_compression(mode, compression);
let hw_counter = HardwareCounterCell::new();
let hw_counter_ref = hw_counter.ref_payload_io_write_counter();
let rng = &mut rand::make_rng::<rand::rngs::SmallRng>();
let payload_0 = random_payload(rng, 2);
let payload_1 = random_payload(rng, 2);
// Serialized bytes put with `put_value_bytes` read back as the parsed value
storage
.put_value_bytes(0, payload_0.to_bytes(), hw_counter_ref)
.unwrap();
assert_eq!(
storage.get_value::<Random>(0, &hw_counter).unwrap(),
Some(payload_0.clone()),
);
// Value put with `put_value` reads back with `get_value_bytes` as its `Blob` encoding
storage.put_value(1, &payload_1, hw_counter_ref).unwrap();
assert_eq!(
storage.get_value_bytes::<Random>(1, &hw_counter).unwrap(),
Some(payload_1.to_bytes()),
);
assert_eq!(
storage.get_value_bytes::<Random>(0, &hw_counter).unwrap(),
Some(payload_0.to_bytes()),
);
// Missing point offset reads as None
assert_eq!(
storage.get_value_bytes::<Random>(2, &hw_counter).unwrap(),
None,
);
}
#[rstest]
fn test_value_bytes_cross_compression(
#[values(Mode::Mutable, Mode::AppendOnly)] source_mode: Mode,
#[values(Mode::Mutable, Mode::AppendOnly)] target_mode: Mode,
) {
let (_source_dir, mut source) = empty_storage_compression(source_mode, Compression::LZ4);
let (_target_dir, mut target) = empty_storage_compression(target_mode, Compression::None);
let hw_counter = HardwareCounterCell::new();
let hw_counter_ref = hw_counter.ref_payload_io_write_counter();
let rng = &mut rand::make_rng::<rand::rngs::SmallRng>();
let payloads = (0..10).map(|_| random_payload(rng, 2)).collect::<Vec<_>>();
for (point_offset, payload) in payloads.iter().enumerate() {
source
.put_value(point_offset as PointOffset, payload, hw_counter_ref)
.unwrap();
}
// Transfer stored bytes as-is, without parsing
for point_offset in 0..payloads.len() as PointOffset {
let bytes = source
.get_value_bytes::<Random>(point_offset, &hw_counter)
.unwrap()
.unwrap();
target
.put_value_bytes(point_offset, bytes, hw_counter_ref)
.unwrap();
}
for (point_offset, payload) in payloads.iter().enumerate() {
assert_eq!(
target
.get_value::<Random>(point_offset as PointOffset, &hw_counter)
.unwrap()
.as_ref(),
Some(payload),
);
}
}
#[rstest]
fn test_storage_files(#[values(Mode::Mutable, Mode::AppendOnly)] mode: Mode) {
let (dir, mut storage) = empty_storage_mode(mode);