bplus_index/

lib.rs

//! Arena-backed B+Tree for in-memory sorted indexes.
//!
//! `BPlusIndex<K, V>` is a sorted multimap where entries are `(K, V)` pairs.
//! Multiple values per key are supported natively -- the `(K, V)` pair is the
//! unique identity of each entry. This makes it suitable for secondary indexes
//! in data-oriented designs where one key maps to many row identifiers.
//!
//! # Design
//!
//! - **Arena storage**: all nodes live in contiguous `Vec`s, indexed by `u32` ids.
//!   No `Box`, no pointer chasing between siblings during range scans.
//! - **Cache-friendly leaves**: each leaf holds up to 64 keys and 64 values in
//!   inline `ArrayVec`s. Binary search within a leaf touches 1-2 cache lines.
//! - **Linked leaves**: leaves form a doubly-linked list for O(log n + k) range
//!   scans without climbing back up the tree.
//! - **Zero `unsafe`**: built on top of `arrayvec` which encapsulates all the
//!   `MaybeUninit` machinery.
//! - **Non-unique key support**: inner node separators store `(K, V)` tuples for
//!   correct routing even when thousands of entries share the same key.
//!
//! # Complexity
//!
//! | Operation      | Time           | Notes                                   |
//! |----------------|----------------|-----------------------------------------|
//! | `insert`       | O(log n)       | Amortized, with rare leaf/inner splits  |
//! | `delete`       | O(log n)       | With borrow-or-merge rebalancing        |
//! | `update`       | O(log n)       | Delete + insert                         |
//! | `lookup_unique`| O(log n)       | Returns first value for a given key     |
//! | `lookup_range` | O(log n + k)   | All values for a given key              |
//! | `range`        | O(log n + k)   | Arbitrary key range with bounds         |
//! | `iter`         | O(n)           | Full scan via leaf chain                |
//!
//! # Example
//!
//! ```
//! use bplus_index::BPlusIndex;
//!
//! let mut idx = BPlusIndex::<i32, u64>::new();
//!
//! // Non-unique: multiple values for the same key
//! idx.insert(10, 100);
//! idx.insert(10, 200);
//! idx.insert(20, 300);
//!
//! // Lookup all values for key 10
//! let vals: Vec<_> = idx.lookup_range(&10).copied().collect();
//! assert_eq!(vals, vec![100, 200]);
//!
//! // Range scan
//! let entries: Vec<_> = idx.range(5..15).collect();
//! assert_eq!(entries.len(), 2);
//! ```

use arrayvec::ArrayVec;
use std::cmp::Ordering;
use std::ops::{Bound, RangeBounds};

// ---------------------------------------------------------------------------
// Constants
// ---------------------------------------------------------------------------

/// Maximum number of keys per leaf node.
const LEAF_CAP: usize = 64;

/// Maximum number of keys (separators) per inner node.
const INNER_CAP: usize = 64;

/// Maximum number of children per inner node (INNER_CAP + 1).
const INNER_CHILDREN: usize = INNER_CAP + 1;

/// Minimum occupancy for a leaf before rebalancing is triggered.
const MIN_LEAF: usize = LEAF_CAP / 4;

/// Minimum occupancy for an inner node before rebalancing is triggered.
const MIN_INNER: usize = INNER_CAP / 4;

/// Sentinel value indicating "no node" in prev/next links.
const NONE: u32 = u32::MAX;

/// Maximum tree depth. Supports up to 64^20 entries (effectively unlimited).
const MAX_DEPTH: usize = 20;

// ---------------------------------------------------------------------------
// ArrayVec helpers
// ---------------------------------------------------------------------------

/// Insert `val` at position `idx`, shifting elements to the right.
fn av_insert<T, const N: usize>(av: &mut ArrayVec<T, N>, idx: usize, val: T) {
    av.insert(idx, val);
}

/// Split `av` at position `at`, returning a new `ArrayVec` with `[at..]`.
/// The original keeps `[..at]`.
fn av_split_off<T, const N: usize>(av: &mut ArrayVec<T, N>, at: usize) -> ArrayVec<T, N> {
    let mut right = ArrayVec::new();
    for item in av.drain(at..) {
        right.push(item);
    }
    right
}

/// Move all elements from `src` into `dst`.
fn av_append<T, const N: usize>(dst: &mut ArrayVec<T, N>, src: &mut ArrayVec<T, N>) {
    for item in src.drain(..) {
        dst.push(item);
    }
}

// ---------------------------------------------------------------------------
// Leaf node
// ---------------------------------------------------------------------------

/// A leaf node in the B+Tree.
///
/// Stores up to [`LEAF_CAP`] key-value pairs in sorted order. Leaves are
/// linked via `prev`/`next` indices for efficient range scans.
struct LeafNode<K, V> {
    keys: ArrayVec<K, LEAF_CAP>,
    vals: ArrayVec<V, LEAF_CAP>,
    prev: u32,
    next: u32,
}

