shodh_redb/

multimap_table.rs

use crate::cdc::types::{CdcEvent, ChangeOp};
use crate::compat::{HashMap, Mutex};
#[cfg(feature = "std")]
use crate::db::CorruptPageInfo;
use crate::db::TransactionGuard;
use crate::multimap_table::DynamicCollectionType::{Inline, SubtreeV2};
use crate::sealed::Sealed;
use crate::table::{ReadableTableMetadata, TableStats};
use crate::tree_store::{
    AllPageNumbersBtreeIter, BRANCH, BranchAccessor, BranchMutator, Btree, BtreeHeader, BtreeMut,
    BtreeRangeIter, BtreeStats, Checksum, DEFERRED, LEAF, LeafAccessor, LeafMutator,
    MAX_PAIR_LENGTH, MAX_VALUE_LENGTH, Page, PageHint, PageNumber, PagePath, PageTrackerPolicy,
    RawBtree, RawLeafBuilder, TransactionalMemory, UntypedBtree, UntypedBtreeMut, btree_stats,
};
use crate::types::{Key, TypeName, Value};
use crate::{AccessGuard, MultimapTableHandle, Result, StorageError, WriteTransaction};
use alloc::format;
use alloc::string::{String, ToString};
use alloc::sync::Arc;
use alloc::vec;
use alloc::vec::Vec;
use core::borrow::Borrow;
use core::cmp::max;
use core::convert::TryInto;
use core::marker::PhantomData;
use core::mem;
use core::mem::size_of;
use core::ops::{RangeBounds, RangeFull};

pub(crate) fn multimap_btree_stats(
    root: Option<PageNumber>,
    mem: &TransactionalMemory,
    fixed_key_size: Option<usize>,
    fixed_value_size: Option<usize>,
) -> Result<BtreeStats> {
    if let Some(root) = root {
        multimap_stats_helper(root, mem, fixed_key_size, fixed_value_size)
    } else {
        Ok(BtreeStats {
            tree_height: 0,
            leaf_pages: 0,
            branch_pages: 0,
            stored_leaf_bytes: 0,
            metadata_bytes: 0,
            fragmented_bytes: 0,
        })
    }
}

fn multimap_stats_helper(
    page_number: PageNumber,
    mem: &TransactionalMemory,
    fixed_key_size: Option<usize>,
    fixed_value_size: Option<usize>,
) -> Result<BtreeStats> {
    let page = mem.get_page(page_number)?;
    let node_mem = page.memory();
    match node_mem[0] {
        LEAF => {
            let accessor = LeafAccessor::new(
                page.memory(),
                fixed_key_size,
                DynamicCollection::<()>::fixed_width_with(fixed_value_size),
            )?;
            let mut leaf_bytes = 0u64;
            let mut is_branch = false;
            for i in 0..accessor.num_pairs() {
                let entry = accessor
                    .entry(i)
                    .ok_or_else(|| StorageError::invalid_entry_index(page_number, i))?;
                let collection: &UntypedDynamicCollection =
                    UntypedDynamicCollection::new(entry.value());
                match collection.collection_type()? {
                    Inline => {
                        let inline_accessor = LeafAccessor::new(
                            collection.as_inline(),
                            fixed_value_size,
                            <() as Value>::fixed_width(),
                        )?;
                        leaf_bytes +=
                            inline_accessor.length_of_pairs(0, inline_accessor.num_pairs()) as u64;
                    }
                    SubtreeV2 => {
                        is_branch = true;
                    }
                }
            }
            let mut overhead_bytes = (accessor.total_length() as u64) - leaf_bytes;
            let mut fragmented_bytes = (page.memory().len() - accessor.total_length()) as u64;
            let mut max_child_height = 0;
            let (mut leaf_pages, mut branch_pages) = if is_branch { (0, 1) } else { (1, 0) };

            for i in 0..accessor.num_pairs() {
                let entry = accessor
                    .entry(i)
                    .ok_or_else(|| StorageError::invalid_entry_index(page_number, i))?;
                let collection: &UntypedDynamicCollection =
                    UntypedDynamicCollection::new(entry.value());
                match collection.collection_type()? {
                    Inline => {
                        // data is inline, so it was already counted above
                    }
                    SubtreeV2 => {
                        // this is a sub-tree, so traverse it
                        let stats = btree_stats(
                            Some(collection.as_subtree().root),
                            mem,
                            fixed_value_size,
                            <() as Value>::fixed_width(),
                        )?;
                        max_child_height = max(max_child_height, stats.tree_height);
                        branch_pages += stats.branch_pages;
                        leaf_pages += stats.leaf_pages;
                        fragmented_bytes += stats.fragmented_bytes;
                        overhead_bytes += stats.metadata_bytes;
                        leaf_bytes += stats.stored_leaf_bytes;
                    }
                }
            }

            Ok(BtreeStats {
                tree_height: max_child_height + 1,
                leaf_pages,
                branch_pages,
                stored_leaf_bytes: leaf_bytes,
                metadata_bytes: overhead_bytes,
                fragmented_bytes,
            })
        }
        BRANCH => {
            let accessor = BranchAccessor::new(&page, fixed_key_size)?;
            let mut max_child_height = 0;
            let mut leaf_pages = 0;
            let mut branch_pages = 1;
            let mut stored_leaf_bytes = 0;
            let mut metadata_bytes = accessor.total_length() as u64;
            let mut fragmented_bytes = (page.memory().len() - accessor.total_length()) as u64;
            for i in 0..accessor.count_children() {
                if let Some(child) = accessor.child_page(i) {
                    let stats =
                        multimap_stats_helper(child, mem, fixed_key_size, fixed_value_size)?;
                    max_child_height = max(max_child_height, stats.tree_height);
                    leaf_pages += stats.leaf_pages;
                    branch_pages += stats.branch_pages;
                    stored_leaf_bytes += stats.stored_leaf_bytes;
                    metadata_bytes += stats.metadata_bytes;
                    fragmented_bytes += stats.fragmented_bytes;
                }
            }

            Ok(BtreeStats {
                tree_height: max_child_height + 1,
                leaf_pages,
                branch_pages,
                stored_leaf_bytes,
                metadata_bytes,
                fragmented_bytes,
            })
        }
        other => Err(StorageError::invalid_page_type(page_number, other)),
    }
}

// Verify all the checksums in the tree, including any Dynamic collection subtrees
pub(crate) fn verify_tree_and_subtree_checksums(
    root: Option<BtreeHeader>,
    key_size: Option<usize>,
    value_size: Option<usize>,
    mem: Arc<TransactionalMemory>,
) -> Result<bool> {
    if let Some(header) = root {
        if !RawBtree::new(
            Some(header),
            key_size,
            DynamicCollection::<()>::fixed_width_with(value_size),
            mem.clone(),
        )
        .verify_checksum()?
        {
            return Ok(false);
        }

        let table_pages_iter = AllPageNumbersBtreeIter::new(
            header.root,
            key_size,
            DynamicCollection::<()>::fixed_width_with(value_size),
            mem.clone(),
        )?;
        for table_page in table_pages_iter {
            let page = mem.get_page(table_page?)?;
            let subtree_roots = parse_subtree_roots(&page, key_size, value_size)?;
            for header in subtree_roots {
                if !RawBtree::new(Some(header), value_size, <()>::fixed_width(), mem.clone())
                    .verify_checksum()?
                {
                    return Ok(false);
                }
            }
        }
    }

    Ok(true)
}

