feat(btree)!: carry inline values as Bytes end to end

BREAKING: `put_batch` takes `Vec<(Bytes, Bytes)>` and `bulk_load` /
`BulkLoader::push` take their values as `Bytes`.

`LeafValue::Inline` held a `Vec<u8>`, so every hop between pages copied the
value even when nothing about it changed. A repack reads each record out of one
page and stages it into another; a history relocation and a rekey do the same.
With the read side already producing `Bytes`, each of those boundaries had to
copy back down to `Vec<u8>` purely to satisfy the type.

Hold it as `Bytes` instead. Values now move by refcount from the page they were
read out of, through the scan, into the loader staging the page they are
written to. Reads of an uncommitted or dirty leaf stop copying as well: they
share what the leaf already holds.

`put_batch` takes the row shape the scans return, so read output feeds straight
back in without a conversion — the asymmetry only existed because the two sides
were changed at different times.

Keys are untouched: prefix compression means a full key is not contiguous in
the page, so a record owns its key either way.

Bulk load allocates 1885 -> 1646 bytes per row.

`PageGuard::body` also stops copying, which it only did because the buffer
behind it used to be a `Vec`.

Author: farhan-syah

CHANGELOG.md

@@ -14,7 +14,8 @@ version, since none exists yet.
 
 - **B+ tree surface** — sorted `bytes → bytes` store with copy-on-write shadow
   paging, A/B headers, ACID transactions, range scans, monotonic append, and
-  bulk load.
+  bulk load. Reads return `Bytes` borrowed from the page cache; scans come
+  bounded (`scan_from`, `scan_prefix_from`) or materialising.
 - **Segment File API** — engine-owned, append-mostly, atomically sealed
   encrypted files for formats that own their own layout (vectors, columnar
   blocks, FTS postings, R-trees).

src/btree/leaf.rs

@@ -1,5 +1,7 @@
 //! Leaf node operations.
 
+use bytes::Bytes;
+
 use crate::Result;
 use crate::errors::PagedbError;
 use crate::pager::PageGuard;
@@ -15,8 +17,14 @@ use super::node::{
 /// overflow chain.
 #[derive(Debug, Clone)]
 pub enum LeafValue {
-    Inline(Vec<u8>),
-    Overflow { total_len: u64, root_page_id: u64 },
+    /// Held as [`Bytes`] so a value read out of one page and written into
+    /// another — a repack, a history relocation, a rekey — moves by refcount
+    /// instead of being copied at each hop.
+    Inline(Bytes),
+    Overflow {
+        total_len: u64,
+        root_page_id: u64,
+    },
 }
 
 /// Bytes an overflow reference occupies in an encoded record body: the
@@ -78,8 +86,8 @@ impl Leaf {
                 }
             } else {
                 let vlen = value_len_raw as usize;
-                let v = body[off + 2 + suffix_len + 2..off + 2 + suffix_len + 2 + vlen].to_vec();
-                LeafValue::Inline(v)
+                let start = off + 2 + suffix_len + 2;
+                LeafValue::Inline(Bytes::copy_from_slice(&body[start..start + vlen]))
             };
             // Reconstruct full key by prepending the common prefix.
             let mut full_key = prefix_bytes.clone();
