feat(txn)!: return scan rows as Bytes

BREAKING: the `ReadTxn` scans return `Vec<(Bytes, Bytes)>` rather than
`Vec<(Vec<u8>, Vec<u8>)>`, matching `get`.

Done now for the same reason `get` was: the signature is the part that cannot
change after 0.1. Rows still cost an allocation each — keys are prefix-
compressed, so a full key is not contiguous in the page and has to be
reassembled — but with the type in place that becomes an internal
optimisation rather than another breaking change.

Callers that stage scan output into pages they own — the compaction repack
loader, the history relocator, the rekey segment walk — copy at the boundary,
which is what they already did.

Author: farhan-syah

src/btree/tests/basic.rs

@@ -180,7 +180,7 @@ async fn forward_scan_returns_sorted() {
     let got = tree.collect_range(b"k0010", b"k0020").await.unwrap();
     let keys: Vec<String> = got
         .into_iter()
-        .map(|(k, _)| String::from_utf8(k).unwrap())
+        .map(|(k, _)| String::from_utf8(k.to_vec()).unwrap())
         .collect();
     let expected: Vec<String> = (10..20).map(|i| format!("k{i:04}")).collect();
     assert_eq!(keys, expected);
@@ -206,7 +206,7 @@ async fn collect_all_returns_every_key_including_the_top_of_the_keyspace() {
 
     let all = tree.collect_all().await.unwrap();
     assert_eq!(all.len(), 202);
-    let keys: Vec<&[u8]> = all.iter().map(|(k, _)| k.as_slice()).collect();
+    let keys: Vec<&[u8]> = all.iter().map(|(k, _)| k.as_ref()).collect();
     assert!(keys.windows(2).all(|w| w[0] < w[1]), "not ascending");
     assert_eq!(keys[200], &[0xFF; 256]);
     assert_eq!(keys[201], beyond.as_slice());
@@ -816,7 +816,7 @@ async fn bulk_loader_builds_a_multi_level_tree_that_scans_in_order() {
     let all = tree.collect_all().await.unwrap();
     assert_eq!(all.len(), n);
     for (i, (key, value)) in all.iter().enumerate() {
-        assert_eq!(key.as_slice(), format!("k-{i:05}").as_bytes());
+        assert_eq!(key.as_ref(), format!("k-{i:05}").as_bytes());
         let want = if i % 11 == 0 { &spilled } else { &inline };
         assert_eq!(value, want, "value mismatch at record {i}");
     }

src/btree/tests/tree_ops.rs

@@ -100,7 +100,7 @@ async fn scan_rev_returns_descending() {
     let got = tree.scan_rev(b"k0010", b"k0020").await.unwrap();
     let keys: Vec<String> = got
         .into_iter()
-        .map(|(k, _)| String::from_utf8(k).unwrap())
+        .map(|(k, _)| String::from_utf8(k.to_vec()).unwrap())
         .collect();
     let expected: Vec<String> = (10..20).rev().map(|i| format!("k{i:04}")).collect();
     assert_eq!(keys, expected);
@@ -115,7 +115,7 @@ async fn scan_prefix_short_circuits() {
         tree.put(word.as_bytes(), &v).await.unwrap();
     }
     let got = tree.scan_prefix(b"app").await.unwrap();
-    let keys: Vec<&[u8]> = got.iter().map(|(k, _)| k.as_slice()).collect();
+    let keys: Vec<&[u8]> = got.iter().map(|(k, _)| k.as_ref()).collect();
     assert_eq!(keys, vec![b"apple".as_ref(), b"apply".as_ref()]);
 }
 

src/btree/tree/read.rs

@@ -170,13 +170,15 @@ impl<V: Vfs> BTree<V> {
 
     /// Resolve a `LeafValue` to its raw bytes. Follows overflow chains as
     /// needed.
-    pub(super) async fn resolve_leaf_value(&self, v: &LeafValue) -> Result<Vec<u8>> {
+    pub(super) async fn resolve_leaf_value(&self, v: &LeafValue) -> Result<Bytes> {
         match v {
-            LeafValue::Inline(b) => Ok(b.clone()),
+            LeafValue::Inline(b) => Ok(Bytes::copy_from_slice(b)),
             LeafValue::Overflow {
                 total_len,
                 root_page_id,
-            } => overflow::read_chain(&self.pager, self.realm_id, *root_page_id, *total_len).await,
+            } => overflow::read_chain(&self.pager, self.realm_id, *root_page_id, *total_len)
+                .await
+                .map(Bytes::from),
         }
     }
 }

src/btree/tree/scan.rs

@@ -1,5 +1,7 @@
 //! Range, reverse, and prefix scans.
 
+use bytes::Bytes;
+
 use crate::Result;
 use crate::vfs::Vfs;
 
@@ -33,13 +35,13 @@ impl<V: Vfs> BTree<V> {
     /// lets the write path skip sibling-pointer `CoW` (saves ~2 leaf rewrites
     /// per put) at the cost of one extra internal-node lookup per leaf
     /// boundary crossed during a scan.
-    pub async fn collect_range(&self, start: &[u8], end: &[u8]) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
+    pub async fn collect_range(&self, start: &[u8], end: &[u8]) -> Result<Vec<(Bytes, Bytes)>> {
         if self.root_page_id == 0 {
             return Ok(Vec::new());
         }
         let mut path = self.path_to_leaf_for_key(start).await?;
         let mut seen_leaves = scan_guard();
-        let mut out: Vec<(Vec<u8>, Vec<u8>)> = Vec::new();
+        let mut out: Vec<(Bytes, Bytes)> = Vec::new();
         loop {
             let leaf_id = *path.last().expect("non-empty path");
             seen_leaves.insert(leaf_id)?;
@@ -50,7 +52,7 @@ impl<V: Vfs> BTree<V> {
                 }
                 if k.as_slice() >= start {
                     let val = self.resolve_leaf_value(v).await?;
-                    out.push((k.clone(), val));
+                    out.push((Bytes::copy_from_slice(k), val));
                 }
             }
             match self.next_leaf_after(&path).await? {
@@ -68,22 +70,22 @@ impl<V: Vfs> BTree<V> {
     /// bound is beyond the valid key domain: a bounded "scan everything" would
     /// silently drop records at the top of the keyspace (`[0xFF; N]` and any
     /// key extending it). Callers that need the whole tree must use this.
-    pub async fn collect_all(&self) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
+    pub async fn collect_all(&self) -> Result<Vec<(Bytes, Bytes)>> {
         if self.root_page_id == 0 {
             return Ok(Vec::new());
         }
         // The empty key sorts below every stored key, so the descent lands on
         // the leftmost leaf.
         let mut path = self.path_to_leaf_for_key(&[]).await?;
         let mut seen_leaves = scan_guard();
-        let mut out: Vec<(Vec<u8>, Vec<u8>)> = Vec::new();
+        let mut out: Vec<(Bytes, Bytes)> = Vec::new();
         loop {
             let leaf_id = *path.last().expect("non-empty path");
             seen_leaves.insert(leaf_id)?;
             let leaf = self.read_leaf(leaf_id).await?;
             for (k, v) in &leaf.records {
                 let val = self.resolve_leaf_value(v).await?;
-                out.push((k.clone(), val));
+                out.push((Bytes::copy_from_slice(k), val));
             }
             match self.next_leaf_after(&path).await? {
                 Some(next_path) => path = next_path,
@@ -109,13 +111,13 @@ impl<V: Vfs> BTree<V> {
         &self,
         start: &[u8],
         limit: usize,
-    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
+    ) -> Result<Vec<(Bytes, Bytes)>> {
         if self.root_page_id == 0 || limit == 0 {
             return Ok(Vec::new());
         }
         let mut path = self.path_to_leaf_for_key(start).await?;
         let mut seen_leaves = scan_guard();
-        let mut out: Vec<(Vec<u8>, Vec<u8>)> = Vec::with_capacity(limit);
+        let mut out: Vec<(Bytes, Bytes)> = Vec::with_capacity(limit);
         loop {
             let leaf_id = *path.last().expect("non-empty path");
             seen_leaves.insert(leaf_id)?;
@@ -128,7 +130,7 @@ impl<V: Vfs> BTree<V> {
                     return Ok(out);
                 }
                 let val = self.resolve_leaf_value(v).await?;
-                out.push((k.clone(), val));
+                out.push((Bytes::copy_from_slice(k), val));
             }
             if out.len() == limit {
                 return Ok(out);
@@ -155,7 +157,7 @@ impl<V: Vfs> BTree<V> {
         prefix: &[u8],
         start: &[u8],
         limit: usize,
-    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
+    ) -> Result<Vec<(Bytes, Bytes)>> {
         let mut batch = self.collect_batch_from(start, limit).await?;
         if let Some(end) = batch.iter().position(|(key, _)| !key.starts_with(prefix)) {
             batch.truncate(end);
@@ -179,21 +181,21 @@ impl<V: Vfs> BTree<V> {
 
     /// Reverse range scan: `start` inclusive, `end` exclusive. Returns results
     /// in descending key order. Collects matching records forward then reverses.
-    pub async fn scan_rev(&self, start: &[u8], end: &[u8]) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
+    pub async fn scan_rev(&self, start: &[u8], end: &[u8]) -> Result<Vec<(Bytes, Bytes)>> {
         let mut forward = self.collect_range(start, end).await?;
         forward.reverse();
         Ok(forward)
     }
 
     /// Prefix scan: returns all records whose key starts with `prefix`, in
     /// ascending order.
-    pub async fn scan_prefix(&self, prefix: &[u8]) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
+    pub async fn scan_prefix(&self, prefix: &[u8]) -> Result<Vec<(Bytes, Bytes)>> {
         if self.root_page_id == 0 {
             return Ok(Vec::new());
         }
         let mut path = self.path_to_leaf_for_key(prefix).await?;
         let mut seen_leaves = scan_guard();
-        let mut out: Vec<(Vec<u8>, Vec<u8>)> = Vec::new();
+        let mut out: Vec<(Bytes, Bytes)> = Vec::new();
         loop {
             let leaf_id = *path.last().expect("non-empty path");
             seen_leaves.insert(leaf_id)?;
@@ -208,7 +210,7 @@ impl<V: Vfs> BTree<V> {
                     break;
                 }
                 let val = self.resolve_leaf_value(v).await?;
-                out.push((k.clone(), val));
+                out.push((Bytes::copy_from_slice(k), val));
             }
             if past_prefix {
                 return Ok(out);

src/compaction/helpers.rs

@@ -77,7 +77,9 @@ pub(super) async fn stream_dense_tree<V: Vfs + Clone>(
             if !keep(&key) {
                 continue;
             }
-            loader.push(key, value).await?;
+            // 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?;
             // 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.
@@ -136,7 +138,7 @@ pub(super) async fn segment_rows_from<V: Vfs + Clone>(
     let mut out = Vec::with_capacity(rows.len());
     for (key, value) in rows {
         let meta = Catalog::decode_segment_meta(&value)?;
-        out.push((key, meta));
+        out.push((key.to_vec(), meta));
     }
     Ok(out)
 }

src/txn/db/history_carry.rs

@@ -80,7 +80,14 @@ impl<V: Vfs + Clone> Db<V> {
             return Ok(unmoved);
         }
 
-        let rows = base_tree.collect_all().await?;
+        // The loader stages these into pages it owns, so the scan's borrowed
+        // view is copied out here.
+        let rows: Vec<(Vec<u8>, Vec<u8>)> = base_tree
+            .collect_all()
+            .await?
+            .into_iter()
+            .map(|(key, value)| (key.to_vec(), value.to_vec()))
+            .collect();
         let entry_count = u64::try_from(rows.len()).ok();
         let mut relocated = BTree::open(
             self.pager.clone(),

src/txn/db/rekey/segments.rs

@@ -500,7 +500,7 @@ impl<V: Vfs + Clone> Db<V> {
             .into_iter()
             .map(|(key, value)| {
                 Ok(SegmentEntry {
-                    key,
+                    key: key.to_vec(),
                     meta: Catalog::decode_segment_meta(&value)?,
                 })
             })

src/txn/read/txn.rs

@@ -122,7 +122,7 @@ impl<'db, V: Vfs + Clone> ReadTxn<'db, V> {
     /// To take the first few rows at or after a key, use
     /// [`scan_from`](Self::scan_from) — bounding by count there is what keeps a
     /// short read short.
-    pub async fn scan(&self, start: &[u8], end: &[u8]) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
+    pub async fn scan(&self, start: &[u8], end: &[u8]) -> Result<Vec<(Bytes, Bytes)>> {
         self.check_abort()?;
         self.tree().collect_range(start, end).await
     }
@@ -145,7 +145,7 @@ impl<'db, V: Vfs + Clone> ReadTxn<'db, V> {
     /// that is its immediate successor in the key ordering, so paging this way
     /// never skips a record and never returns one twice. A batch shorter than
     /// `limit` means the tree ended.
-    pub async fn scan_from(&self, start: &[u8], limit: usize) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
+    pub async fn scan_from(&self, start: &[u8], limit: usize) -> Result<Vec<(Bytes, Bytes)>> {
         self.check_abort()?;
         self.tree().collect_batch_from(start, limit).await
     }
@@ -162,19 +162,19 @@ impl<'db, V: Vfs + Clone> ReadTxn<'db, V> {
         prefix: &[u8],
         start: &[u8],
         limit: usize,
-    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
+    ) -> Result<Vec<(Bytes, Bytes)>> {
         self.check_abort()?;
         self.tree()
             .collect_prefix_batch_from(prefix, start, limit)
             .await
     }
 
-    pub async fn scan_rev(&self, start: &[u8], end: &[u8]) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
+    pub async fn scan_rev(&self, start: &[u8], end: &[u8]) -> Result<Vec<(Bytes, Bytes)>> {
         self.check_abort()?;
         self.tree().scan_rev(start, end).await
     }
 
-    pub async fn scan_prefix(&self, prefix: &[u8]) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
+    pub async fn scan_prefix(&self, prefix: &[u8]) -> Result<Vec<(Bytes, Bytes)>> {
         self.check_abort()?;
         self.tree().scan_prefix(prefix).await
     }

tests/txn_basic.rs

@@ -212,8 +212,8 @@ async fn scan_from_stops_at_limit() {
     let r = db.begin_read().await.unwrap();
     let rows = r.scan_from(b"row:0000", 10).await.unwrap();
     assert_eq!(rows.len(), 10);
-    assert_eq!(rows[0].0, b"row:0000");
-    assert_eq!(rows[9].0, b"row:0009");
+    assert_eq!(rows[0].0.as_ref(), b"row:0000");
+    assert_eq!(rows[9].0.as_ref(), b"row:0009");
 }
 
 #[tokio::test(flavor = "current_thread")]
@@ -222,9 +222,9 @@ async fn scan_from_starts_at_or_after_key() {
     let r = db.begin_read().await.unwrap();
     // Start key need not exist: the scan lands on the first row at or after it.
     let rows = r.scan_from(b"row:0020", 3).await.unwrap();
-    assert_eq!(rows[0].0, b"row:0020");
+    assert_eq!(rows[0].0.as_ref(), b"row:0020");
     let rows = r.scan_from(b"row:0019z", 1).await.unwrap();
-    assert_eq!(rows[0].0, b"row:0020");
+    assert_eq!(rows[0].0.as_ref(), b"row:0020");
 }
 
 #[tokio::test(flavor = "current_thread")]
@@ -233,7 +233,7 @@ async fn scan_from_short_batch_means_end_of_tree() {
     let r = db.begin_read().await.unwrap();
     let rows = r.scan_from(b"row:0045", 10).await.unwrap();
     assert_eq!(rows.len(), 5);
-    assert_eq!(rows[4].0, b"row:0049");
+    assert_eq!(rows[4].0.as_ref(), b"row:0049");
     assert!(r.scan_from(b"zzz", 10).await.unwrap().is_empty());
     assert!(r.scan_from(b"row:0000", 0).await.unwrap().is_empty());
 }
@@ -251,9 +251,9 @@ async fn scan_from_resume_protocol_pages_every_row_once() {
         if batch.is_empty() {
             break;
         }
-        cursor = batch.last().unwrap().0.clone();
+        cursor = batch.last().unwrap().0.to_vec();
         cursor.push(0x00);
-        seen.extend(batch.into_iter().map(|(k, _)| k));
+        seen.extend(batch.into_iter().map(|(k, _)| k.to_vec()));
     }
     assert_eq!(seen.len(), 50);
     let mut expected: Vec<Vec<u8>> = (0..50)