/// Like `verify_tree_and_subtree_checksums` but collects corruption details.
#[cfg(feature = "std")]
pub(crate) fn verify_tree_and_subtree_checksums_detailed(
    root: Option<BtreeHeader>,
    key_size: Option<usize>,
    value_size: Option<usize>,
    mem: Arc<TransactionalMemory>,
) -> Result<(u64, Vec<CorruptPageInfo>)> {
    let mut total_pages = 0u64;
    let mut all_corruptions = Vec::new();

    if let Some(header) = root {
        let tree = RawBtree::new(
            Some(header),
            key_size,
            DynamicCollection::<()>::fixed_width_with(value_size),
            mem.clone(),
        );
        let (pages, corruptions) = tree.verify_checksum_detailed()?;
        total_pages += pages;
        all_corruptions.extend(corruptions);

        let table_pages_iter = AllPageNumbersBtreeIter::new(
            header.root,
            key_size,
            DynamicCollection::<()>::fixed_width_with(value_size),
            mem.clone(),
        )?;
        for table_page in table_pages_iter {
            let page = mem.get_page(table_page?)?;
            let subtree_roots = parse_subtree_roots(&page, key_size, value_size)?;
            for sub_header in subtree_roots {
                let sub_tree = RawBtree::new(
                    Some(sub_header),
                    value_size,
                    <()>::fixed_width(),
                    mem.clone(),
                );
                let (pages, corruptions) = sub_tree.verify_checksum_detailed()?;
                total_pages += pages;
                all_corruptions.extend(corruptions);
            }
        }
    }

    Ok((total_pages, all_corruptions))
}

/// Like `verify_tree_and_subtree_checksums` but verifies structural integrity.
#[cfg(feature = "std")]
pub(crate) fn verify_tree_and_subtree_structure(
    root: Option<BtreeHeader>,
    key_size: Option<usize>,
    value_size: Option<usize>,
    mem: Arc<TransactionalMemory>,
) -> Result<Vec<CorruptPageInfo>> {
    let mut all_corruptions = Vec::new();

    if let Some(header) = root {
        let tree = RawBtree::new(
            Some(header),
            key_size,
            DynamicCollection::<()>::fixed_width_with(value_size),
            mem.clone(),
        );
        all_corruptions.extend(tree.verify_structure()?);

        let table_pages_iter = AllPageNumbersBtreeIter::new(
            header.root,
            key_size,
            DynamicCollection::<()>::fixed_width_with(value_size),
            mem.clone(),
        )?;
        for table_page in table_pages_iter {
            let page = mem.get_page(table_page?)?;
            let subtree_roots = parse_subtree_roots(&page, key_size, value_size)?;
            for sub_header in subtree_roots {
                let sub_tree = RawBtree::new(
                    Some(sub_header),
                    value_size,
                    <()>::fixed_width(),
                    mem.clone(),
                );
                all_corruptions.extend(sub_tree.verify_structure()?);
            }
        }
    }

    Ok(all_corruptions)
}

// Relocate all subtrees to lower index pages, if possible
pub(crate) fn relocate_subtrees(
    root: (PageNumber, Checksum),
    key_size: Option<usize>,
    value_size: Option<usize>,
    mem: Arc<TransactionalMemory>,
    freed_pages: Arc<Mutex<Vec<PageNumber>>>,
    relocation_map: &HashMap<PageNumber, PageNumber>,
) -> Result<(PageNumber, Checksum)> {
    let old_page = mem.get_page(root.0)?;
    let mut new_page = if let Some(new_page_number) = relocation_map.get(&root.0) {
        mem.get_page_mut(*new_page_number)?
    } else {
        return Ok(root);
    };
    let new_page_number = new_page.get_page_number();
    new_page.memory_mut()?.copy_from_slice(old_page.memory());

    match old_page.memory()[0] {
        LEAF => {
            let accessor = LeafAccessor::new(
                old_page.memory(),
                key_size,
                UntypedDynamicCollection::fixed_width_with(value_size),
            )?;
            // TODO: maybe there's a better abstraction, so that we don't need to call into this low-level method?
            let mut mutator = LeafMutator::new(
                new_page.memory_mut()?,
                key_size,
                UntypedDynamicCollection::fixed_width_with(value_size),
            );
            for i in 0..accessor.num_pairs() {
                let entry = accessor
                    .entry(i)
                    .ok_or_else(|| StorageError::invalid_entry_index(root.0, i))?;
                let collection = UntypedDynamicCollection::from_bytes(entry.value());
                if matches!(collection.collection_type()?, SubtreeV2) {
                    let sub_root = collection.as_subtree();
                    let mut tree = UntypedBtreeMut::new(
                        Some(sub_root),
                        mem.clone(),
                        freed_pages.clone(),
                        value_size,
                        <() as Value>::fixed_width(),
                    );
                    tree.relocate(relocation_map)?;
                    let new_root = tree.get_root().ok_or_else(|| {
                        StorageError::Internal(
                            "missing subtree root after relocate in relocate_subtrees".to_string(),
                        )
                    })?;
                    if sub_root != new_root {
                        let new_collection = UntypedDynamicCollection::make_subtree_data(new_root);
                        mutator.insert(i, true, entry.key(), &new_collection);
                    }
                }
            }
        }
        BRANCH => {
            let accessor = BranchAccessor::new(&old_page, key_size)?;
            let mut mutator = BranchMutator::new(new_page.memory_mut()?);
            for i in 0..accessor.count_children() {
                if let Some(child) = accessor.child_page(i) {
                    let child_checksum = accessor.child_checksum(i).unwrap();
                    let (new_child, new_checksum) = relocate_subtrees(
                        (child, child_checksum),
                        key_size,
                        value_size,
                        mem.clone(),
                        freed_pages.clone(),
                        relocation_map,
                    )?;
                    mutator.write_child_page(i, new_child, new_checksum);
                }
            }
        }
        _ => {
            return Err(StorageError::invalid_page_type(
                root.0,
                old_page.memory()[0],
            ));
        }
    }

    let old_page_number = old_page.get_page_number();
    drop(old_page);
    // No need to track allocations, because this method is only called during compaction when
    // there can't be any savepoints
    let mut ignore = PageTrackerPolicy::Ignore;
    if !mem.free_if_uncommitted(old_page_number, &mut ignore) {
        freed_pages.lock().push(old_page_number);
    }
    Ok((new_page_number, DEFERRED))
}

// Finalize all the checksums in the tree, including any Dynamic collection subtrees
// Returns the root checksum
pub(crate) fn finalize_tree_and_subtree_checksums(
    root: Option<BtreeHeader>,
    key_size: Option<usize>,
    value_size: Option<usize>,
    mem: Arc<TransactionalMemory>,
) -> Result<Option<BtreeHeader>> {
    let freed_pages = Arc::new(Mutex::new(vec![]));
    let mut tree = UntypedBtreeMut::new(
        root,
        mem.clone(),
        freed_pages.clone(),
        key_size,
        DynamicCollection::<()>::fixed_width_with(value_size),
    );
    tree.dirty_leaf_visitor(|mut leaf_page| {
        let mut sub_root_updates = vec![];
        let accessor = LeafAccessor::new(
            leaf_page.memory(),
            key_size,
            DynamicCollection::<()>::fixed_width_with(value_size),
        )?;
        for i in 0..accessor.num_pairs() {
            let entry = accessor
                .entry(i)
                .ok_or_else(|| StorageError::invalid_entry_index(leaf_page.get_page_number(), i))?;
            let collection = <&DynamicCollection<()>>::from_bytes(entry.value());
            if matches!(collection.collection_type()?, SubtreeV2) {
                let sub_root = collection.as_subtree();
                if mem.uncommitted(sub_root.root) {
                    let mut subtree = UntypedBtreeMut::new(
                        Some(sub_root),
                        mem.clone(),
                        freed_pages.clone(),
                        value_size,
                        <()>::fixed_width(),
                    );
                    let subtree_root = subtree.finalize_dirty_checksums()?.unwrap();
                    sub_root_updates.push((i, entry.key().to_vec(), subtree_root));
                }
            }
        }
        // TODO: maybe there's a better abstraction, so that we don't need to call into this low-level method?
        let mut mutator = LeafMutator::new(
            leaf_page.memory_mut()?,
            key_size,
            DynamicCollection::<()>::fixed_width_with(value_size),
        );
        for (i, key, sub_root) in sub_root_updates {
            let collection = DynamicCollection::<()>::make_subtree_data(sub_root);
            mutator.insert(i, true, &key, &collection);
        }

        Ok(())
    })?;

    let root = tree.finalize_dirty_checksums()?;
    // No pages should have been freed by this operation
    assert!(freed_pages.lock().is_empty());
    Ok(root)
}

