perf: binary-search index leaf position instead of full-page decode

src/btree/index.rs

@@ -473,12 +473,20 @@ pub(super) fn collect_index_leaf_cells(
 ) -> Result<Vec<IndexLeafCell>, BtreeError> {
     let num_cells = read_num_cells(buf, header_start, page_num)?;
     let ptr_base = header_start.saturating_add(8);
+    let page: Rc<[u8]> = Rc::from(buf);
     let mut out = Vec::with_capacity(num_cells);
     for i in 0..num_cells {
         let ptr_off = cell_ptr_offset(ptr_base, i);
         let cell_start = read_cell_pointer(buf, ptr_off, page_num, i)?;
-        let (key, cell_bytes) =
-            decode_value_cell(source, buf, cell_start, page_num, usable_size, encoding)?;
+        let (key, cell_bytes) = decode_value_cell(
+            source,
+            buf,
+            &page,
+            cell_start,
+            page_num,
+            usable_size,
+            encoding,
+        )?;
         out.push((key, cell_bytes));
     }
     Ok(out)
@@ -499,14 +507,22 @@ pub(super) fn collect_index_interior_entries(
 ) -> Result<(Vec<IndexInteriorEntry>, u32), BtreeError> {
     let num_cells = read_num_cells(buf, header_start, page_num)?;
     let ptr_base = header_start.saturating_add(12);
+    let page: Rc<[u8]> = Rc::from(buf);
     let mut out = Vec::with_capacity(num_cells);
     for i in 0..num_cells {
         let ptr_off = cell_ptr_offset(ptr_base, i);
         let cell_start = read_cell_pointer(buf, ptr_off, page_num, i)?;
         let child = read_u32(buf, cell_start, page_num)?;
         let value_start = cell_start.saturating_add(4);
-        let (key, cell_bytes) =
-            decode_value_cell(source, buf, value_start, page_num, usable_size, encoding)?;
+        let (key, cell_bytes) = decode_value_cell(
+            source,
+            buf,
+            &page,
+            value_start,
+            page_num,
+            usable_size,
+            encoding,
+        )?;
         out.push((child, key, cell_bytes));
     }
     let rightmost = read_u32(buf, header_start.saturating_add(8), page_num)?;
@@ -518,9 +534,14 @@ pub(super) fn collect_index_interior_entries(
 /// interior cells: returns the decoded key (for ordering) and the raw,
 /// verbatim cell bytes (varint + local bytes + optional overflow
 /// pointer), starting at `value_start`.
+/// `page` must be the same bytes as `buf`, shared as an `Rc` so
+/// `reassemble_payload` can follow an overflow chain past `buf`'s
+/// borrow — callers decoding multiple cells off one page build `page`
+/// once and pass it in, rather than paying a full-page copy per cell.
 fn decode_value_cell(
     source: &Pager,
     buf: &[u8],
+    page: &Rc<[u8]>,
     value_start: usize,
     page_num: u32,
     usable_size: u32,
@@ -536,12 +557,11 @@ fn decode_value_cell(
         .get(value_start..cell_end)
         .ok_or(BtreeError::PayloadTooShort { page_num })?
         .to_vec();
-    let page: Rc<[u8]> = Rc::from(buf);
     let payload = reassemble_payload(
         source,
         usable_size,
         page_num,
-        &page,
+        page,
         tail_start,
         payload_len,
         true,
@@ -550,6 +570,76 @@ fn decode_value_cell(
     Ok((key, cell_bytes))
 }
 
+/// Byte length of the value-cell (payload-length varint, local payload,
+/// plus optional 4-byte overflow pointer) starting at `cell_start`,
+/// without decoding its key or copying its bytes. Used for page-space
+/// bookkeeping where only the cell's on-disk size matters, not its sort
+/// key — letting [`super::insert::insert_into_index_leaf`]'s space check
+/// avoid a full [`decode_value_cell`] per existing cell.
+pub(super) fn value_cell_len(
+    buf: &[u8],
+    cell_start: usize,
+    page_num: u32,
+    usable_size: u32,
+) -> Result<usize, BtreeError> {
+    let (payload_len, tail_start) = decode_payload_len(buf, cell_start, page_num)?;
+    let local_size = local_payload_size(usable_size, payload_len, true) as usize;
+    let has_overflow = (local_size as u64) < payload_len;
+    let cell_end = tail_start
+        .saturating_add(local_size)
+        .saturating_add(if has_overflow { 4 } else { 0 });
+    Ok(cell_end.saturating_sub(cell_start))
+}
+
+/// Outcome of [`search_index_leaf`]: either an exact key match (with its
+/// decoded cell, so callers don't need to decode it again) or the
+/// cell-pointer-array position a new entry with this key would sort into.
+pub(super) enum LeafSearch {
+    Found(usize, IndexLeafCell),
+    NotFound(usize),
+}
+
+/// Binary search for `key` among an index leaf page's sorted entries,
+/// decoding only the O(log n) cells actually compared — unlike
+/// [`collect_index_leaf_cells`], which decodes every cell on the page.
+/// Used by the index insert/delete write paths to locate a position
+/// without paying for a full-page decode on every write (#648).
+pub(super) fn search_index_leaf(
+    source: &Pager,
+    buf: &[u8],
+    header_start: usize,
+    page_num: u32,
+    usable_size: u32,
+    encoding: TextEncoding,
+    key: &[Value],
+) -> Result<LeafSearch, BtreeError> {
+    let num_cells = read_num_cells(buf, header_start, page_num)?;
+    let ptr_base = header_start.saturating_add(8);
+    let page: Rc<[u8]> = Rc::from(buf);
+    let mut lo = 0usize;
+    let mut hi = num_cells;
+    while lo < hi {
+        let mid = lo.saturating_add(hi.saturating_sub(lo) / 2);
+        let ptr_off = cell_ptr_offset(ptr_base, mid);
+        let cell_start = read_cell_pointer(buf, ptr_off, page_num, mid)?;
+        let decoded = decode_value_cell(
+            source,
+            buf,
+            &page,
+            cell_start,
+            page_num,
+            usable_size,
+            encoding,
+        )?;
+        match compare_keys(key, &decoded.0) {
+            Ordering::Equal => return Ok(LeafSearch::Found(mid, decoded)),
+            Ordering::Less => hi = mid,
+            Ordering::Greater => lo = mid.saturating_add(1),
+        }
+    }
+    Ok(LeafSearch::NotFound(lo))
+}
+
 /// Builds an index interior cell: 4-byte left-child page number followed
 /// by `value_cell_bytes` verbatim (the same payload-length-varint +
 /// payload shape as a leaf cell — index interior cells carry a full

src/btree/index/delete.rs

@@ -37,12 +37,10 @@
 //! everything in it was already deleted earlier), the matched entry is
 //! removed outright instead of swapped.
 
-use std::cmp::Ordering;
-
 use crate::btree::index::{
     build_index_interior_cell, collect_index_interior_entries, collect_index_leaf_cells,
-    compare_keys, decode_payload_len, descend_index_tree, write_index_interior_page, IndexDescent,
-    INTERIOR_INDEX, LEAF_INDEX,
+    decode_payload_len, descend_index_tree, search_index_leaf, write_index_interior_page,
+    IndexDescent, LeafSearch, INTERIOR_INDEX, LEAF_INDEX,
 };
 use crate::btree::{
     local_payload_size, page1_header_start, read_page_type, read_u32, splice_delete_cell,
@@ -134,20 +132,22 @@ fn delete_from_leaf(
 ) -> Result<(), BtreeError> {
     let header_start = page1_header_start(leaf_page);
     let buf = pager.get_page_mut(leaf_page)?.clone();
-    let cells =
-        collect_index_leaf_cells(pager, &buf, header_start, leaf_page, usable_size, encoding)?;
 
-    let pos = cells
-        .iter()
-        .position(|(existing_key, _)| compare_keys(existing_key, key) == Ordering::Equal)
-        .ok_or(BtreeError::KeyNotFound)?;
-    let overflow_page = overflow_page_of(
-        &cells
-            .get(pos)
-            .ok_or(BtreeError::Internal("delete_from_leaf: pos out of bounds"))?
-            .1,
+    // Binary search for the exact match, decoding only the O(log n)
+    // cells actually compared instead of every cell on the page (#648).
+    let (pos, matched_cell) = match search_index_leaf(
+        pager,
+        &buf,
+        header_start,
+        leaf_page,
         usable_size,
-    )?;
+        encoding,
+        key,
+    )? {
+        LeafSearch::Found(pos, cell) => (pos, cell),
+        LeafSearch::NotFound(_) => return Err(BtreeError::KeyNotFound),
+    };
+    let overflow_page = overflow_page_of(&matched_cell.1, usable_size)?;
 
     let buf = pager.get_page_mut(leaf_page)?;
     splice_delete_cell(buf, header_start, leaf_page, usable_size, pos, false)?;

src/btree/index/insert.rs

@@ -22,12 +22,12 @@
 
 use crate::btree::index::{
     build_index_interior_cell, collect_index_interior_entries, collect_index_leaf_cells,
-    compare_keys, descend_index_tree, write_index_interior_page, write_index_leaf_page,
-    IndexDescent, INTERIOR_INDEX, LEAF_INDEX,
+    descend_index_tree, search_index_leaf, value_cell_len, write_index_interior_page,
+    write_index_leaf_page, IndexDescent, LeafSearch, INTERIOR_INDEX, LEAF_INDEX,
 };
 use crate::btree::{
-    cell_bytes, local_payload_size, page1_header_start, put, read_page_type, splice_insert_cell,
-    BtreeError,
+    cell_bytes, cell_ptr_offset, local_payload_size, page1_header_start, put, read_cell_pointer,
+    read_num_cells, read_page_type, splice_insert_cell, BtreeError,
 };
 use crate::header::DatabaseHeader;
 use crate::pager::Pager;
@@ -151,41 +151,74 @@ fn insert_into_index_leaf(
 ) -> Result<(), BtreeError> {
     let header_start = page1_header_start(leaf_page);
     let buf = pager.get_page_mut(leaf_page)?.clone();
-    let mut cells =
-        collect_index_leaf_cells(pager, &buf, header_start, leaf_page, usable_size, encoding)?;
 
-    let mut insert_pos = cells.len();
-    for (i, (existing_key, _)) in cells.iter().enumerate() {
-        match compare_keys(key, existing_key) {
-            std::cmp::Ordering::Equal => return Err(BtreeError::DuplicateKey),
-            std::cmp::Ordering::Less => {
-                insert_pos = i;
-                break;
-            }
-            std::cmp::Ordering::Greater => {}
-        }
-    }
+    // Binary search for the insert position, decoding only the O(log n)
+    // cells actually compared — a full `collect_index_leaf_cells` decode
+    // of every cell on the page is deferred to the split/fallback paths
+    // below, which are the only ones that actually need every cell's
+    // contents (#648).
+    let insert_pos = match search_index_leaf(
+        pager,
+        &buf,
+        header_start,
+        leaf_page,
+        usable_size,
+        encoding,
+        key,
+    )? {
+        LeafSearch::Found(..) => return Err(BtreeError::DuplicateKey),
+        LeafSearch::NotFound(pos) => pos,
+    };
     // No duplicate found — safe to allocate overflow pages (if any) now.
     let cell = encode_index_cell(pager, usable_size, payload)?;
-    cells.insert(insert_pos, (key.to_vec(), cell.clone()));
 
-    let total_bytes: usize = cells.iter().map(|(_, c)| c.len()).sum();
+    let num_cells = read_num_cells(&buf, header_start, leaf_page)?;
+    let ptr_base = header_start.saturating_add(8);
+    let mut total_bytes = cell.len();
+    for i in 0..num_cells {
+        let ptr_off = cell_ptr_offset(ptr_base, i);
+        let cell_start = read_cell_pointer(&buf, ptr_off, leaf_page, i)?;
+        total_bytes =
+            total_bytes.saturating_add(value_cell_len(&buf, cell_start, leaf_page, usable_size)?);
+    }
     let header_len = 8;
     let needed = header_start
         .saturating_add(header_len)
-        .saturating_add(cells.len().saturating_mul(2))
+        .saturating_add(num_cells.saturating_add(1).saturating_mul(2))
         .saturating_add(total_bytes);
     if needed <= page_len {
-        let buf = pager.get_page_mut(leaf_page)?;
         // Fast path: splice directly into the page (O(1) relative to the
         // other cells) when there's enough contiguous free space; falls
-        // back to a full rebuild otherwise (see #337).
-        if !splice_insert_cell(buf, header_start, leaf_page, insert_pos, &cell)? {
-            write_index_leaf_page(buf, header_start, leaf_page, &cell_bytes(cells))?;
+        // back to a full rebuild otherwise (see #337). `buf` (the
+        // pre-mutation snapshot above) still matches on-disk content here,
+        // since nothing has written to the page yet.
+        let spliced = {
+            let page_buf = pager.get_page_mut(leaf_page)?;
+            splice_insert_cell(page_buf, header_start, leaf_page, insert_pos, &cell)?
+        };
+        if !spliced {
+            let mut cells = collect_index_leaf_cells(
+                pager,
+                &buf,
+                header_start,
+                leaf_page,
+                usable_size,
+                encoding,
+            )?;
+            cells.insert(insert_pos, (key.to_vec(), cell.clone()));
+            let page_buf = pager.get_page_mut(leaf_page)?;
+            write_index_leaf_page(page_buf, header_start, leaf_page, &cell_bytes(cells))?;
         }
         return Ok(());
     }
 
+    // Split: needs every cell's contents to redistribute across the two
+    // resulting pages, so the full decode is unavoidable (and correct)
+    // here.
+    let mut cells =
+        collect_index_leaf_cells(pager, &buf, header_start, leaf_page, usable_size, encoding)?;
+    cells.insert(insert_pos, (key.to_vec(), cell.clone()));
+
     // Split: the median entry is promoted into the parent (removed from
     // both halves); left keeps entries less than it, right (a freshly
     // allocated page) keeps entries greater.