impl<K, V> LeafNode<K, V> {
    fn new() -> Self {
        Self {
            keys: ArrayVec::new(),
            vals: ArrayVec::new(),
            prev: NONE,
            next: NONE,
        }
    }

    #[inline]
    fn len(&self) -> usize {
        self.keys.len()
    }

    #[inline]
    fn is_full(&self) -> bool {
        self.keys.is_full()
    }

    /// Returns `true` if this leaf has enough entries to lend one to a sibling.
    fn can_lend(&self) -> bool {
        self.keys.len() > MIN_LEAF
    }
}

impl<K: Ord, V> LeafNode<K, V> {
    /// Binary search on the composite `(K, V)` ordering.
    ///
    /// Returns `Ok(idx)` on exact match, `Err(idx)` for the insertion point.
    fn search_kv_pos(&self, k: &K, v: &V) -> Result<usize, usize>
    where
        V: Ord,
    {
        let mut lo = 0usize;
        let mut hi = self.len();
        while lo < hi {
            let mid = lo + (hi - lo) / 2;
            match self.keys[mid].cmp(k).then_with(|| self.vals[mid].cmp(v)) {
                Ordering::Less => lo = mid + 1,
                Ordering::Greater => hi = mid,
                Ordering::Equal => return Ok(mid),
            }
        }
        Err(lo)
    }

    /// First position where `key >= k` (lower bound).
    fn lower_bound_k(&self, k: &K) -> usize {
        self.keys.partition_point(|sk| sk < k)
    }

    /// First position where `key > k` (upper bound).
    fn upper_bound_k(&self, k: &K) -> usize {
        self.keys.partition_point(|sk| sk <= k)
    }
}

// ---------------------------------------------------------------------------
// Inner node
// ---------------------------------------------------------------------------

/// An inner (branch) node in the B+Tree.
///
/// Stores separators as `(K, V)` pairs rather than just `K`. This is required
/// for correct child routing when keys are not unique: if many entries share
/// the same K, the V component of the separator disambiguates which subtree
/// contains a given `(K, V)` pair.
struct InnerNode<K, V> {
    keys: ArrayVec<K, INNER_CAP>,
    vals: ArrayVec<V, INNER_CAP>,
    children: ArrayVec<u32, INNER_CHILDREN>,
}

impl<K, V> InnerNode<K, V> {
    fn new() -> Self {
        Self {
            keys: ArrayVec::new(),
            vals: ArrayVec::new(),
            children: ArrayVec::new(),
        }
    }

    fn len(&self) -> usize {
        self.keys.len()
    }

    fn is_full(&self) -> bool {
        self.keys.is_full()
    }

    /// Returns `true` if this inner node can lend a child to a sibling.
    fn can_lend(&self) -> bool {
        self.keys.len() > MIN_INNER
    }
}

impl<K: Ord, V: Ord> InnerNode<K, V> {
    /// Find the child index for a given `(K, V)` using upper-bound semantics.
    ///
    /// Returns the index of the first separator strictly greater than `(k, v)`.
    /// The child at that index is the correct subtree for descent.
    fn find_child_kv(&self, k: &K, v: &V) -> usize {
        let mut lo = 0usize;
        let mut hi = self.len();
        while lo < hi {
            let mid = lo + (hi - lo) / 2;
            match self.keys[mid].cmp(k).then_with(|| self.vals[mid].cmp(v)) {
                Ordering::Less | Ordering::Equal => lo = mid + 1,
                Ordering::Greater => hi = mid,
            }
        }
        lo
    }

    /// Find the child index using K-only lower-bound semantics.
    ///
    /// Used for `lookup_unique`, `lookup_range`, and `range` queries where we
    /// want to find the leftmost subtree that might contain a given key.
    fn find_child_k_lower(&self, k: &K) -> usize {
        self.keys.partition_point(|sk| sk < k)
    }
}

// ---------------------------------------------------------------------------
// Descent path (stack-allocated)
// ---------------------------------------------------------------------------

#[derive(Clone, Copy)]
struct PathEntry {
    node: u32,
    child_pos: usize,
}

/// Records the path from root to leaf during descent, so that insert/delete
/// can propagate splits and merges back up without re-descending.
struct Path {
    entries: [PathEntry; MAX_DEPTH],
    len: usize,
}

impl Path {
    fn new() -> Self {
        Self {
            entries: [PathEntry {
                node: 0,
                child_pos: 0,
            }; MAX_DEPTH],
            len: 0,
        }
    }

    fn push(&mut self, node: u32, child_pos: usize) {
        self.entries[self.len] = PathEntry { node, child_pos };
        self.len += 1;
    }
}

// ---------------------------------------------------------------------------
// BPlusIndex
// ---------------------------------------------------------------------------