fn parse_subtree_roots<T: Page>(
    page: &T,
    fixed_key_size: Option<usize>,
    fixed_value_size: Option<usize>,
) -> crate::Result<Vec<BtreeHeader>> {
    match page.memory()[0] {
        BRANCH => Ok(vec![]),
        LEAF => {
            let mut result = vec![];
            let accessor = LeafAccessor::new(
                page.memory(),
                fixed_key_size,
                DynamicCollection::<()>::fixed_width_with(fixed_value_size),
            )?;
            for i in 0..accessor.num_pairs() {
                let entry = accessor
                    .entry(i)
                    .ok_or_else(|| StorageError::invalid_entry_index(page.get_page_number(), i))?;
                let collection = <&DynamicCollection<()>>::from_bytes(entry.value());
                if matches!(collection.collection_type()?, SubtreeV2) {
                    result.push(collection.as_subtree());
                }
            }

            Ok(result)
        }
        _ => Err(StorageError::invalid_page_type(
            page.get_page_number(),
            page.memory()[0],
        )),
    }
}

pub(crate) struct UntypedMultiBtree {
    mem: Arc<TransactionalMemory>,
    root: Option<BtreeHeader>,
    key_width: Option<usize>,
    value_width: Option<usize>,
}

impl UntypedMultiBtree {
    pub(crate) fn new(
        root: Option<BtreeHeader>,
        mem: Arc<TransactionalMemory>,
        key_width: Option<usize>,
        value_width: Option<usize>,
    ) -> Self {
        Self {
            mem,
            root,
            key_width,
            value_width,
        }
    }

    // Applies visitor to pages in the tree
    pub(crate) fn visit_all_pages<F>(&self, mut visitor: F) -> Result
    where
        F: FnMut(&PagePath) -> Result,
    {
        let tree = UntypedBtree::new(
            self.root,
            self.mem.clone(),
            self.key_width,
            UntypedDynamicCollection::fixed_width_with(self.value_width),
        );
        tree.visit_all_pages(|path| {
            visitor(path)?;
            let page = self.mem.get_page(path.page_number())?;
            match page.memory()[0] {
                LEAF => {
                    for header in parse_subtree_roots(&page, self.key_width, self.value_width)? {
                        let subtree = UntypedBtree::new(
                            Some(header),
                            self.mem.clone(),
                            self.value_width,
                            <() as Value>::fixed_width(),
                        );
                        subtree.visit_all_pages(|subpath| {
                            let full_path = path.with_subpath(subpath);
                            visitor(&full_path)
                        })?;
                    }
                }
                BRANCH => {
                    // No-op. The tree.visit_pages() call will process this sub-tree
                }
                other => {
                    return Err(StorageError::invalid_page_type(path.page_number(), other));
                }
            }
            Ok(())
        })?;

        Ok(())
    }
}

pub(crate) struct LeafKeyIter<'a, V: Key + 'static> {
    inline_collection: AccessGuard<'a, &'static DynamicCollection<V>>,
    fixed_key_size: Option<usize>,
    fixed_value_size: Option<usize>,
    start_entry: isize, // inclusive
    end_entry: isize,   // inclusive
}

impl<'a, V: Key> LeafKeyIter<'a, V> {
    fn new(
        data: AccessGuard<'a, &'static DynamicCollection<V>>,
        fixed_key_size: Option<usize>,
        fixed_value_size: Option<usize>,
    ) -> Result<Self> {
        let accessor =
            LeafAccessor::new(data.value().as_inline(), fixed_key_size, fixed_value_size)?;
        let num_pairs = accessor.num_pairs();
        let end_entry = if num_pairs == 0 {
            -1
        } else {
            isize::try_from(num_pairs).unwrap_or(isize::MAX) - 1
        };
        Ok(Self {
            inline_collection: data,
            fixed_key_size,
            fixed_value_size,
            start_entry: 0,
            end_entry,
        })
    }

    fn next_key(&mut self) -> Option<Result<&[u8]>> {
        if self.end_entry < self.start_entry {
            return None;
        }
        let accessor = match LeafAccessor::new(
            self.inline_collection.value().as_inline(),
            self.fixed_key_size,
            self.fixed_value_size,
        ) {
            Ok(a) => a,
            Err(e) => return Some(Err(e)),
        };
        self.start_entry += 1;
        accessor
            .entry((self.start_entry - 1).try_into().unwrap())
            .map(|e| Ok(e.key()))
    }

    fn next_key_back(&mut self) -> Option<Result<&[u8]>> {
        if self.end_entry < self.start_entry {
            return None;
        }
        let accessor = match LeafAccessor::new(
            self.inline_collection.value().as_inline(),
            self.fixed_key_size,
            self.fixed_value_size,
        ) {
            Ok(a) => a,
            Err(e) => return Some(Err(e)),
        };
        self.end_entry -= 1;
        accessor
            .entry((self.end_entry + 1).try_into().unwrap())
            .map(|e| Ok(e.key()))
    }
}

enum DynamicCollectionType {
    Inline,
    // Was used in file format version 1
    // Subtree,
    SubtreeV2,
}

impl DynamicCollectionType {
    fn try_from_byte(value: u8) -> crate::Result<Self> {
        match value {
            LEAF => Ok(Inline),
            // 2 => Subtree,
            3 => Ok(SubtreeV2),
            _ => Err(StorageError::Corrupted(format!(
                "invalid DynamicCollectionType byte: {value}"
            ))),
        }
    }
}

#[allow(clippy::from_over_into)]
impl Into<u8> for DynamicCollectionType {
    fn into(self) -> u8 {
        match self {
            // Reuse the LEAF type id, so that we can cast this directly into the format used by
            // LeafAccessor
            Inline => LEAF,
            // Subtree => 2,
            SubtreeV2 => 3,
        }
    }
}

/// Layout:
/// type (1 byte):
/// * 1 = inline data
/// * 2 = sub tree
///
/// (when type = 1) data (n bytes): inlined leaf node
///
/// (when type = 2) root (8 bytes): sub tree root page number
/// (when type = 2) checksum (16 bytes): sub tree checksum
///
/// NOTE: Even though the [`PhantomData`] is zero-sized, the inner data DST must be placed last.
/// See [Exotically Sized Types](https://doc.rust-lang.org/nomicon/exotic-sizes.html#dynamically-sized-types-dsts)
/// section of the Rustonomicon for more details.
#[repr(transparent)]
struct DynamicCollection<V: Key> {
    _value_type: PhantomData<V>,
    data: [u8],
}

impl<V: Key> core::fmt::Debug for DynamicCollection<V> {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("DynamicCollection")
            .field("data", &&self.data)
            .finish()
    }
}

impl<V: Key> Value for &DynamicCollection<V> {
    type SelfType<'a>
        = &'a DynamicCollection<V>
    where
        Self: 'a;
    type AsBytes<'a>
        = &'a [u8]
    where
        Self: 'a;

    fn fixed_width() -> Option<usize> {
        None
    }

    fn from_bytes<'a>(data: &'a [u8]) -> &'a DynamicCollection<V>
    where
        Self: 'a,
    {
        DynamicCollection::new(data)
    }

    fn as_bytes<'a, 'b: 'a>(value: &'a Self::SelfType<'b>) -> &'a [u8]
    where
        Self: 'b,
    {
        &value.data
    }

    fn type_name() -> TypeName {
        TypeName::internal("redb::DynamicCollection")
    }
}

impl<V: Key> DynamicCollection<V> {
    fn new(data: &[u8]) -> &Self {
        // SAFETY: DynamicCollection<V> is #[repr(transparent)] over [u8] (with a
        // zero-sized PhantomData<V>), so it has identical size, alignment, and
        // memory layout. The pointer cast preserves the slice length metadata and
        // the shared borrow lifetime is propagated to the returned reference.
        unsafe { &*(core::ptr::from_ref::<[u8]>(data) as *const DynamicCollection<V>) }
    }

    fn collection_type(&self) -> crate::Result<DynamicCollectionType> {
        DynamicCollectionType::try_from_byte(self.data[0])
    }

