perf: batch multi_get + PinnableSlice + WriteBatch

Cargo.toml

@@ -35,7 +35,7 @@ interval-heap = "0.0.5"
 log = "0.4.27"
 lz4_flex = { version = "0.13.0", optional = true, default-features = false }
 zstd = { version = "0.13", optional = true, default-features = false }
-structured-zstd = { version = "0.0.1", optional = true, default-features = false, features = ["std"] }
+structured-zstd = { version = "0.0.6", optional = true, default-features = false, features = ["std"] }
 quick_cache = { version = "0.6.16", default-features = false, features = [] }
 rustc-hash = "2.1.1"
 self_cell = "1.2.0"

README.md

@@ -12,7 +12,7 @@
 [![License](https://img.shields.io/badge/license-Apache--2.0-blue)](#license)
 
 > LSM-tree engine for [CoordiNode](https://github.com/structured-world/coordinode), maintained by [Structured World Foundation](https://sw.foundation).
-> Derivative work of [fjall-rs/lsm-tree](https://github.com/fjall-rs/lsm-tree), developed independently with diverging features: zstd dictionary compression, custom sequence number generators, multi_get, intra-L0 compaction, and security hardening.
+> Derivative work of [fjall-rs/lsm-tree](https://github.com/fjall-rs/lsm-tree), developed independently with diverging features: zstd dictionary compression, custom sequence number generators, multi_get (batch-optimized), PinnableSlice zero-copy reads, WriteBatch atomic writes, intra-L0 compaction, and security hardening.
 
 > [!IMPORTANT]
 > This fork now introduces a fork-specific **disk format V4** compatibility boundary.

src/abstract_tree.rs

@@ -554,6 +554,43 @@ pub trait AbstractTree: sealed::Sealed {
     /// Will return `Err` if an IO error occurs.
     fn get<K: AsRef<[u8]>>(&self, key: K, seqno: SeqNo) -> crate::Result<Option<UserValue>>;
 
+    /// Retrieves an item from the tree as a [`PinnableSlice`].
+    ///
+    /// When the value is backed by an on-disk data block, the returned
+    /// [`PinnableSlice::Pinned`] variant holds a reference to that block's
+    /// decompressed buffer, avoiding a data copy. Memtable and blob-resolved
+    /// values use the [`PinnableSlice::Owned`] variant.
+    ///
+    /// The existing [`AbstractTree::get`] method is unaffected.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # let folder = tempfile::tempdir()?;
+    /// use lsm_tree::{AbstractTree, Config, Tree};
+    ///
+    /// let tree = Config::new(&folder, Default::default(), Default::default()).open()?;
+    /// tree.insert("a", "my_value", 0);
+    ///
+    /// let item = tree.get_pinned("a", 1)?;
+    /// assert_eq!(item.as_ref().map(|v| v.as_ref()), Some("my_value".as_bytes()));
+    /// #
+    /// # Ok::<(), lsm_tree::Error>(())
+    /// ```
+    ///
+    /// # Errors
+    ///
+    /// Will return `Err` if an IO error occurs.
+    fn get_pinned<K: AsRef<[u8]>>(
+        &self,
+        key: K,
+        seqno: SeqNo,
+    ) -> crate::Result<Option<crate::PinnableSlice>> {
+        // Default: delegate to get() and wrap as Owned
+        self.get(key, seqno)
+            .map(|opt| opt.map(crate::PinnableSlice::owned))
+    }
+
     /// Returns `true` if the tree contains the specified key.
     ///
     /// # Examples
@@ -656,6 +693,39 @@ pub trait AbstractTree: sealed::Sealed {
         keys.into_iter().map(|key| self.get(key, seqno)).collect()
     }
 
+    /// Applies a [`WriteBatch`] with the given sequence number.
+    ///
+    /// All entries share a single seqno. This is more efficient than individual
+    /// writes because the version-history lock and memtable size accounting
+    /// are performed only once for the entire batch.
+    ///
+    /// **Visibility:** entries become individually visible to concurrent readers
+    /// as they are inserted. For atomic batch visibility, the caller must
+    /// publish `seqno` (via `visible_seqno.fetch_max(seqno + 1)`) only
+    /// **after** this method returns.
+    ///
+    /// Returns the total bytes added and new size of the memtable.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # let folder = tempfile::tempdir()?;
+    /// use lsm_tree::{AbstractTree, Config, Tree, WriteBatch};
+    ///
+    /// let tree = Config::new(&folder, Default::default(), Default::default()).open()?;
+    ///
+    /// let mut batch = WriteBatch::new();
+    /// batch.insert("key1", "value1");
+    /// batch.insert("key2", "value2");
+    /// batch.remove("key3");
+    ///
+    /// let (bytes_added, memtable_size) = tree.apply_batch(batch, 0);
+    /// assert!(bytes_added > 0);
+    /// #
+    /// # Ok::<(), lsm_tree::Error>(())
+    /// ```
+    fn apply_batch(&self, batch: crate::WriteBatch, seqno: SeqNo) -> (u64, u64);
+
     /// Inserts a key-value pair into the tree.
     ///
     /// If the key already exists, the item will be overwritten.

src/blob_tree/mod.rs

@@ -667,6 +667,10 @@ impl AbstractTree for BlobTree {
         self.index.get_highest_persisted_seqno()
     }
 
+    fn apply_batch(&self, batch: crate::WriteBatch, seqno: SeqNo) -> (u64, u64) {
+        self.index.apply_batch(batch, seqno)
+    }
+
     fn insert<K: Into<UserKey>, V: Into<UserValue>>(
         &self,
         key: K,

src/lib.rs

@@ -133,6 +133,7 @@ pub(crate) mod metrics;
 pub mod mvcc_stream;
 
 mod path;
+mod pinnable_slice;
 mod prefix;
 
 #[doc(hidden)]
@@ -161,6 +162,7 @@ pub mod util;
 
 mod value;
 mod value_type;
+mod write_batch;
 
 /// Integrity verification for SST and blob files.
 pub mod verify;
@@ -196,6 +198,9 @@ pub use encryption::EncryptionProvider;
 #[cfg(feature = "encryption")]
 pub use encryption::Aes256GcmProvider;
 
+pub use pinnable_slice::PinnableSlice;
+pub use write_batch::WriteBatch;
+
 pub use {
     abstract_tree::AbstractTree,
     any_tree::AnyTree,

src/memtable/mod.rs

@@ -185,6 +185,59 @@ impl Memtable {
         self.items.is_empty() && self.range_tombstone_count() == 0
     }
 
+    /// Inserts multiple items into the memtable in bulk.
+    ///
+    /// More efficient than calling [`Memtable::insert`] in a loop because it
+    /// performs a single `fetch_add` for the total size and a single
+    /// `fetch_max` for the highest seqno.
+    ///
+    /// Returns `(total_bytes_added, new_memtable_size)`.
+    #[doc(hidden)]
+    pub fn insert_batch(&self, items: Vec<InternalValue>) -> (u64, u64) {
+        if items.is_empty() {
+            let size = self
+                .approximate_size
+                .load(std::sync::atomic::Ordering::Acquire);
+            return (0, size);
+        }
+
+        let mut total_size: u64 = 0;
+        let mut max_seqno: u64 = 0;
+
+        let overhead =
+            std::mem::size_of::<InternalValue>() + std::mem::size_of::<SharedComparator>();
+
+        for item in &items {
+            #[expect(
+                clippy::expect_used,
+                reason = "keys are limited to 16-bit length + values are limited to 32-bit length"
+            )]
+            let item_size: u64 = (item.key.user_key.len() + item.value.len() + overhead)
+                .try_into()
+                .expect("should fit into u64");
+
+            total_size = total_size.saturating_add(item_size);
+
+            if item.key.seqno > max_seqno {
+                max_seqno = item.key.seqno;
+            }
+        }
+
+        let size_before = self
+            .approximate_size
+            .fetch_add(total_size, std::sync::atomic::Ordering::AcqRel);
+
+        for item in items {
+            let key = InternalKey::new(item.key.user_key, item.key.seqno, item.key.value_type);
+            self.items.insert(&key, &item.value);
+        }
+
+        self.highest_seqno
+            .fetch_max(max_seqno, std::sync::atomic::Ordering::AcqRel);
+
+        (total_size, size_before + total_size)
+    }
+
     /// Inserts an item into the memtable
     #[doc(hidden)]
     pub fn insert(&self, item: InternalValue) -> (u64, u64) {

src/pinnable_slice.rs

@@ -0,0 +1,142 @@
+// Copyright (c) 2024-present, fjall-rs
+// This source code is licensed under both the Apache 2.0 and MIT License
+// (found in the LICENSE-* files in the repository)
+
+//! Zero-copy value reference that keeps the decompressed block buffer alive.
+//!
+//! [`PinnableSlice`] is inspired by `RocksDB`'s `PinnableSlice`
+//! (`include/rocksdb/slice.h:179-263`). It wraps a value that was read from
+//! the LSM tree and indicates whether the underlying data shares the
+//! decompressed block buffer or is independently owned (e.g. from a memtable
+//! or merge result).
+//!
+//! When the value comes from an on-disk data block, holding a
+//! `PinnableSlice::Pinned` keeps the block's decompressed buffer alive
+//! (via the refcounted [`Slice`] / `ByteView` backing) for the duration of
+//! the reference. The value bytes are a sub-slice of that buffer — no copy
+//! is performed. Note: this does **not** prevent the block cache from
+//! evicting its entry; it only ensures the backing memory remains valid.
+//!
+//! Memtable and blob-resolved values use the `Owned` variant.
+
+use crate::{Slice, UserValue, table::Block};
+
+/// A value reference that may share the decompressed block buffer.
+///
+/// Use [`PinnableSlice::as_ref`] to access the raw bytes regardless of variant.
+///
+/// # Lifetime
+///
+/// The `Pinned` variant holds a [`Block`] clone whose `data` field is a
+/// refcounted [`Slice`]. As long as the `PinnableSlice` is alive, the
+/// decompressed block buffer remains valid. Dropping it releases the
+/// reference count on the underlying `ByteView` allocation.
+#[derive(Clone)]
+pub enum PinnableSlice {
+    /// Value sharing the decompressed block buffer — zero copy.
+    ///
+    /// The [`Block`] keeps the decompressed data alive via refcounted
+    /// `Slice` / `ByteView`. `value` is a sub-slice created via
+    /// [`Slice::slice`], sharing the same backing allocation.
+    Pinned {
+        /// Keeps the decompressed block buffer alive via refcount.
+        _block: Block,
+        /// Zero-copy sub-slice into the block's decompressed data.
+        value: Slice,
+    },
+
+    /// Value owned independently (memtable, blob, merge result).
+    Owned(UserValue),
+}
+
+impl PinnableSlice {
+    /// Creates a pinned value sharing the decompressed block buffer.
+    #[must_use]
+    pub fn pinned(block: Block, value: Slice) -> Self {
+        Self::Pinned {
+            _block: block,
+            value,
+        }
+    }
+
+    /// Creates an owned value (not sharing any block buffer).
+    #[must_use]
+    pub fn owned(value: UserValue) -> Self {
+        Self::Owned(value)
+    }
+
+    /// Returns `true` if this value shares the decompressed block buffer.
+    #[must_use]
+    pub fn is_pinned(&self) -> bool {
+        matches!(self, Self::Pinned { .. })
+    }
+
+    /// Returns the raw value bytes.
+    #[must_use]
+    pub fn value(&self) -> &[u8] {
+        self.as_ref()
+    }
+
+    /// Returns the length of the value in bytes.
+    #[must_use]
+    pub fn len(&self) -> usize {
+        self.as_ref().len()
+    }
+
+    /// Returns `true` if the value is empty.
+    #[must_use]
+    pub fn is_empty(&self) -> bool {
+        self.as_ref().is_empty()
+    }
+
+    /// Converts this `PinnableSlice` into an owned `UserValue`.
+    ///
+    /// For the `Pinned` variant, the `Block` is dropped but the returned
+    /// `Slice` still shares the same `ByteView` backing allocation.
+    /// For the `Owned` variant, the value is returned directly.
+    #[must_use]
+    pub fn into_value(self) -> UserValue {
+        match self {
+            Self::Pinned { value, .. } => value,
+            Self::Owned(v) => v,
+        }
+    }
+}
+
+impl std::fmt::Debug for PinnableSlice {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Pinned { value, .. } => {
+                f.debug_struct("Pinned").field("len", &value.len()).finish()
+            }
+            Self::Owned(v) => f.debug_tuple("Owned").field(&v.len()).finish(),
+        }
+    }
+}
+
+impl AsRef<[u8]> for PinnableSlice {
+    fn as_ref(&self) -> &[u8] {
+        match self {
+            Self::Pinned { value, .. } => value.as_ref(),
+            Self::Owned(v) => v.as_ref(),
+        }
+    }
+}
+
+impl PartialEq<[u8]> for PinnableSlice {
+    fn eq(&self, other: &[u8]) -> bool {
+        self.as_ref() == other
+    }
+}
+
+impl PartialEq<&[u8]> for PinnableSlice {
+    fn eq(&self, other: &&[u8]) -> bool {
+        self.as_ref() == *other
+    }
+}
+
+impl From<PinnableSlice> for UserValue {
+    fn from(ps: PinnableSlice) -> Self {
+        ps.into_value()
+    }
+}