/// An arena-backed B+Tree sorted multimap.
///
/// Entries are `(K, V)` pairs sorted lexicographically. The pair is the unique
/// identity: inserting the same `(K, V)` twice is a no-op, but inserting
/// `(K, V1)` and `(K, V2)` stores both.
///
/// Nodes are stored in contiguous `Vec`s and referenced by `u32` indices.
/// Freed nodes are recycled via an internal free list.
pub struct BPlusIndex<K, V> {
    leaves: Vec<LeafNode<K, V>>,
    inners: Vec<InnerNode<K, V>>,
    free_leaves: Vec<u32>,
    free_inners: Vec<u32>,
    root: u32,
    height: u32,
    len: usize,
    first_leaf: u32,
}

impl<K: Ord + Clone, V: Ord + Clone> BPlusIndex<K, V> {
    /// Creates a new empty index.
    pub fn new() -> Self {
        let mut idx = Self {
            leaves: Vec::new(),
            inners: Vec::new(),
            free_leaves: Vec::new(),
            free_inners: Vec::new(),
            root: 0,
            height: 0,
            len: 0,
            first_leaf: 0,
        };
        idx.alloc_leaf(); // leaf 0 = empty root
        idx
    }

    /// Returns the number of entries in the index.
    pub fn len(&self) -> usize {
        self.len
    }

    /// Returns `true` if the index contains no entries.
    pub fn is_empty(&self) -> bool {
        self.len == 0
    }

    // -- Arena helpers ------------------------------------------------------

    fn alloc_leaf(&mut self) -> u32 {
        if let Some(id) = self.free_leaves.pop() {
            self.leaves[id as usize] = LeafNode::new();
            id
        } else {
            let id = self.leaves.len() as u32;
            self.leaves.push(LeafNode::new());
            id
        }
    }

    fn free_leaf(&mut self, id: u32) {
        self.free_leaves.push(id);
    }

    fn alloc_inner(&mut self) -> u32 {
        if let Some(id) = self.free_inners.pop() {
            self.inners[id as usize] = InnerNode::new();
            id
        } else {
            let id = self.inners.len() as u32;
            self.inners.push(InnerNode::new());
            id
        }
    }

    fn free_inner(&mut self, id: u32) {
        self.free_inners.push(id);
    }

    #[inline]
    fn leaf(&self, id: u32) -> &LeafNode<K, V> {
        &self.leaves[id as usize]
    }

    #[inline]
    fn leaf_mut(&mut self, id: u32) -> &mut LeafNode<K, V> {
        &mut self.leaves[id as usize]
    }

    #[inline]
    fn inner(&self, id: u32) -> &InnerNode<K, V> {
        &self.inners[id as usize]
    }

    #[inline]
    fn inner_mut(&mut self, id: u32) -> &mut InnerNode<K, V> {
        &mut self.inners[id as usize]
    }

    // -- Descent ------------------------------------------------------------

    /// Descend using full `(K, V)` comparison. Used by insert and delete to
    /// find the exact leaf and record the path for split/merge propagation.
    fn descend_kv(&self, k: &K, v: &V) -> (u32, Path) {
        let mut path = Path::new();
        if self.height == 0 {
            return (self.root, path);
        }
        let mut node = self.root;
        for level in (1..=self.height).rev() {
            let inner = self.inner(node);
            let child_pos = inner.find_child_kv(k, v);
            path.push(node, child_pos);
            node = inner.children[child_pos];
            if level == 1 {
                return (node, path);
            }
        }
        unreachable!()
    }

    /// Descend using K-only lower-bound. Used by lookups and range queries to
    /// find the leftmost leaf that could contain the key.
    fn descend_k(&self, k: &K) -> u32 {
        if self.height == 0 {
            return self.root;
        }
        let mut node = self.root;
        for level in (1..=self.height).rev() {
            let inner = self.inner(node);
            let child_pos = inner.find_child_k_lower(k);
            node = inner.children[child_pos];
            if level == 1 {
                return node;
            }
        }
        unreachable!()
    }

    // -- Insert -------------------------------------------------------------

    /// Inserts a `(key, val)` pair into the index.
    ///
    /// If the exact `(key, val)` pair already exists, this is a no-op.
    /// Multiple values for the same key are allowed.
    pub fn insert(&mut self, key: K, val: V) {
        let (leaf_id, path) = self.descend_kv(&key, &val);
        let leaf = self.leaf(leaf_id);

        let pos = match leaf.search_kv_pos(&key, &val) {
            Ok(_) => return, // duplicate (K, V)
            Err(p) => p,
        };

        if !self.leaf(leaf_id).is_full() {
            let leaf = self.leaf_mut(leaf_id);
            av_insert(&mut leaf.keys, pos, key);
            av_insert(&mut leaf.vals, pos, val);
            self.len += 1;
            return;
        }

        // Leaf is full: split then propagate upward.
        let (right_id, sep_k, sep_v) = self.split_leaf(leaf_id, pos, key, val);
        self.propagate_split(path, sep_k, sep_v, right_id);
        self.len += 1;
    }