    fn as_inline(&self) -> &[u8] {
        &self.data[1..]
    }

    fn as_subtree(&self) -> BtreeHeader {
        BtreeHeader::from_le_bytes(
            self.data[1..=BtreeHeader::serialized_size()]
                .try_into()
                .unwrap(),
        )
    }

    fn get_num_values(&self) -> crate::Result<u64> {
        Ok(match self.collection_type()? {
            Inline => {
                let leaf_data = self.as_inline();
                let accessor =
                    LeafAccessor::new(leaf_data, V::fixed_width(), <() as Value>::fixed_width())?;
                accessor.num_pairs() as u64
            }
            SubtreeV2 => {
                let offset = 1 + PageNumber::serialized_size() + size_of::<Checksum>();
                u64::from_le_bytes(
                    self.data[offset..(offset + size_of::<u64>())]
                        .try_into()
                        .unwrap(),
                )
            }
        })
    }

    fn make_inline_data(data: &[u8]) -> Vec<u8> {
        let mut result = vec![Inline.into()];
        result.extend_from_slice(data);

        result
    }

    fn make_subtree_data(header: BtreeHeader) -> Vec<u8> {
        let mut result = vec![SubtreeV2.into()];
        result.extend_from_slice(&header.to_le_bytes());
        result
    }

    pub(crate) fn fixed_width_with(_value_width: Option<usize>) -> Option<usize> {
        None
    }
}

impl<V: Key> DynamicCollection<V> {
    fn iter<'a>(
        collection: AccessGuard<'a, &'static DynamicCollection<V>>,
        guard: Arc<TransactionGuard>,
        mem: Arc<TransactionalMemory>,
    ) -> Result<MultimapValue<'a, V>> {
        Ok(match collection.value().collection_type()? {
            Inline => {
                let leaf_iter =
                    LeafKeyIter::new(collection, V::fixed_width(), <() as Value>::fixed_width())?;
                MultimapValue::new_inline(leaf_iter, guard)?
            }
            SubtreeV2 => {
                let root = collection.value().as_subtree().root;
                MultimapValue::new_subtree(
                    BtreeRangeIter::new::<RangeFull, &V::SelfType<'_>>(
                        &(..),
                        Some(root),
                        None,
                        if mem.compression().is_enabled() {
                            None
                        } else {
                            <() as Value>::fixed_width()
                        },
                        mem,
                        false,
                    )?,
                    collection.value().get_num_values()?,
                    guard,
                )
            }
        })
    }

    fn iter_free_on_drop<'a>(
        collection: AccessGuard<'a, &'static DynamicCollection<V>>,
        pages: Vec<PageNumber>,
        freed_pages: Arc<Mutex<Vec<PageNumber>>>,
        allocated_pages: Arc<Mutex<PageTrackerPolicy>>,
        guard: Arc<TransactionGuard>,
        mem: Arc<TransactionalMemory>,
    ) -> Result<MultimapValue<'a, V>> {
        let num_values = collection.value().get_num_values()?;
        Ok(match collection.value().collection_type()? {
            Inline => {
                let leaf_iter =
                    LeafKeyIter::new(collection, V::fixed_width(), <() as Value>::fixed_width())?;
                MultimapValue::new_inline(leaf_iter, guard)?
            }
            SubtreeV2 => {
                let root = collection.value().as_subtree().root;
                let inner = BtreeRangeIter::new::<RangeFull, &V::SelfType<'_>>(
                    &(..),
                    Some(root),
                    None,
                    if mem.compression().is_enabled() {
                        None
                    } else {
                        <() as Value>::fixed_width()
                    },
                    mem.clone(),
                    false,
                )?;
                MultimapValue::new_subtree_free_on_drop(
                    inner,
                    num_values,
                    freed_pages,
                    allocated_pages,
                    pages,
                    guard,
                    mem,
                )
            }
        })
    }
}

#[repr(transparent)]
struct UntypedDynamicCollection {
    data: [u8],
}

impl UntypedDynamicCollection {
    pub(crate) fn fixed_width_with(_value_width: Option<usize>) -> Option<usize> {
        None
    }

    fn new(data: &[u8]) -> &Self {
        // SAFETY: UntypedDynamicCollection is #[repr(transparent)] over [u8],
        // so it has identical size, alignment, and memory layout. The pointer
        // cast preserves the slice length metadata and the shared borrow
        // lifetime is propagated to the returned reference.
        unsafe { &*(core::ptr::from_ref::<[u8]>(data) as *const UntypedDynamicCollection) }
    }

    fn make_subtree_data(header: BtreeHeader) -> Vec<u8> {
        let mut result = vec![SubtreeV2.into()];
        result.extend_from_slice(&header.to_le_bytes());
        result
    }

    fn from_bytes(data: &[u8]) -> &Self {
        Self::new(data)
    }

    fn collection_type(&self) -> crate::Result<DynamicCollectionType> {
        DynamicCollectionType::try_from_byte(self.data[0])
    }

    fn as_inline(&self) -> &[u8] {
        &self.data[1..]
    }

    fn as_subtree(&self) -> BtreeHeader {
        BtreeHeader::from_le_bytes(
            self.data[1..=BtreeHeader::serialized_size()]
                .try_into()
                .unwrap(),
        )
    }
}

enum ValueIterState<'a, V: Key + 'static> {
    Subtree(BtreeRangeIter<V, ()>),
    InlineLeaf(LeafKeyIter<'a, V>),
}

pub struct MultimapValue<'a, V: Key + 'static> {
    inner: Option<ValueIterState<'a, V>>,
    remaining: u64,
    freed_pages: Option<Arc<Mutex<Vec<PageNumber>>>>,
    allocated_pages: Arc<Mutex<PageTrackerPolicy>>,
    free_on_drop: Vec<PageNumber>,
    _transaction_guard: Arc<TransactionGuard>,
    mem: Option<Arc<TransactionalMemory>>,
    _value_type: PhantomData<V>,
}

impl<'a, V: Key + 'static> MultimapValue<'a, V> {
    fn new_subtree(
        inner: BtreeRangeIter<V, ()>,
        num_values: u64,
        guard: Arc<TransactionGuard>,
    ) -> Self {
        Self {
            inner: Some(ValueIterState::Subtree(inner)),
            remaining: num_values,
            freed_pages: None,
            allocated_pages: Arc::new(Mutex::new(PageTrackerPolicy::Closed)),
            free_on_drop: vec![],
            _transaction_guard: guard,
            mem: None,
            _value_type: Default::default(),
        }
    }

    fn new_subtree_free_on_drop(
        inner: BtreeRangeIter<V, ()>,
        num_values: u64,
        freed_pages: Arc<Mutex<Vec<PageNumber>>>,
        allocated_pages: Arc<Mutex<PageTrackerPolicy>>,
        pages: Vec<PageNumber>,
        guard: Arc<TransactionGuard>,
        mem: Arc<TransactionalMemory>,
    ) -> Self {
        Self {
            inner: Some(ValueIterState::Subtree(inner)),
            remaining: num_values,
            freed_pages: Some(freed_pages),
            free_on_drop: pages,
            allocated_pages,
            _transaction_guard: guard,
            mem: Some(mem),
            _value_type: Default::default(),
        }
    }

    fn new_inline(inner: LeafKeyIter<'a, V>, guard: Arc<TransactionGuard>) -> Result<Self> {
        let remaining = inner.inline_collection.value().get_num_values()?;
        Ok(Self {
            inner: Some(ValueIterState::InlineLeaf(inner)),
            remaining,
            freed_pages: None,
            allocated_pages: Arc::new(Mutex::new(PageTrackerPolicy::Closed)),
            free_on_drop: vec![],
            _transaction_guard: guard,
            mem: None,
            _value_type: Default::default(),
        })
    }

