From af4e9e5d779fb5bf0c48dc334cffb1ef30700ee4 Mon Sep 17 00:00:00 2001 From: Ivan Pleshkov Date: Thu, 30 Jul 2026 11:39:37 +0200 Subject: [PATCH] blobstore with raw bytes api (#10024) --- lib/blobstore/src/blobstore/gridstore/mod.rs | 31 +++++- .../src/blobstore/gridstore/reader.rs | 13 +++ lib/blobstore/src/blobstore/gridstore/view.rs | 17 ++- lib/blobstore/src/blobstore/logstore/mod.rs | 35 +++++- .../src/blobstore/logstore/reader.rs | 13 +++ lib/blobstore/src/blobstore/logstore/view.rs | 15 ++- lib/blobstore/src/blobstore/mod.rs | 42 ++++++++ lib/blobstore/src/blobstore/reader.rs | 15 +++ lib/blobstore/src/blobstore/tests.rs | 102 ++++++++++++++++++ 9 files changed, 273 insertions(+), 10 deletions(-) diff --git a/lib/blobstore/src/blobstore/gridstore/mod.rs b/lib/blobstore/src/blobstore/gridstore/mod.rs index 59e2cc9752..8496456ba9 100644 --- a/lib/blobstore/src/blobstore/gridstore/mod.rs +++ b/lib/blobstore/src/blobstore/gridstore/mod.rs @@ -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 { + 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, + hw_counter: HwMetricRefCounter, ) -> Result { // 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::

(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( + &self, + point_offset: PointOffset, + hw_counter: &HardwareCounterCell, + ) -> Result>> { + self.with_view(|view| { + let bytes = view.get_value_bytes::

(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. diff --git a/lib/blobstore/src/blobstore/gridstore/reader.rs b/lib/blobstore/src/blobstore/gridstore/reader.rs index 39c41ebcdf..13b5d4e2c0 100644 --- a/lib/blobstore/src/blobstore/gridstore/reader.rs +++ b/lib/blobstore/src/blobstore/gridstore/reader.rs @@ -115,6 +115,19 @@ impl GridstoreReader { self.view().get_value::

(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( + &self, + point_offset: PointOffset, + hw_counter: &HardwareCounterCell, + ) -> Result>> { + let view = self.view(); + let bytes = view.get_value_bytes::

(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. /// diff --git a/lib/blobstore/src/blobstore/gridstore/view.rs b/lib/blobstore/src/blobstore/gridstore/view.rs index 644fd9ce4d..77cb13b21d 100644 --- a/lib/blobstore/src/blobstore/gridstore/view.rs +++ b/lib/blobstore/src/blobstore/gridstore/view.rs @@ -77,6 +77,18 @@ impl<'a, V: Blob, S: UniversalRead, T: TrackerRead> GridstoreView<'a, V, S, T point_offset: PointOffset, hw_counter: &HardwareCounterCell, ) -> Result> { + let bytes = self.get_value_bytes::

(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( + &self, + point_offset: PointOffset, + hw_counter: &HardwareCounterCell, + ) -> Result>> { let Some(pointer) = self.get_pointer(point_offset)? else { return Ok(None); }; @@ -84,10 +96,7 @@ impl<'a, V: Blob, S: UniversalRead, T: TrackerRead> GridstoreView<'a, V, S, T let raw = self.read_from_pages::

(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( diff --git a/lib/blobstore/src/blobstore/logstore/mod.rs b/lib/blobstore/src/blobstore/logstore/mod.rs index f5970b583a..3c44cd32d0 100644 --- a/lib/blobstore/src/blobstore/logstore/mod.rs +++ b/lib/blobstore/src/blobstore/logstore/mod.rs @@ -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 { + 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, + hw_counter: HwMetricRefCounter, ) -> Result { // 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::

(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( + &self, + point_offset: PointOffset, + hw_counter: &HardwareCounterCell, + ) -> Result>> { + self.with_view(|view| { + let bytes = view.get_value_bytes::

(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( &self, diff --git a/lib/blobstore/src/blobstore/logstore/reader.rs b/lib/blobstore/src/blobstore/logstore/reader.rs index 8365b9ce7a..27765ec567 100644 --- a/lib/blobstore/src/blobstore/logstore/reader.rs +++ b/lib/blobstore/src/blobstore/logstore/reader.rs @@ -96,6 +96,19 @@ impl LogstoreReader { self.view().get_value::

(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( + &self, + point_offset: PointOffset, + hw_counter: &HardwareCounterCell, + ) -> Result>> { + let view = self.view(); + let bytes = view.get_value_bytes::

(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. /// diff --git a/lib/blobstore/src/blobstore/logstore/view.rs b/lib/blobstore/src/blobstore/logstore/view.rs index 7a4a5702de..f8709db364 100644 --- a/lib/blobstore/src/blobstore/logstore/view.rs +++ b/lib/blobstore/src/blobstore/logstore/view.rs @@ -66,6 +66,18 @@ impl<'a, V: Blob, S: UniversalRead> LogstoreView<'a, V, S> { point_offset: PointOffset, hw_counter: &HardwareCounterCell, ) -> Result> { + let bytes = self.get_value_bytes::

(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( + &self, + point_offset: PointOffset, + hw_counter: &HardwareCounterCell, + ) -> Result>> { let Some(pointer) = self.tracker.get::

(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::

(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. diff --git a/lib/blobstore/src/blobstore/mod.rs b/lib/blobstore/src/blobstore/mod.rs index 8f9eaa2e03..0909acdcb7 100644 --- a/lib/blobstore/src/blobstore/mod.rs +++ b/lib/blobstore/src/blobstore/mod.rs @@ -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, + hw_counter: HwMetricRefCounter, + ) -> Result { + 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( + &self, + point_offset: PointOffset, + hw_counter: &HardwareCounterCell, + ) -> Result>> { + match self { + Blobstore::Gridstore(storage) => storage.get_value_bytes::

(point_offset, hw_counter), + Blobstore::Logstore(storage) => storage.get_value_bytes::

(point_offset, hw_counter), + } + } + /// Iterate over all given values and execute callback for each one. /// /// Return `false` from the callback to stop iteration early. diff --git a/lib/blobstore/src/blobstore/reader.rs b/lib/blobstore/src/blobstore/reader.rs index 9f0ff8af15..4931d33bf7 100644 --- a/lib/blobstore/src/blobstore/reader.rs +++ b/lib/blobstore/src/blobstore/reader.rs @@ -128,6 +128,21 @@ impl BlobstoreReader { } } + /// 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( + &self, + point_offset: PointOffset, + hw_counter: &HardwareCounterCell, + ) -> Result>> { + match self { + Self::Gridstore(reader) => reader.get_value_bytes::

(point_offset, hw_counter), + Self::Logstore(reader) => reader.get_value_bytes::

(point_offset, hw_counter), + } + } + /// Iterate over all values with point offsets below `max_id` and execute callback for each one. /// Missing values are skipped. /// diff --git a/lib/blobstore/src/blobstore/tests.rs b/lib/blobstore/src/blobstore/tests.rs index aa75e72b33..1cb7996288 100644 --- a/lib/blobstore/src/blobstore/tests.rs +++ b/lib/blobstore/src/blobstore/tests.rs @@ -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) { + 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::(); + 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::(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::(1, &hw_counter).unwrap(), + Some(payload_1.to_bytes()), + ); + assert_eq!( + storage.get_value_bytes::(0, &hw_counter).unwrap(), + Some(payload_0.to_bytes()), + ); + + // Missing point offset reads as None + assert_eq!( + storage.get_value_bytes::(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::(); + let payloads = (0..10).map(|_| random_payload(rng, 2)).collect::>(); + + 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::(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::(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);