    /// Split a full leaf at the midpoint, inserting `(key, val)` at `pos`.
    /// Returns the new right leaf id and the separator to promote.
    fn split_leaf(&mut self, leaf_id: u32, pos: usize, key: K, val: V) -> (u32, K, V) {
        let split = LEAF_CAP / 2;

        let mut right_keys = av_split_off(&mut self.leaf_mut(leaf_id).keys, split);
        let mut right_vals = av_split_off(&mut self.leaf_mut(leaf_id).vals, split);

        if pos <= split {
            let leaf = self.leaf_mut(leaf_id);
            av_insert(&mut leaf.keys, pos, key);
            av_insert(&mut leaf.vals, pos, val);
        } else {
            av_insert(&mut right_keys, pos - split, key);
            av_insert(&mut right_vals, pos - split, val);
        }

        let sep_k = right_keys[0].clone();
        let sep_v = right_vals[0].clone();

        let right_id = self.alloc_leaf();
        let old_next = self.leaf(leaf_id).next;

        let right = self.leaf_mut(right_id);
        right.keys = right_keys;
        right.vals = right_vals;
        right.prev = leaf_id;
        right.next = old_next;

        self.leaf_mut(leaf_id).next = right_id;

        if old_next != NONE {
            self.leaf_mut(old_next).prev = right_id;
        }

        (right_id, sep_k, sep_v)
    }

    /// Propagate a split upward through inner nodes. If the root overflows,
    /// a new root is created (increasing tree height by 1).
    fn propagate_split(&mut self, path: Path, mut sep_k: K, mut sep_v: V, mut right_child: u32) {
        for i in (0..path.len).rev() {
            let inner_id = path.entries[i].node;
            let pos = self.inner(inner_id).find_child_kv(&sep_k, &sep_v);

            if !self.inner(inner_id).is_full() {
                let inner = self.inner_mut(inner_id);
                av_insert(&mut inner.keys, pos, sep_k);
                av_insert(&mut inner.vals, pos, sep_v);
                av_insert(&mut inner.children, pos + 1, right_child);
                return;
            }

            let (new_right, med_k, med_v) =
                self.split_inner(inner_id, pos, sep_k, sep_v, right_child);
            sep_k = med_k;
            sep_v = med_v;
            right_child = new_right;
        }

        self.grow_root(sep_k, sep_v, right_child);
    }

    /// Split a full inner node, returning the new right node and the median
    /// separator to promote to the parent.
    fn split_inner(
        &mut self,
        inner_id: u32,
        pos: usize,
        key: K,
        val: V,
        child: u32,
    ) -> (u32, K, V) {
        let split = INNER_CAP / 2;

        let mut right_keys = av_split_off(&mut self.inner_mut(inner_id).keys, split + 1);
        let mut right_vals = av_split_off(&mut self.inner_mut(inner_id).vals, split + 1);
        let mut right_children = av_split_off(&mut self.inner_mut(inner_id).children, split + 1);

        let med_k = self.inner_mut(inner_id).keys.remove(split);
        let med_v = self.inner_mut(inner_id).vals.remove(split);

        if pos <= split {
            let inner = self.inner_mut(inner_id);
            av_insert(&mut inner.keys, pos, key);
            av_insert(&mut inner.vals, pos, val);
            av_insert(&mut inner.children, pos + 1, child);
        } else {
            let rp = pos - split - 1;
            av_insert(&mut right_keys, rp, key);
            av_insert(&mut right_vals, rp, val);
            av_insert(&mut right_children, rp + 1, child);
        }

        let right_id = self.alloc_inner();
        let right = self.inner_mut(right_id);
        right.keys = right_keys;
        right.vals = right_vals;
        right.children = right_children;

        (right_id, med_k, med_v)
    }

    /// Create a new root containing one separator and two children.
    fn grow_root(&mut self, sep_k: K, sep_v: V, right_child: u32) {
        let old_root = self.root;
        let new_root = self.alloc_inner();
        let root = self.inner_mut(new_root);
        root.keys.push(sep_k);
        root.vals.push(sep_v);
        root.children.push(old_root);
        root.children.push(right_child);
        self.root = new_root;
        self.height += 1;
    }

    // -- Delete -------------------------------------------------------------

    /// Removes the exact `(key, val)` pair from the index.
    ///
    /// Returns `true` if the entry was found and removed, `false` otherwise.
    pub fn delete(&mut self, key: &K, val: &V) -> bool {
        let (leaf_id, path) = self.descend_kv(key, val);

        let pos = match self.leaf(leaf_id).search_kv_pos(key, val) {
            Ok(p) => p,
            Err(_) => return false,
        };

        self.leaf_mut(leaf_id).keys.remove(pos);
        self.leaf_mut(leaf_id).vals.remove(pos);
        self.len -= 1;

        // Root leaf never needs rebalancing.
        if self.height == 0 {
            return true;
        }

        if self.leaf(leaf_id).len() >= MIN_LEAF {
            return true;
        }

        self.rebalance_leaf(leaf_id, &path);
        true
    }