    /// Returns the number of times this iterator will return `Some(Ok(_))`
    ///
    /// Note that `Some` may be returned from `next()` more than `len()` times if `Some(Err(_))` is returned
    pub fn len(&self) -> u64 {
        self.remaining
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

impl<'a, V: Key + 'static> Iterator for MultimapValue<'a, V> {
    type Item = Result<AccessGuard<'a, V>>;

    fn next(&mut self) -> Option<Self::Item> {
        let bytes = match self.inner.as_mut()? {
            ValueIterState::Subtree(iter) => match iter.next()? {
                Ok(e) => e.key_data(),
                Err(err) => {
                    return Some(Err(err));
                }
            },
            ValueIterState::InlineLeaf(iter) => match iter.next_key()? {
                Ok(bytes) => bytes.to_vec(),
                Err(err) => return Some(Err(err)),
            },
        };
        self.remaining -= 1;
        Some(Ok(AccessGuard::with_owned_value(bytes)))
    }
}

impl<V: Key + 'static> DoubleEndedIterator for MultimapValue<'_, V> {
    fn next_back(&mut self) -> Option<Self::Item> {
        let bytes = match self.inner.as_mut()? {
            ValueIterState::Subtree(iter) => match iter.next_back()? {
                Ok(e) => e.key_data(),
                Err(err) => {
                    return Some(Err(err));
                }
            },
            ValueIterState::InlineLeaf(iter) => match iter.next_key_back()? {
                Ok(bytes) => bytes.to_vec(),
                Err(err) => return Some(Err(err)),
            },
        };
        self.remaining -= 1;
        Some(Ok(AccessGuard::with_owned_value(bytes)))
    }
}

impl<V: Key + 'static> Drop for MultimapValue<'_, V> {
    fn drop(&mut self) {
        // Drop our references to the pages that are about to be freed
        drop(mem::take(&mut self.inner));
        if !self.free_on_drop.is_empty()
            && let (Some(freed_pages_arc), Some(mem_arc)) =
                (self.freed_pages.as_ref(), self.mem.as_ref())
        {
            let mut freed_pages = freed_pages_arc.lock();
            let mut allocated_pages = self.allocated_pages.lock();
            for page in &self.free_on_drop {
                if !mem_arc.free_if_uncommitted(*page, &mut allocated_pages) {
                    freed_pages.push(*page);
                }
            }
        }
    }
}

pub struct MultimapRange<'a, K: Key + 'static, V: Key + 'static> {
    inner: BtreeRangeIter<K, &'static DynamicCollection<V>>,
    mem: Arc<TransactionalMemory>,
    transaction_guard: Arc<TransactionGuard>,
    _key_type: PhantomData<K>,
    _value_type: PhantomData<V>,
    _lifetime: PhantomData<&'a ()>,
}

impl<K: Key + 'static, V: Key + 'static> MultimapRange<'_, K, V> {
    fn new(
        inner: BtreeRangeIter<K, &'static DynamicCollection<V>>,
        guard: Arc<TransactionGuard>,
        mem: Arc<TransactionalMemory>,
    ) -> Self {
        Self {
            inner,
            mem,
            transaction_guard: guard,
            _key_type: Default::default(),
            _value_type: Default::default(),
            _lifetime: Default::default(),
        }
    }
}

impl<'a, K: Key + 'static, V: Key + 'static> Iterator for MultimapRange<'a, K, V> {
    type Item = Result<(AccessGuard<'a, K>, MultimapValue<'a, V>)>;

    fn next(&mut self) -> Option<Self::Item> {
        match self.inner.next()? {
            Ok(entry) => {
                let key = AccessGuard::with_owned_value(entry.key_data());
                let (page, _, value_range, decompressed_value) = entry.into_raw();
                let collection = if let Some(bytes) = decompressed_value {
                    AccessGuard::with_owned_value(bytes)
                } else {
                    AccessGuard::with_page(page, value_range)
                };
                Some(
                    DynamicCollection::iter(
                        collection,
                        self.transaction_guard.clone(),
                        self.mem.clone(),
                    )
                    .map(|iter| (key, iter)),
                )
            }
            Err(err) => Some(Err(err)),
        }
    }
}

impl<K: Key + 'static, V: Key + 'static> DoubleEndedIterator for MultimapRange<'_, K, V> {
    fn next_back(&mut self) -> Option<Self::Item> {
        match self.inner.next_back()? {
            Ok(entry) => {
                let key = AccessGuard::with_owned_value(entry.key_data());
                let (page, _, value_range, decompressed_value) = entry.into_raw();
                let collection = if let Some(bytes) = decompressed_value {
                    AccessGuard::with_owned_value(bytes)
                } else {
                    AccessGuard::with_page(page, value_range)
                };
                Some(
                    DynamicCollection::iter(
                        collection,
                        self.transaction_guard.clone(),
                        self.mem.clone(),
                    )
                    .map(|iter| (key, iter)),
                )
            }
            Err(err) => Some(Err(err)),
        }
    }
}

/// A multimap table
///
/// [Multimap tables](https://en.wikipedia.org/wiki/Multimap) may have multiple values associated with each key
pub struct MultimapTable<'txn, K: Key + 'static, V: Key + 'static> {
    name: String,
    num_values: u64,
    transaction: &'txn WriteTransaction,
    freed_pages: Arc<Mutex<Vec<PageNumber>>>,
    allocated_pages: Arc<Mutex<PageTrackerPolicy>>,
    tree: BtreeMut<'txn, K, &'static DynamicCollection<V>>,
    mem: Arc<TransactionalMemory>,
    _value_type: PhantomData<V>,
}

impl<K: Key + 'static, V: Key + 'static> MultimapTableHandle for MultimapTable<'_, K, V> {
    fn name(&self) -> &str {
        &self.name
    }
}