@@ -439,14 +447,14 @@ mod tests {
     #[test]
     fn inline_round_trip() {
         let mut leaf = Leaf::new();
-        leaf.upsert(b"hello", LeafValue::Inline(b"world".to_vec()));
-        leaf.upsert(b"hello2", LeafValue::Inline(b"world2".to_vec()));
+        leaf.upsert(b"hello", LeafValue::Inline(b"world".as_slice().into()));
+        leaf.upsert(b"hello2", LeafValue::Inline(b"world2".as_slice().into()));
         let mut body = make_body();
         leaf.encode(&mut body).unwrap();
         let decoded = Leaf::decode(&body).unwrap();
         assert_eq!(decoded.records.len(), 2);
         match &decoded.records[0].1 {
-            LeafValue::Inline(v) => assert_eq!(v, b"world"),
+            LeafValue::Inline(v) => assert_eq!(v.as_ref(), b"world"),
             LeafValue::Overflow { .. } => panic!("expected inline"),
         }
     }
@@ -479,9 +487,9 @@ mod tests {
     #[test]
     fn prefix_compression_applied() {
         let mut leaf = Leaf::new();
-        leaf.upsert(b"prefix/aaa", LeafValue::Inline(b"v1".to_vec()));
-        leaf.upsert(b"prefix/bbb", LeafValue::Inline(b"v2".to_vec()));
-        leaf.upsert(b"prefix/ccc", LeafValue::Inline(b"v3".to_vec()));
+        leaf.upsert(b"prefix/aaa", LeafValue::Inline(b"v1".as_slice().into()));
+        leaf.upsert(b"prefix/bbb", LeafValue::Inline(b"v2".as_slice().into()));
+        leaf.upsert(b"prefix/ccc", LeafValue::Inline(b"v3".as_slice().into()));
         let mut body = make_body();
         leaf.encode(&mut body).unwrap();
         // Check prefix_len in header is 7 ("prefix/")
@@ -497,11 +505,11 @@ mod tests {
     fn lcp_edge_cases() {
         let records: Vec<(Vec<u8>, LeafValue)> = Vec::new();
         assert_eq!(lcp(&records), b"");
-        let one = vec![(b"hello".to_vec(), LeafValue::Inline(vec![]))];
+        let one = vec![(b"hello".to_vec(), LeafValue::Inline(Bytes::new()))];
         assert_eq!(lcp(&one), b"hello");
         let two = vec![
-            (b"abc".to_vec(), LeafValue::Inline(vec![])),
-            (b"abd".to_vec(), LeafValue::Inline(vec![])),
+            (b"abc".to_vec(), LeafValue::Inline(Bytes::new())),
+            (b"abd".to_vec(), LeafValue::Inline(Bytes::new())),
         ];
         assert_eq!(lcp(&two), b"ab");
     }

src/btree/tests/basic.rs

@@ -2,6 +2,8 @@ use std::sync::Arc;
 use std::sync::mpsc;
 use std::time::Duration;
 
+use bytes::Bytes;
+
 use crate::btree::BTree;
 use crate::btree::internal::{Internal, InternalEntry};
 use crate::btree::leaf::{Leaf, LeafValue};
@@ -328,7 +330,7 @@ async fn read_node_rejects_authenticated_envelope_body_kind_mismatch() {
     let realm = RealmId::new([1; 16]);
 
     let mut leaf = Leaf::new();
-    leaf.upsert(b"k", LeafValue::Inline(b"v".to_vec()));
+    leaf.upsert(b"k", LeafValue::Inline(b"v".as_slice().into()));
     let mut leaf_body = vec![0u8; body_capacity(PAGE)];
     leaf.encode(&mut leaf_body).unwrap();
 
@@ -468,7 +470,7 @@ fn delete_range_rejects_leaf_sibling_cycle_without_hanging() {
 
             let mut leaf = Leaf::new();
             leaf.right_sibling = leaf_page_id;
-            leaf.upsert(b"k", LeafValue::Inline(b"v".to_vec()));
+            leaf.upsert(b"k", LeafValue::Inline(b"v".as_slice().into()));
             let mut body = vec![0u8; body_capacity(PAGE)];
             leaf.encode(&mut body).unwrap();
             pager
@@ -607,7 +609,7 @@ fn collect_all_rejects_alternating_leaf_successors_without_hanging() {
 
             for (page_id, key) in [(left_leaf, b"a"), (right_leaf, b"b")] {
                 let mut leaf = Leaf::new();
-                leaf.upsert(key, LeafValue::Inline(b"v".to_vec()));
+                leaf.upsert(key, LeafValue::Inline(b"v".as_slice().into()));
                 let mut body = vec![0u8; body_capacity(PAGE)];
                 leaf.encode(&mut body).unwrap();
                 pager
@@ -674,8 +676,8 @@ fn bulk_load_rejects_separator_that_cannot_fit_without_hanging() {
             let mut tree = fresh_tree(pager);
             let key_len = body_capacity(PAGE) - 32;
             tree.bulk_load(vec![
-                (vec![b'a'; key_len], Vec::new()),
-                (vec![b'b'; key_len], Vec::new()),
+                (vec![b'a'; key_len], Bytes::new()),
+                (vec![b'b'; key_len], Bytes::new()),
             ])
             .await
         });
@@ -693,12 +695,12 @@ fn bulk_load_rejects_separator_that_cannot_fit_without_hanging() {
 async fn bulk_load_rejects_non_strict_key_order_without_poisoning_tree() {
     let cases = [
         vec![
-            (b"b".to_vec(), b"two".to_vec()),
-            (b"a".to_vec(), b"one".to_vec()),
+            (b"b".to_vec(), Bytes::from_static(b"two")),
+            (b"a".to_vec(), Bytes::from_static(b"one")),
         ],
         vec![
-            (b"a".to_vec(), b"one".to_vec()),
-            (b"a".to_vec(), b"two".to_vec()),
+            (b"a".to_vec(), Bytes::from_static(b"one")),
+            (b"a".to_vec(), Bytes::from_static(b"two")),
         ],
     ];
 
@@ -717,8 +719,8 @@ async fn bulk_load_rejects_non_strict_key_order_without_poisoning_tree() {
         assert_eq!(tree.next_page_id(), 4);
 
         tree.bulk_load(vec![
-            (b"a".to_vec(), b"one".to_vec()),
-            (b"b".to_vec(), b"two".to_vec()),
+            (b"a".to_vec(), b"one".as_slice().into()),
+            (b"b".to_vec(), b"two".as_slice().into()),
         ])
         .await
         .unwrap();
@@ -743,9 +745,12 @@ async fn bulk_loader_rejects_non_increasing_keys_across_pushes() {
         let mut tree = fresh_tree(pager);
         {
             let mut loader = tree.bulk_loader().unwrap();
-            loader.push(b"b".to_vec(), b"one".to_vec()).await.unwrap();
+            loader
+                .push(b"b".to_vec(), b"one".as_slice().into())
+                .await
+                .unwrap();
             let error = loader
-                .push(second_key.to_vec(), b"two".to_vec())
+                .push(second_key.to_vec(), b"two".as_slice().into())
                 .await
                 .expect_err("descending and duplicate keys must both be rejected");
             assert!(
@@ -769,7 +774,7 @@ async fn bulk_loader_rejects_oversized_key_before_allocating() {
     {
         let mut loader = tree.bulk_loader().unwrap();
         let error = loader
-            .push(vec![b'a'; key_len], Vec::new())
+            .push(vec![b'a'; key_len], Bytes::new())
             .await
             .expect_err("a key too large to separate must be rejected");
         assert!(matches!(error, PagedbError::PayloadTooLarge));
@@ -800,7 +805,7 @@ async fn bulk_loader_builds_a_multi_level_tree_that_scans_in_order() {
         for i in 0..n {
             let value = if i % 11 == 0 { &spilled } else { &inline };
             loader
-                .push(format!("k-{i:05}").into_bytes(), value.clone())
+                .push(format!("k-{i:05}").into_bytes(), value.clone().into())
                 .await
                 .unwrap();
         }

src/btree/tests/generated_nodes.rs

@@ -78,7 +78,7 @@ fn encoded_leaf(records: &[(Vec<u8>, Vec<u8>)], overflow_every: Option<u64>) ->
                 total_len: value.len() as u64,
                 root_page_id,
             },
-            _ => LeafValue::Inline(value.clone()),
+            _ => LeafValue::Inline(value.clone().into()),
         };
         leaf.upsert(key, value);
     }

src/btree/tests/generated_structural_walks.rs

@@ -187,7 +187,7 @@ async fn lay_out_graph(pager: &Arc<Pager<MemVfs>>, specs: &[PageSpec]) {
                 leaf.left_sibling = slot_to_page_id(*left_sibling);
                 leaf.right_sibling = slot_to_page_id(*right_sibling);
                 for key in keys {
-                    leaf.upsert(&[*key], LeafValue::Inline(vec![*key; 4]));
+                    leaf.upsert(&[*key], LeafValue::Inline(vec![*key; 4].into()));
                 }
                 if leaf.encode(&mut body).is_err() {
                     continue;

src/btree/tests/tree_ops.rs

@@ -1,6 +1,8 @@
 use std::collections::BTreeMap;
 use std::sync::Arc;
 
+use bytes::Bytes;
+
 use crate::RealmId;
 use crate::btree::BTree;
 use crate::crypto::CipherId;
@@ -124,13 +126,18 @@ async fn put_batch_inserts_all() {
     let pager = fresh_pager().await;
     let mut tree = fresh_tree(pager);
     let v = vec![0u8; 16];
-    let batch: Vec<(Vec<u8>, Vec<u8>)> = (0..200u32)
-        .map(|i| (format!("k{i:04}").into_bytes(), v.clone()))
+    let batch: Vec<(Bytes, Bytes)> = (0..200u32)
+        .map(|i| {
+            (
+                Bytes::from(format!("k{i:04}").into_bytes()),
+                Bytes::from(v.clone()),
+            )
+        })
         .collect();
     tree.put_batch(batch.clone()).await.unwrap();
     for (k, expected) in &batch {
         let got = tree.get(k).await.unwrap();
-        assert_eq!(got.as_deref(), Some(expected.as_slice()));
+        assert_eq!(got.as_deref(), Some(expected.as_ref()));
     }
 }
 

src/btree/tree/bulk/loader.rs

@@ -7,6 +7,8 @@
 
 use std::sync::Arc;
 
+use bytes::Bytes;
+
 use crate::Result;
 use crate::errors::PagedbError;
 use crate::vfs::Vfs;
@@ -92,7 +94,10 @@ impl<'a, V: Vfs> BulkLoader<'a, V> {
     /// large for a leaf or a separator with `PayloadTooLarge`. Both checks run
     /// before this record allocates a page or touches the cache, so a rejected
     /// record leaves the tree exactly as it found it.
-    pub async fn push(&mut self, key: Vec<u8>, value: Vec<u8>) -> Result<()> {
+    /// `value` is taken as [`Bytes`] so a repack, which reads each record out
+    /// of one page and stages it into another, moves it by refcount instead of
+    /// copying it at the boundary.
+    pub async fn push(&mut self, key: Vec<u8>, value: Bytes) -> Result<()> {
         if let Some(last) = &self.last_key {
             if key <= *last {
                 return Err(PagedbError::BulkLoadNotMonotonic);
@@ -287,7 +292,7 @@ impl<V: Vfs> BTree<V> {
     /// caller holding a whole input is the incremental-apply history carry,
     /// which is native-only.
     #[cfg(any(test, not(target_arch = "wasm32")))]
-    pub async fn bulk_load(&mut self, pairs: Vec<(Vec<u8>, Vec<u8>)>) -> Result<()> {
+    pub async fn bulk_load(&mut self, pairs: Vec<(Vec<u8>, Bytes)>) -> Result<()> {
         let mut loader = self.bulk_loader()?;
         for (key, value) in pairs {
             loader.push(key, value).await?;

src/btree/tree/read.rs

@@ -48,9 +48,9 @@ impl<V: Vfs> BTree<V> {
     ) -> Result<Option<Bytes>> {
         match leaf.get(key) {
             None => Ok(None),
-            // Uncommitted, so it lives in the decoded leaf rather than in any
-            // page a slice could borrow from.
-            Some(LeafValue::Inline(v)) => Ok(Some(Bytes::copy_from_slice(v))),
+            // Uncommitted: held by the decoded leaf rather than by a page, but
+            // still shared rather than copied.
+            Some(LeafValue::Inline(v)) => Ok(Some(v.clone())),
             Some(LeafValue::Overflow {
                 total_len,
                 root_page_id,
@@ -136,9 +136,9 @@ impl<V: Vfs> BTree<V> {
         {
             return match leaf.get(key) {
                 None => Ok(None),
-                // Uncommitted, so it lives in the decoded leaf rather than in
-                // any page a slice could borrow from.
-                Some(LeafValue::Inline(v)) => Ok(Some(Bytes::copy_from_slice(v))),
+                // Uncommitted: held by the decoded leaf rather than by a page,
+                // but still shared rather than copied.
+                Some(LeafValue::Inline(v)) => Ok(Some(v.clone())),
                 Some(LeafValue::Overflow {
                     total_len,
                     root_page_id,
@@ -172,7 +172,8 @@ impl<V: Vfs> BTree<V> {
     /// needed.
     pub(super) async fn resolve_leaf_value(&self, v: &LeafValue) -> Result<Bytes> {
         match v {
-            LeafValue::Inline(b) => Ok(Bytes::copy_from_slice(b)),
+            // Refcount bump, not a copy.
+            LeafValue::Inline(b) => Ok(b.clone()),
             LeafValue::Overflow {
                 total_len,
                 root_page_id,

src/btree/tree/write.rs

@@ -2,6 +2,8 @@
 
 use std::sync::Arc;
 
+use bytes::Bytes;
+
 use crate::Result;
 use crate::errors::PagedbError;
 use crate::vfs::Vfs;
@@ -102,7 +104,7 @@ impl<V: Vfs> BTree<V> {
                 root_page_id,
             })
         } else {
-            Ok(LeafValue::Inline(value.to_vec()))
+            Ok(LeafValue::Inline(Bytes::copy_from_slice(value)))
         }
     }
 
@@ -378,7 +380,7 @@ impl<V: Vfs> BTree<V> {
     ///
     /// Per-leaf batching (amortising `CoW`) is a deferred performance
     /// optimisation; this implementation is correct for all inputs.
-    pub async fn put_batch(&mut self, pairs: Vec<(Vec<u8>, Vec<u8>)>) -> Result<()> {
+    pub async fn put_batch(&mut self, pairs: Vec<(Bytes, Bytes)>) -> Result<()> {
         for (k, v) in pairs {
             self.put(&k, &v).await?;
         }

src/compaction/helpers.rs

@@ -77,9 +77,9 @@ pub(super) async fn stream_dense_tree<V: Vfs + Clone>(
             if !keep(&key) {
                 continue;
             }
-            // The loader owns what it stages into pages it is building, so the
-            // scan's borrowed view has to be copied out here.
-            loader.push(key.to_vec(), value.to_vec()).await?;
+            // The value carries through by refcount; only the key, which the
+            // staged record owns, is copied.
+            loader.push(key.to_vec(), value).await?;
             // A compacted page is written once and never read back, so flushing
             // it out mid-batch costs nothing and keeps the pool from growing
             // with the tree instead of with the budget.