    /// Rebalance a leaf that has fallen below minimum occupancy.
    ///
    /// Tries in order: borrow from right sibling, borrow from left sibling,
    /// merge with a sibling. Merges may cascade up through inner nodes.
    fn rebalance_leaf(&mut self, leaf_id: u32, path: &Path) {
        let parent_entry = path.entries[path.len - 1];
        let parent_id = parent_entry.node;
        let child_pos = parent_entry.child_pos;

        // Try borrow from right sibling.
        if child_pos + 1 < self.inner(parent_id).children.len() {
            let right_id = self.inner(parent_id).children[child_pos + 1];
            if self.leaf(right_id).can_lend() {
                let rk = self.leaf_mut(right_id).keys.remove(0);
                let rv = self.leaf_mut(right_id).vals.remove(0);
                self.leaf_mut(leaf_id).keys.push(rk);
                self.leaf_mut(leaf_id).vals.push(rv);

                let new_sep_k = self.leaf(right_id).keys[0].clone();
                let new_sep_v = self.leaf(right_id).vals[0].clone();
                self.inner_mut(parent_id).keys[child_pos] = new_sep_k;
                self.inner_mut(parent_id).vals[child_pos] = new_sep_v;
                return;
            }
        }

        // Try borrow from left sibling.
        if child_pos > 0 {
            let left_id = self.inner(parent_id).children[child_pos - 1];
            if self.leaf(left_id).can_lend() {
                let lk = self.leaf_mut(left_id).keys.pop().unwrap();
                let lv = self.leaf_mut(left_id).vals.pop().unwrap();
                av_insert(&mut self.leaf_mut(leaf_id).keys, 0, lk);
                av_insert(&mut self.leaf_mut(leaf_id).vals, 0, lv);

                let new_sep_k = self.leaf(leaf_id).keys[0].clone();
                let new_sep_v = self.leaf(leaf_id).vals[0].clone();
                self.inner_mut(parent_id).keys[child_pos - 1] = new_sep_k;
                self.inner_mut(parent_id).vals[child_pos - 1] = new_sep_v;
                return;
            }
        }

        // Merge with a sibling.
        if child_pos + 1 < self.inner(parent_id).children.len() {
            let right_id = self.inner(parent_id).children[child_pos + 1];
            self.merge_leaves(leaf_id, right_id);
            self.inner_mut(parent_id).keys.remove(child_pos);
            self.inner_mut(parent_id).vals.remove(child_pos);
            self.inner_mut(parent_id).children.remove(child_pos + 1);
            self.free_leaf(right_id);
        } else {
            let left_id = self.inner(parent_id).children[child_pos - 1];
            self.merge_leaves(left_id, leaf_id);
            self.inner_mut(parent_id).keys.remove(child_pos - 1);
            self.inner_mut(parent_id).vals.remove(child_pos - 1);
            self.inner_mut(parent_id).children.remove(child_pos);
            self.free_leaf(leaf_id);
        }

        self.maybe_shrink_or_rebalance_inner(parent_id, path, path.len - 1);
    }

    /// Merge the right leaf into the left leaf and fix the linked list.
    fn merge_leaves(&mut self, left_id: u32, right_id: u32) {
        let right = self.leaf_mut(right_id);
        let mut rk = std::mem::replace(&mut right.keys, ArrayVec::new());
        let mut rv = std::mem::replace(&mut right.vals, ArrayVec::new());
        let right_next = right.next;

        let left = self.leaf_mut(left_id);
        av_append(&mut left.keys, &mut rk);
        av_append(&mut left.vals, &mut rv);
        left.next = right_next;

        if right_next != NONE {
            self.leaf_mut(right_next).prev = left_id;
        }
    }

    /// After a merge, check if the parent needs shrinking (root with one child)
    /// or rebalancing (inner node below minimum occupancy).
    fn maybe_shrink_or_rebalance_inner(&mut self, inner_id: u32, path: &Path, path_idx: usize) {
        // Root with zero separators -> shrink the tree by one level.
        if inner_id == self.root {
            if self.inner(inner_id).keys.is_empty() {
                let only_child = self.inner(inner_id).children[0];
                self.free_inner(inner_id);
                self.root = only_child;
                self.height -= 1;
                if self.height == 0 {
                    self.first_leaf = self.root;
                }
            }
            return;
        }

        if self.inner(inner_id).len() >= MIN_INNER {
            return;
        }

        self.rebalance_inner(inner_id, path, path_idx);
    }