impl<'txn, K: Key + 'static, V: Key + 'static> MultimapTable<'txn, K, V> {
    pub(crate) fn new(
        name: &str,
        table_root: Option<BtreeHeader>,
        num_values: u64,
        freed_pages: Arc<Mutex<Vec<PageNumber>>>,
        allocated_pages: Arc<Mutex<PageTrackerPolicy>>,
        mem: Arc<TransactionalMemory>,
        transaction: &'txn WriteTransaction,
    ) -> MultimapTable<'txn, K, V> {
        MultimapTable {
            name: name.to_string(),
            num_values,
            transaction,
            freed_pages: freed_pages.clone(),
            allocated_pages: allocated_pages.clone(),
            tree: BtreeMut::new(
                table_root,
                transaction.transaction_guard(),
                mem.clone(),
                freed_pages,
                allocated_pages,
            ),
            mem,
            _value_type: Default::default(),
        }
    }

    #[allow(dead_code)]
    #[cfg(feature = "std")]
    pub(crate) fn print_debug(&self, include_values: bool) -> Result {
        self.tree.print_debug(include_values)
    }

    /// Add the given value to the mapping of the key
    ///
    /// Returns `true` if the key-value pair was present
    pub fn insert<'k, 'v>(
        &mut self,
        key: impl Borrow<K::SelfType<'k>>,
        value: impl Borrow<V::SelfType<'v>>,
    ) -> Result<bool> {
        let value_bytes = V::as_bytes(value.borrow());
        let value_bytes_ref = value_bytes.as_ref();
        if value_bytes_ref.len() > MAX_VALUE_LENGTH {
            return Err(StorageError::ValueTooLarge(value_bytes_ref.len()));
        }
        let key_len = K::as_bytes(key.borrow()).as_ref().len();
        if key_len > MAX_VALUE_LENGTH {
            return Err(StorageError::ValueTooLarge(key_len));
        }
        if value_bytes_ref.len() + key_len > MAX_PAIR_LENGTH {
            return Err(StorageError::ValueTooLarge(value_bytes_ref.len() + key_len));
        }
        let get_result = self.tree.get(key.borrow())?;
        let existed = if get_result.is_some() {
            #[allow(clippy::unnecessary_unwrap)]
            let guard = get_result.unwrap();
            let collection_type = guard.value().collection_type()?;
            match collection_type {
                Inline => {
                    let leaf_data = guard.value().as_inline();
                    let accessor = LeafAccessor::new(
                        leaf_data,
                        V::fixed_width(),
                        <() as Value>::fixed_width(),
                    )?;
                    let (position, found) = accessor.position::<V>(value_bytes_ref);
                    if found {
                        return Ok(true);
                    }

                    let num_pairs = accessor.num_pairs();
                    let new_pairs = num_pairs + 1;
                    let new_pair_bytes =
                        accessor.length_of_pairs(0, accessor.num_pairs()) + value_bytes_ref.len();
                    let new_key_bytes =
                        accessor.length_of_keys(0, accessor.num_pairs()) + value_bytes_ref.len();
                    let required_inline_bytes = RawLeafBuilder::required_bytes(
                        new_pairs,
                        new_pair_bytes,
                        V::fixed_width(),
                        <() as Value>::fixed_width(),
                    );

                    if required_inline_bytes < self.mem.get_page_size() / 2 {
                        let mut data = vec![0; required_inline_bytes];
                        let mut builder = RawLeafBuilder::new(
                            &mut data,
                            new_pairs,
                            V::fixed_width(),
                            <() as Value>::fixed_width(),
                            new_key_bytes,
                        );
                        for i in 0..accessor.num_pairs() {
                            if i == position {
                                builder
                                    .append(value_bytes_ref, <() as Value>::as_bytes(&()).as_ref());
                            }
                            let entry = accessor.entry(i).ok_or_else(|| {
                                StorageError::Corrupted(format!(
                                    "invalid entry index {i} in multimap insert inline expansion"
                                ))
                            })?;
                            builder.append(entry.key(), entry.value());
                        }
                        if position == accessor.num_pairs() {
                            builder.append(value_bytes_ref, <() as Value>::as_bytes(&()).as_ref());
                        }
                        drop(builder);
                        drop(guard);
                        let inline_data = DynamicCollection::<V>::make_inline_data(&data);
                        self.tree
                            .insert(key.borrow(), &DynamicCollection::new(&inline_data))?;
                    } else {
                        // convert into a subtree
                        let mut allocated = self.allocated_pages.lock();
                        let mut page = self.mem.allocate(leaf_data.len(), &mut allocated)?;
                        drop(allocated);
                        page.memory_mut()?[..leaf_data.len()].copy_from_slice(leaf_data);
                        let page_number = page.get_page_number();
                        drop(page);
                        drop(guard);

                        // Don't bother computing the checksum, since we're about to modify the tree
                        let mut subtree: BtreeMut<'_, V, ()> = BtreeMut::new(
                            Some(BtreeHeader::new(page_number, 0, num_pairs as u64)),
                            self.transaction.transaction_guard(),
                            self.mem.clone(),
                            self.freed_pages.clone(),
                            self.allocated_pages.clone(),
                        );
                        let existed = subtree.insert(value.borrow(), &())?.is_some();
                        assert_eq!(existed, found);
                        let subtree_root = subtree.get_root().ok_or_else(|| {
                            StorageError::Internal(
                                "missing subtree root after insert in inline-to-subtree conversion"
                                    .to_string(),
                            )
                        })?;
                        let subtree_data = DynamicCollection::<V>::make_subtree_data(subtree_root);
                        self.tree
                            .insert(key.borrow(), &DynamicCollection::new(&subtree_data))?;
                    }

                    found
                }
                SubtreeV2 => {
                    let mut subtree: BtreeMut<'_, V, ()> = BtreeMut::new(
                        Some(guard.value().as_subtree()),
                        self.transaction.transaction_guard(),
                        self.mem.clone(),
                        self.freed_pages.clone(),
                        self.allocated_pages.clone(),
                    );
                    drop(guard);
                    let existed = subtree.insert(value.borrow(), &())?.is_some();
                    let subtree_root = subtree.get_root().ok_or_else(|| {
                        StorageError::Internal(
                            "missing subtree root after insert into existing subtree".to_string(),
                        )
                    })?;
                    let subtree_data = DynamicCollection::<V>::make_subtree_data(subtree_root);
                    self.tree
                        .insert(key.borrow(), &DynamicCollection::new(&subtree_data))?;

                    existed
                }
            }
        } else {
            drop(get_result);
            let required_inline_bytes = RawLeafBuilder::required_bytes(
                1,
                value_bytes_ref.len(),
                V::fixed_width(),
                <() as Value>::fixed_width(),
            );
            if required_inline_bytes < self.mem.get_page_size() / 2 {
                let mut data = vec![0; required_inline_bytes];
                let mut builder = RawLeafBuilder::new(
                    &mut data,
                    1,
                    V::fixed_width(),
                    <() as Value>::fixed_width(),
                    value_bytes_ref.len(),
                );
                builder.append(value_bytes_ref, <() as Value>::as_bytes(&()).as_ref());
                drop(builder);
                let inline_data = DynamicCollection::<V>::make_inline_data(&data);
                self.tree
                    .insert(key.borrow(), &DynamicCollection::new(&inline_data))?;
            } else {
                let mut subtree: BtreeMut<'_, V, ()> = BtreeMut::new(
                    None,
                    self.transaction.transaction_guard(),
                    self.mem.clone(),
                    self.freed_pages.clone(),
                    self.allocated_pages.clone(),
                );
                subtree.insert(value.borrow(), &())?;
                let subtree_root = subtree.get_root().ok_or_else(|| {
                    StorageError::Internal(
                        "missing subtree root after insert into new subtree".to_string(),
                    )
                })?;
                let subtree_data = DynamicCollection::<V>::make_subtree_data(subtree_root);
                self.tree
                    .insert(key.borrow(), &DynamicCollection::new(&subtree_data))?;
            }
            false
        };

        if !existed {
            self.num_values += 1;
            if self.transaction.cdc_log.is_some() {
                self.transaction.record_cdc(CdcEvent {
                    table_name: self.name.clone(),
                    op: ChangeOp::Insert,
                    key: K::as_bytes(key.borrow()).as_ref().to_vec(),
                    new_value: Some(V::as_bytes(value.borrow()).as_ref().to_vec()),
                    old_value: None,
                });
            }
        }

        Ok(existed)
    }

    /// Removes the given key-value pair
    ///
    /// Returns `true` if the key-value pair was present
    pub fn remove<'k, 'v>(
        &mut self,
        key: impl Borrow<K::SelfType<'k>>,
        value: impl Borrow<V::SelfType<'v>>,
    ) -> Result<bool> {
        let get_result = self.tree.get(key.borrow())?;
        if get_result.is_none() {
            return Ok(false);
        }
        let guard = get_result.unwrap();
        let v = guard.value();
        let existed = match v.collection_type()? {
            Inline => {
                let leaf_data = v.as_inline();
                let accessor =
                    LeafAccessor::new(leaf_data, V::fixed_width(), <() as Value>::fixed_width())?;
                if let Some(position) = accessor.find_key::<V>(V::as_bytes(value.borrow()).as_ref())
                {
                    let old_num_pairs = accessor.num_pairs();
                    if old_num_pairs == 1 {
                        drop(guard);
                        self.tree.remove(key.borrow())?;
                    } else {
                        let old_pairs_len = accessor.length_of_pairs(0, old_num_pairs);
                        let removed_value_len = accessor
                            .entry(position)
                            .ok_or_else(|| {
                                StorageError::Corrupted(format!(
                                    "invalid entry index {position} in multimap remove"
                                ))
                            })?
                            .key()
                            .len();
                        let required = RawLeafBuilder::required_bytes(
                            old_num_pairs - 1,
                            old_pairs_len - removed_value_len,
                            V::fixed_width(),
                            <() as Value>::fixed_width(),
                        );
                        let mut new_data = vec![0; required];
                        let new_key_len =
                            accessor.length_of_keys(0, old_num_pairs) - removed_value_len;
                        let mut builder = RawLeafBuilder::new(
                            &mut new_data,
                            old_num_pairs - 1,
                            V::fixed_width(),
                            <() as Value>::fixed_width(),
                            new_key_len,
                        );
                        for i in 0..old_num_pairs {
                            if i != position {
                                let entry = accessor.entry(i).ok_or_else(|| {
                                    StorageError::Corrupted(format!(
                                        "invalid entry index {i} in multimap remove rebuild"
                                    ))
                                })?;
                                builder.append(entry.key(), entry.value());
                            }
                        }
                        drop(builder);
                        drop(guard);

                        let inline_data = DynamicCollection::<V>::make_inline_data(&new_data);
                        self.tree
                            .insert(key.borrow(), &DynamicCollection::new(&inline_data))?;
                    }
                    true
                } else {
                    drop(guard);
                    false
                }
            }
            SubtreeV2 => {
                let mut subtree: BtreeMut<V, ()> = BtreeMut::new(
                    Some(v.as_subtree()),
                    self.transaction.transaction_guard(),
                    self.mem.clone(),
                    self.freed_pages.clone(),
                    self.allocated_pages.clone(),
                );
                drop(guard);
                let existed = subtree.remove(value.borrow())?.is_some();

                if let Some(BtreeHeader {
                    root: new_root,
                    checksum: new_checksum,
                    length: new_length,
                }) = subtree.get_root()
                {
                    let page = self.mem.get_page(new_root)?;
                    match page.memory()[0] {
                        LEAF => {
                            let accessor = LeafAccessor::new(
                                page.memory(),
                                V::fixed_width(),
                                <() as Value>::fixed_width(),
                            )?;
                            let len = accessor.total_length();
                            if len < self.mem.get_page_size() / 2 {
                                let inline_data =
                                    DynamicCollection::<V>::make_inline_data(&page.memory()[..len]);
                                self.tree
                                    .insert(key.borrow(), &DynamicCollection::new(&inline_data))?;
                                drop(page);
                                let mut allocated_pages = self.allocated_pages.lock();
                                if !self.mem.free_if_uncommitted(new_root, &mut allocated_pages) {
                                    (*self.freed_pages).lock().push(new_root);
                                }
                            } else {
                                let subtree_data =
                                    DynamicCollection::<V>::make_subtree_data(BtreeHeader::new(
                                        new_root,
                                        new_checksum,
                                        accessor.num_pairs() as u64,
                                    ));
                                self.tree
                                    .insert(key.borrow(), &DynamicCollection::new(&subtree_data))?;
                            }
                        }
                        BRANCH => {
                            let subtree_data = DynamicCollection::<V>::make_subtree_data(
                                BtreeHeader::new(new_root, new_checksum, new_length),
                            );
                            self.tree
                                .insert(key.borrow(), &DynamicCollection::new(&subtree_data))?;
                        }
                        other => {
                            return Err(StorageError::invalid_page_type(new_root, other));
                        }
                    }
                } else {
                    self.tree.remove(key.borrow())?;
                }

                existed
            }
        };

        if existed {
            self.num_values -= 1;
            if self.transaction.cdc_log.is_some() {
                self.transaction.record_cdc(CdcEvent {
                    table_name: self.name.clone(),
                    op: ChangeOp::Delete,
                    key: K::as_bytes(key.borrow()).as_ref().to_vec(),
                    new_value: None,
                    old_value: Some(V::as_bytes(value.borrow()).as_ref().to_vec()),
                });
            }
        }

        Ok(existed)
    }

    /// Removes all keys (and their associated values) in the given range
    ///
    /// Returns the number of keys removed
    pub fn drain<'a, KR>(&mut self, range: impl RangeBounds<KR> + 'a) -> Result<u64>
    where
        KR: Borrow<K::SelfType<'a>> + 'a,
    {
        // Collect keys first to avoid borrow conflict with remove_all
        let keys: Vec<Vec<u8>> = self
            .range(range)?
            .map(|result| result.map(|(k, _)| K::as_bytes(&k.value()).as_ref().to_vec()))
            .collect::<Result<Vec<_>>>()?;

        let mut count = 0u64;
        for key_bytes in &keys {
            let key = K::from_bytes(key_bytes);
            // Consume the returned iterator to trigger page cleanup
            for result in self.remove_all(&key)? {
                result?;
            }
            count += 1;
        }
        Ok(count)
    }

    /// Removes all keys (and their associated values) from the table
    ///
    /// Returns the number of keys removed
    pub fn drain_all(&mut self) -> Result<u64> {
        self.drain::<K::SelfType<'_>>(..)
    }

    /// Removes all values for the given key
    ///
    /// Returns an iterator over the removed values. Values are in ascending order.
    pub fn remove_all<'a>(
        &mut self,
        key: impl Borrow<K::SelfType<'a>>,
    ) -> Result<MultimapValue<'_, V>> {
        // Collect values for CDC before removal
        if self.transaction.cdc_log.is_some() {
            let key_bytes = K::as_bytes(key.borrow()).as_ref().to_vec();
            let guard = self.transaction.transaction_guard();
            if let Some(collection) = self.tree.get(key.borrow())? {
                let value_iter = DynamicCollection::iter(collection, guard, self.mem.clone())?;
                let mut cdc_events: Vec<CdcEvent> = Vec::new();
                for result in value_iter {
                    let val = result?;
                    cdc_events.push(CdcEvent {
                        table_name: self.name.clone(),
                        op: ChangeOp::Delete,
                        key: key_bytes.clone(),
                        new_value: None,
                        old_value: Some(V::as_bytes(&val.value()).as_ref().to_vec()),
                    });
                }
                for event in cdc_events {
                    self.transaction.record_cdc(event);
                }
            }
        }

        let iter = if let Some(collection) = self.tree.remove(key.borrow())? {
            let mut pages = vec![];
            if matches!(
                collection.value().collection_type()?,
                DynamicCollectionType::SubtreeV2
            ) {
                let root = collection.value().as_subtree().root;
                let all_pages = AllPageNumbersBtreeIter::new(
                    root,
                    V::fixed_width(),
                    <() as Value>::fixed_width(),
                    self.mem.clone(),
                )?;
                for page in all_pages {
                    pages.push(page?);
                }
            }

            self.num_values -= collection.value().get_num_values()?;

            DynamicCollection::iter_free_on_drop(
                collection,
                pages,
                self.freed_pages.clone(),
                self.allocated_pages.clone(),
                self.transaction.transaction_guard(),
                self.mem.clone(),
            )?
        } else {
            MultimapValue::new_subtree(
                BtreeRangeIter::new::<RangeFull, &V::SelfType<'_>>(
                    &(..),
                    None,
                    None,
                    if self.mem.compression().is_enabled() {
                        None
                    } else {
                        <() as Value>::fixed_width()
                    },
                    self.mem.clone(),
                    false,
                )?,
                0,
                self.transaction.transaction_guard(),
            )
        };

        Ok(iter)
    }
}

impl<K: Key + 'static, V: Key + 'static> ReadableTableMetadata for MultimapTable<'_, K, V> {
    fn stats(&self) -> Result<TableStats> {
        let tree_stats = multimap_btree_stats(
            self.tree.get_root().map(|x| x.root),
            &self.mem,
            K::fixed_width(),
            V::fixed_width(),
        )?;

        Ok(TableStats {
            tree_height: tree_stats.tree_height,
            leaf_pages: tree_stats.leaf_pages,
            branch_pages: tree_stats.branch_pages,
            stored_leaf_bytes: tree_stats.stored_leaf_bytes,
            metadata_bytes: tree_stats.metadata_bytes,
            fragmented_bytes: tree_stats.fragmented_bytes,
        })
    }

    /// Returns the number of key-value pairs in the table
    fn len(&self) -> Result<u64> {
        Ok(self.num_values)
    }
}

impl<K: Key + 'static, V: Key + 'static> ReadableMultimapTable<K, V> for MultimapTable<'_, K, V> {
    fn get<'a>(&self, key: impl Borrow<K::SelfType<'a>>) -> Result<MultimapValue<'_, V>> {
        let guard = self.transaction.transaction_guard();
        let iter = if let Some(collection) = self.tree.get(key.borrow())? {
            DynamicCollection::iter(collection, guard, self.mem.clone())?
        } else {
            MultimapValue::new_subtree(
                BtreeRangeIter::new::<RangeFull, &V::SelfType<'_>>(
                    &(..),
                    None,
                    None,
                    if self.mem.compression().is_enabled() {
                        None
                    } else {
                        <() as Value>::fixed_width()
                    },
                    self.mem.clone(),
                    false,
                )?,
                0,
                guard,
            )
        };

        Ok(iter)
    }

    fn range<'a, KR>(&self, range: impl RangeBounds<KR> + 'a) -> Result<MultimapRange<'_, K, V>>
    where
        KR: Borrow<K::SelfType<'a>> + 'a,
    {
        let inner = self.tree.range(&range)?;
        Ok(MultimapRange::new(
            inner,
            self.transaction.transaction_guard(),
            self.mem.clone(),
        ))
    }
}