    /// Rebalance an inner node: borrow from a sibling or merge.
    fn rebalance_inner(&mut self, inner_id: u32, path: &Path, path_idx: usize) {
        if path_idx == 0 {
            return;
        }

        let parent_entry = path.entries[path_idx - 1];
        let parent_id = parent_entry.node;
        let child_pos = parent_entry.child_pos;

        // Try borrow from right sibling.
        if child_pos + 1 < self.inner(parent_id).children.len() {
            let right_id = self.inner(parent_id).children[child_pos + 1];
            if self.inner(right_id).can_lend() {
                let sep_k = self.inner_mut(parent_id).keys.remove(child_pos);
                let sep_v = self.inner_mut(parent_id).vals.remove(child_pos);

                let rk = self.inner_mut(right_id).keys.remove(0);
                let rv = self.inner_mut(right_id).vals.remove(0);
                let rc = self.inner_mut(right_id).children.remove(0);

                av_insert(&mut self.inner_mut(parent_id).keys, child_pos, rk);
                av_insert(&mut self.inner_mut(parent_id).vals, child_pos, rv);

                self.inner_mut(inner_id).keys.push(sep_k);
                self.inner_mut(inner_id).vals.push(sep_v);
                self.inner_mut(inner_id).children.push(rc);
                return;
            }
        }

        // Try borrow from left sibling.
        if child_pos > 0 {
            let left_id = self.inner(parent_id).children[child_pos - 1];
            if self.inner(left_id).can_lend() {
                let sep_k = self.inner_mut(parent_id).keys.remove(child_pos - 1);
                let sep_v = self.inner_mut(parent_id).vals.remove(child_pos - 1);

                let lk = self.inner_mut(left_id).keys.pop().unwrap();
                let lv = self.inner_mut(left_id).vals.pop().unwrap();
                let lc = self.inner_mut(left_id).children.pop().unwrap();

                av_insert(&mut self.inner_mut(parent_id).keys, child_pos - 1, lk);
                av_insert(&mut self.inner_mut(parent_id).vals, child_pos - 1, lv);

                av_insert(&mut self.inner_mut(inner_id).keys, 0, sep_k);
                av_insert(&mut self.inner_mut(inner_id).vals, 0, sep_v);
                av_insert(&mut self.inner_mut(inner_id).children, 0, lc);
                return;
            }
        }

        // Merge with right sibling.
        if child_pos + 1 < self.inner(parent_id).children.len() {
            let right_id = self.inner(parent_id).children[child_pos + 1];

            let sep_k = self.inner_mut(parent_id).keys.remove(child_pos);
            let sep_v = self.inner_mut(parent_id).vals.remove(child_pos);
            self.inner_mut(parent_id).children.remove(child_pos + 1);

            self.inner_mut(inner_id).keys.push(sep_k);
            self.inner_mut(inner_id).vals.push(sep_v);

            let right = self.inner_mut(right_id);
            let mut rk = std::mem::replace(&mut right.keys, ArrayVec::new());
            let mut rv = std::mem::replace(&mut right.vals, ArrayVec::new());
            let mut rc = std::mem::replace(&mut right.children, ArrayVec::new());

            av_append(&mut self.inner_mut(inner_id).keys, &mut rk);
            av_append(&mut self.inner_mut(inner_id).vals, &mut rv);
            av_append(&mut self.inner_mut(inner_id).children, &mut rc);

            self.free_inner(right_id);
        } else {
            // Merge into left sibling.
            let left_id = self.inner(parent_id).children[child_pos - 1];

            let sep_k = self.inner_mut(parent_id).keys.remove(child_pos - 1);
            let sep_v = self.inner_mut(parent_id).vals.remove(child_pos - 1);
            self.inner_mut(parent_id).children.remove(child_pos);

            self.inner_mut(left_id).keys.push(sep_k);
            self.inner_mut(left_id).vals.push(sep_v);

            let cur = self.inner_mut(inner_id);
            let mut ck = std::mem::replace(&mut cur.keys, ArrayVec::new());
            let mut cv = std::mem::replace(&mut cur.vals, ArrayVec::new());
            let mut cc = std::mem::replace(&mut cur.children, ArrayVec::new());

            av_append(&mut self.inner_mut(left_id).keys, &mut ck);
            av_append(&mut self.inner_mut(left_id).vals, &mut cv);
            av_append(&mut self.inner_mut(left_id).children, &mut cc);

            self.free_inner(inner_id);
        }

        self.maybe_shrink_or_rebalance_inner(parent_id, path, path_idx - 1);
    }

    // -- Update -------------------------------------------------------------

    /// Re-indexes an entry by deleting `(old_key, val)` and inserting `(new_key, val)`.
    ///
    /// This is a convenience for the common case where the indexed field of a
    /// row changes but the row identifier (the value) stays the same.
    pub fn update(&mut self, old_key: &K, new_key: K, val: V) {
        self.delete(old_key, &val);
        self.insert(new_key, val);
    }

    // -- Lookups ------------------------------------------------------------

    /// Returns a reference to the first value associated with `key`, or `None`.
    ///
    /// For unique indexes this is the only value. For non-unique indexes this
    /// returns the smallest `V` for the given `K`.
    ///
    /// Time: O(log n).
    pub fn lookup_unique(&self, key: &K) -> Option<&V> {
        let leaf_id = self.descend_k(key);
        let leaf = self.leaf(leaf_id);
        let pos = leaf.lower_bound_k(key);
        if pos < leaf.len() && leaf.keys[pos] == *key {
            return Some(&leaf.vals[pos]);
        }
        // Key might be at the start of the next leaf (boundary case).
        if leaf.next != NONE {
            let next = self.leaf(leaf.next);
            if !next.keys.is_empty() && next.keys[0] == *key {
                return Some(&next.vals[0]);
            }
        }
        None
    }

    /// Returns an iterator over all values associated with `key`.
    ///
    /// Values are yielded in ascending `V` order. The iterator follows the
    /// leaf chain, so it works correctly even when entries for one key span
    /// multiple leaves.
    ///
    /// Time: O(log n) for the initial descent, then O(k) for k results.
    pub fn lookup_range<'a>(&'a self, key: &'a K) -> LookupRange<'a, K, V> {
        let leaf_id = self.descend_k(key);
        let leaf = self.leaf(leaf_id);
        let pos = leaf.lower_bound_k(key);
        LookupRange {
            tree: self,
            key,
            leaf: leaf_id,
            pos,
        }
    }

    /// Returns an iterator over all entries `(&K, &V)` whose keys fall within
    /// the given range bounds.
    ///
    /// Supports all standard range syntax: `a..b`, `a..=b`, `a..`, `..b`, `..`.
    ///
    /// Time: O(log n) for the initial descent, then O(k) for k results.
    pub fn range<R: RangeBounds<K>>(&self, bounds: R) -> RangeIter<'_, K, V> {
        let (leaf, pos) = match bounds.start_bound() {
            Bound::Unbounded => (self.first_leaf, 0),
            Bound::Included(k) => {
                let lid = self.descend_k(k);
                let l = self.leaf(lid);
                (lid, l.lower_bound_k(k))
            }
            Bound::Excluded(k) => {
                let lid = self.descend_k(k);
                let l = self.leaf(lid);
                (lid, l.upper_bound_k(k))
            }
        };

        let end = match bounds.end_bound() {
            Bound::Unbounded => RangeEnd::Unbounded,
            Bound::Included(k) => RangeEnd::Included(k.clone()),
            Bound::Excluded(k) => RangeEnd::Excluded(k.clone()),
        };

        RangeIter {
            tree: self,
            leaf,
            pos,
            end,
        }
    }

    /// Returns an iterator over all entries in ascending order.
    ///
    /// Time: O(n).
    pub fn iter(&self) -> RangeIter<'_, K, V> {
        RangeIter {
            tree: self,
            leaf: self.first_leaf,
            pos: 0,
            end: RangeEnd::Unbounded,
        }
    }
}

impl<K: Ord + Clone, V: Ord + Clone> Default for BPlusIndex<K, V> {
    fn default() -> Self {
        Self::new()
    }
}

// ---------------------------------------------------------------------------
// Iterators
// ---------------------------------------------------------------------------

/// Iterator returned by [`BPlusIndex::lookup_range`].
///
/// Yields `&V` for all entries matching a specific key.
pub struct LookupRange<'a, K, V> {
    tree: &'a BPlusIndex<K, V>,
    key: &'a K,
    leaf: u32,
    pos: usize,
}

impl<'a, K: Ord + Clone, V: Ord + Clone> Iterator for LookupRange<'a, K, V> {
    type Item = &'a V;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            if self.leaf == NONE {
                return None;
            }
            let leaf = self.tree.leaf(self.leaf);
            if self.pos < leaf.len() {
                if leaf.keys[self.pos] == *self.key {
                    let v = &leaf.vals[self.pos];
                    self.pos += 1;
                    return Some(v);
                }
                return None;
            }
            self.leaf = leaf.next;
            self.pos = 0;
        }
    }
}

enum RangeEnd<K> {
    Unbounded,
    Included(K),
    Excluded(K),
}

/// Iterator returned by [`BPlusIndex::range`] and [`BPlusIndex::iter`].
///
/// Yields `(&K, &V)` pairs in ascending order. Follows the leaf chain
/// for sequential memory access.
pub struct RangeIter<'a, K, V> {
    tree: &'a BPlusIndex<K, V>,
    leaf: u32,
    pos: usize,
    end: RangeEnd<K>,
}