impl<K: Key + 'static, V: Key + 'static> Sealed for MultimapTable<'_, K, V> {}

impl<K: Key + 'static, V: Key + 'static> Drop for MultimapTable<'_, K, V> {
    fn drop(&mut self) {
        self.transaction
            .close_table(&self.name, &self.tree, self.num_values);
    }
}

pub trait ReadableMultimapTable<K: Key + 'static, V: Key + 'static>: ReadableTableMetadata {
    /// Returns an iterator over all values for the given key. Values are in ascending order.
    fn get<'a>(&self, key: impl Borrow<K::SelfType<'a>>) -> Result<MultimapValue<'_, V>>;

    /// Returns a double-ended iterator over a range of elements in the table
    fn range<'a, KR>(&self, range: impl RangeBounds<KR> + 'a) -> Result<MultimapRange<'_, K, V>>
    where
        KR: Borrow<K::SelfType<'a>> + 'a;

    /// Returns an double-ended iterator over all elements in the table. Values are in ascending
    /// order.
    fn iter(&self) -> Result<MultimapRange<'_, K, V>> {
        self.range::<K::SelfType<'_>>(..)
    }
}

/// A read-only untyped multimap table
pub struct ReadOnlyUntypedMultimapTable {
    num_values: u64,
    tree: RawBtree,
    fixed_key_size: Option<usize>,
    fixed_value_size: Option<usize>,
    mem: Arc<TransactionalMemory>,
}

impl Sealed for ReadOnlyUntypedMultimapTable {}

impl ReadableTableMetadata for ReadOnlyUntypedMultimapTable {
    /// Retrieves information about storage usage for the table
    fn stats(&self) -> Result<TableStats> {
        let tree_stats = multimap_btree_stats(
            self.tree.get_root().map(|x| x.root),
            &self.mem,
            self.fixed_key_size,
            self.fixed_value_size,
        )?;

        Ok(TableStats {
            tree_height: tree_stats.tree_height,
            leaf_pages: tree_stats.leaf_pages,
            branch_pages: tree_stats.branch_pages,
            stored_leaf_bytes: tree_stats.stored_leaf_bytes,
            metadata_bytes: tree_stats.metadata_bytes,
            fragmented_bytes: tree_stats.fragmented_bytes,
        })
    }

    fn len(&self) -> Result<u64> {
        Ok(self.num_values)
    }
}

impl ReadOnlyUntypedMultimapTable {
    pub(crate) fn new(
        root: Option<BtreeHeader>,
        num_values: u64,
        fixed_key_size: Option<usize>,
        fixed_value_size: Option<usize>,
        mem: Arc<TransactionalMemory>,
    ) -> Self {
        Self {
            num_values,
            tree: RawBtree::new(
                root,
                fixed_key_size,
                DynamicCollection::<()>::fixed_width_with(fixed_value_size),
                mem.clone(),
            ),
            fixed_key_size,
            fixed_value_size,
            mem,
        }
    }
}

/// A read-only multimap table
pub struct ReadOnlyMultimapTable<K: Key + 'static, V: Key + 'static> {
    tree: Btree<K, &'static DynamicCollection<V>>,
    num_values: u64,
    mem: Arc<TransactionalMemory>,
    transaction_guard: Arc<TransactionGuard>,
    _value_type: PhantomData<V>,
}

impl<K: Key + 'static, V: Key + 'static> ReadOnlyMultimapTable<K, V> {
    pub(crate) fn new(
        root: Option<BtreeHeader>,
        num_values: u64,
        hint: PageHint,
        guard: Arc<TransactionGuard>,
        mem: Arc<TransactionalMemory>,
    ) -> Result<ReadOnlyMultimapTable<K, V>> {
        Ok(ReadOnlyMultimapTable {
            tree: Btree::new(root, hint, guard.clone(), mem.clone())?,
            num_values,
            mem,
            transaction_guard: guard,
            _value_type: Default::default(),
        })
    }

    /// This method is like [`ReadableMultimapTable::get()`], but the iterator is reference counted and keeps the transaction
    /// alive until it is dropped.
    pub fn get<'a>(&self, key: impl Borrow<K::SelfType<'a>>) -> Result<MultimapValue<'static, V>> {
        let iter = if let Some(collection) = self.tree.get(key.borrow())? {
            DynamicCollection::iter(collection, self.transaction_guard.clone(), self.mem.clone())?
        } else {
            MultimapValue::new_subtree(
                BtreeRangeIter::new::<RangeFull, &V::SelfType<'_>>(
                    &(..),
                    None,
                    None,
                    if self.mem.compression().is_enabled() {
                        None
                    } else {
                        <() as Value>::fixed_width()
                    },
                    self.mem.clone(),
                    false,
                )?,
                0,
                self.transaction_guard.clone(),
            )
        };

        Ok(iter)
    }

    /// This method is like [`ReadableMultimapTable::range()`], but the iterator is reference counted and keeps the transaction
    /// alive until it is dropped.
    pub fn range<'a, KR>(&self, range: impl RangeBounds<KR>) -> Result<MultimapRange<'static, K, V>>
    where
        KR: Borrow<K::SelfType<'a>>,
    {
        let inner = self.tree.range(&range)?;
        Ok(MultimapRange::new(
            inner,
            self.transaction_guard.clone(),
            self.mem.clone(),
        ))
    }
}

impl<K: Key + 'static, V: Key + 'static> ReadableTableMetadata for ReadOnlyMultimapTable<K, V> {
    fn stats(&self) -> Result<TableStats> {
        let tree_stats = multimap_btree_stats(
            self.tree.get_root().map(|x| x.root),
            &self.mem,
            K::fixed_width(),
            V::fixed_width(),
        )?;

        Ok(TableStats {
            tree_height: tree_stats.tree_height,
            leaf_pages: tree_stats.leaf_pages,
            branch_pages: tree_stats.branch_pages,
            stored_leaf_bytes: tree_stats.stored_leaf_bytes,
            metadata_bytes: tree_stats.metadata_bytes,
            fragmented_bytes: tree_stats.fragmented_bytes,
        })
    }

    fn len(&self) -> Result<u64> {
        Ok(self.num_values)
    }
}

impl<K: Key + 'static, V: Key + 'static> ReadableMultimapTable<K, V>
    for ReadOnlyMultimapTable<K, V>
{
    /// Returns an iterator over all values for the given key. Values are in ascending order.
    fn get<'a>(&self, key: impl Borrow<K::SelfType<'a>>) -> Result<MultimapValue<'_, V>> {
        let iter = if let Some(collection) = self.tree.get(key.borrow())? {
            DynamicCollection::iter(collection, self.transaction_guard.clone(), self.mem.clone())?
        } else {
            MultimapValue::new_subtree(
                BtreeRangeIter::new::<RangeFull, &V::SelfType<'_>>(
                    &(..),
                    None,
                    None,
                    if self.mem.compression().is_enabled() {
                        None
                    } else {
                        <() as Value>::fixed_width()
                    },
                    self.mem.clone(),
                    false,
                )?,
                0,
                self.transaction_guard.clone(),
            )
        };

        Ok(iter)
    }

    fn range<'a, KR>(&self, range: impl RangeBounds<KR> + 'a) -> Result<MultimapRange<'_, K, V>>
    where
        KR: Borrow<K::SelfType<'a>> + 'a,
    {
        let inner = self.tree.range(&range)?;
        Ok(MultimapRange::new(
            inner,
            self.transaction_guard.clone(),
            self.mem.clone(),
        ))
    }
}

impl<K: Key, V: Key> Sealed for ReadOnlyMultimapTable<K, V> {}