impl<'a, K: Ord + Clone, V: Ord + Clone> Iterator for RangeIter<'a, K, V> {
    type Item = (&'a K, &'a V);

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            if self.leaf == NONE {
                return None;
            }
            let leaf = self.tree.leaf(self.leaf);
            if self.pos < leaf.len() {
                let k = &leaf.keys[self.pos];
                match &self.end {
                    RangeEnd::Excluded(end) if k >= end => return None,
                    RangeEnd::Included(end) if k > end => return None,
                    _ => {}
                }
                let v = &leaf.vals[self.pos];
                self.pos += 1;
                return Some((k, v));
            }
            self.leaf = leaf.next;
            self.pos = 0;
        }
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn insert_and_lookup_unique() {
        let mut idx = BPlusIndex::<i32, u64>::new();
        for i in 0..200 {
            idx.insert(i, i as u64 * 10);
        }
        assert_eq!(idx.len(), 200);
        for i in 0..200 {
            assert_eq!(idx.lookup_unique(&i), Some(&(i as u64 * 10)));
        }
        assert_eq!(idx.lookup_unique(&999), None);
    }

    #[test]
    fn insert_duplicates_ignored() {
        let mut idx = BPlusIndex::<i32, u64>::new();
        idx.insert(5, 100);
        idx.insert(5, 100);
        assert_eq!(idx.len(), 1);
    }

    #[test]
    fn non_unique_keys() {
        let mut idx = BPlusIndex::<i32, u64>::new();
        idx.insert(5, 100);
        idx.insert(5, 200);
        idx.insert(5, 300);
        idx.insert(10, 400);

        let vals: Vec<_> = idx.lookup_range(&5).copied().collect();
        assert_eq!(vals, vec![100, 200, 300]);
    }

    #[test]
    fn delete_and_rebalance() {
        let mut idx = BPlusIndex::<i32, u64>::new();
        let n = 500;
        for i in 0..n {
            idx.insert(i, i as u64);
        }
        for i in 0..n {
            assert!(idx.delete(&i, &(i as u64)));
        }
        assert_eq!(idx.len(), 0);
        assert!(idx.is_empty());
    }

    #[test]
    fn delete_nonexistent() {
        let mut idx = BPlusIndex::<i32, u64>::new();
        idx.insert(1, 10);
        assert!(!idx.delete(&1, &999));
        assert!(!idx.delete(&999, &10));
        assert_eq!(idx.len(), 1);
    }

    #[test]
    fn range_query() {
        let mut idx = BPlusIndex::<i32, u64>::new();
        for i in 0..100 {
            idx.insert(i, i as u64);
        }
        let vals: Vec<_> = idx.range(10..20).map(|(k, _)| *k).collect();
        assert_eq!(vals, (10..20).collect::<Vec<_>>());
    }

    #[test]
    fn range_inclusive() {
        let mut idx = BPlusIndex::<i32, u64>::new();
        for i in 0..10 {
            idx.insert(i, i as u64);
        }
        let vals: Vec<_> = idx.range(3..=7).map(|(k, _)| *k).collect();
        assert_eq!(vals, vec![3, 4, 5, 6, 7]);
    }

    #[test]
    fn iter_full() {
        let mut idx = BPlusIndex::<i32, u64>::new();
        for i in (0..300).rev() {
            idx.insert(i, i as u64);
        }
        let keys: Vec<_> = idx.iter().map(|(k, _)| *k).collect();
        assert_eq!(keys, (0..300).collect::<Vec<_>>());
    }

    #[test]
    fn update_moves_entry() {
        let mut idx = BPlusIndex::<i32, u64>::new();
        idx.insert(5, 42);
        idx.update(&5, 10, 42);
        assert_eq!(idx.lookup_unique(&5), None);
        assert_eq!(idx.lookup_unique(&10), Some(&42));
    }

    #[test]
    fn stress_insert_delete_interleaved() {
        let mut idx = BPlusIndex::<i32, u64>::new();
        for i in 0..1000 {
            idx.insert(i, i as u64);
        }
        for i in (0..1000).step_by(2) {
            assert!(idx.delete(&i, &(i as u64)));
        }
        assert_eq!(idx.len(), 500);
        for i in (1..1000).step_by(2) {
            assert_eq!(idx.lookup_unique(&i), Some(&(i as u64)));
        }
        for i in (0..1000).step_by(2) {
            idx.insert(i, i as u64);
        }
        assert_eq!(idx.len(), 1000);
        let keys: Vec<_> = idx.iter().map(|(k, _)| *k).collect();
        assert_eq!(keys, (0..1000).collect::<Vec<_>>());
    }

    #[test]
    fn many_duplicates_same_key() {
        let mut idx = BPlusIndex::<i32, u64>::new();
        for v in 0..200u64 {
            idx.insert(42, v);
        }
        assert_eq!(idx.len(), 200);
        let vals: Vec<_> = idx.lookup_range(&42).copied().collect();
        assert_eq!(vals, (0..200u64).collect::<Vec<_>>());

        for v in (0..200u64).step_by(2) {
            assert!(idx.delete(&42, &v));
        }
        assert_eq!(idx.len(), 100);
        let vals: Vec<_> = idx.lookup_range(&42).copied().collect();
        assert_eq!(
            vals,
            (0..200u64).step_by(2).map(|v| v + 1).collect::<Vec<_>>()
        );
    }

    #[test]
    fn reverse_insertion_order() {
        let mut idx = BPlusIndex::<i32, u64>::new();
        for i in (0..500).rev() {
            idx.insert(i, i as u64);
        }
        let keys: Vec<_> = idx.iter().map(|(k, _)| *k).collect();
        assert_eq!(keys, (0..500).collect::<Vec<_>>());
    }
}