Repository: cberner/redb
Pull request #395 (2022-10-11 00:42:56)

Implement Iterator for multimap iterators


Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 80f1cb1..63a63a2 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -29,7 +29,7 @@ jobs:
                 ~/.cargo/registry/cache/
                 ~/.cargo/git/db/
                 target/
-            key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.toml') }}-cargo-deny-0-12-2
+            key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.toml') }}-cargo-deny-0-12-2-cargo-fuzz-0-11-0
 
       - name: Check for forbidden words
         run: "! grep --include='*.rs' -RE 'to_be_bytes|from_be_bytes|dbg!' ."
@@ -63,6 +63,10 @@ jobs:
         if: steps.rust-cache.outputs.cache-hit != 'true'
         run: cargo install --force --version 0.12.2 cargo-deny --locked
 
+      - name: Install cargo-fuzz
+        if: steps.rust-cache.outputs.cache-hit != 'true'
+        run: cargo install --force --version 0.11.0 cargo-fuzz --locked
+
       - name: Compile x64
         run: cargo build --all-targets --all-features
 
@@ -70,6 +74,10 @@ jobs:
         if: startsWith(matrix.os, 'ubuntu')
         run: cargo build --target i686-unknown-linux-gnu
 
+      - name: Fuzzer
+        run: make fuzz_ci
+        if: startsWith(matrix.os, 'ubuntu')
+
       - name: Run tests
         run: make build test
 
diff --git a/Makefile b/Makefile
index 5e18818..599a24c 100644
--- a/Makefile
+++ b/Makefile
@@ -53,6 +53,10 @@ fuzz: pre
 	rustup toolchain install $(NIGHTLY)
 	cargo +$(NIGHTLY) fuzz run fuzz_redb -- -max_len=1000000 -timeout=60
 
+fuzz_ci: pre
+	rustup toolchain install $(NIGHTLY)
+	cargo +$(NIGHTLY) fuzz run fuzz_redb -- -max_len=1000000 -max_total_time=60
+
 .PHONY: fuzz_coverage
 fuzz_coverage: pre
 	rustup toolchain install $(NIGHTLY)
diff --git a/fuzz/fuzz_targets/fuzz_redb.rs b/fuzz/fuzz_targets/fuzz_redb.rs
index 6d5ebb4..6db9fb0 100644
--- a/fuzz/fuzz_targets/fuzz_redb.rs
+++ b/fuzz/fuzz_targets/fuzz_redb.rs
@@ -127,10 +127,10 @@ fn exec_table_inner(db: Arc<Database>, transactions: &[FuzzTransaction], referen
                             } else {
                                 Box::new(local_reference.range(start..end))
                             };
-                        let mut iter = if *reversed {
-                            table.range(start..end).unwrap().rev()
+                        let mut iter: Box<dyn Iterator<Item = (u64, &[u8])>> = if *reversed {
+                            Box::new(table.range(start..end).unwrap().rev())
                         } else {
-                            table.range(start..end).unwrap()
+                            Box::new(table.range(start..end).unwrap())
                         };
                         while let Some((ref_key, ref_value_len)) = reference_iter.next() {
                             let (key, value) = iter.next().unwrap();
@@ -276,10 +276,10 @@ fn exec_multimap_table_inner(db: Arc<Database>, transactions: &[FuzzTransaction]
                             } else {
                                 Box::new(local_reference.range(start..end))
                             };
-                        let mut iter = if *reversed {
-                            table.range(&start..&end).unwrap().rev()
+                        let mut iter: Box<dyn Iterator<Item = (u64, MultimapValueIter<[u8]>)>> = if *reversed {
+                            Box::new(table.range(&start..&end).unwrap().rev())
                         } else {
-                            table.range(&start..&end).unwrap()
+                            Box::new(table.range(&start..&end).unwrap())
                         };
                         while let Some((ref_key, ref_values)) = reference_iter.next() {
                             let (key, value_iter) = iter.next().unwrap();
diff --git a/src/multimap_table.rs b/src/multimap_table.rs
index a3bb066..b2a0188 100644
--- a/src/multimap_table.rs
+++ b/src/multimap_table.rs
@@ -1,10 +1,10 @@
 use crate::multimap_table::DynamicCollectionType::{Inline, Subtree};
 use crate::tree_store::{
-    AllPageNumbersBtreeIter, Btree, BtreeMut, BtreeRangeIter, Checksum, EntryAccessor,
-    LeafAccessor, LeafKeyIter, Page, PageNumber, RawLeafBuilder, TransactionalMemory, BRANCH, LEAF,
+    AllPageNumbersBtreeIter, Btree, BtreeMut, BtreeRangeIter, Checksum, LeafAccessor, LeafKeyIter,
+    Page, PageNumber, RawLeafBuilder, TransactionalMemory, BRANCH, LEAF,
 };
 use crate::types::{
-    AsBytesWithLifetime, OwnedLifetime, RedbKey, RedbValue, RefAsBytesLifetime, WithLifetime,
+    AsBytesWithLifetime, RedbKey, RedbValue, RefAsBytesLifetime, RefLifetime, WithLifetime,
 };
 use crate::{Result, WriteTransaction};
 use std::cell::RefCell;
@@ -78,14 +78,15 @@ impl Into<u8> for DynamicCollectionType {
 /// (when type = 2) root (8 bytes): sub tree root page number
 /// (when type = 2) checksum (16 bytes): sub tree checksum
 #[derive(Debug)]
+#[repr(transparent)]
 struct DynamicCollection {
-    data: Vec<u8>,
+    data: [u8],
     // TODO: include V type when GATs are stable
     // _value_type: PhantomData<V>,
 }
 
 impl RedbValue for DynamicCollection {
-    type View = OwnedLifetime<DynamicCollection>;
+    type View = RefLifetime<DynamicCollection>;
     type ToBytes = RefAsBytesLifetime<[u8]>;
 
     fn fixed_width() -> Option<usize> {
@@ -93,8 +94,7 @@ impl RedbValue for DynamicCollection {
     }
 
     fn from_bytes(data: &[u8]) -> <Self::View as WithLifetime>::Out {
-        // TODO: optimize out this copy
-        Self::new(data.to_vec())
+        Self::new(data)
     }
 
     fn as_bytes(&self) -> <Self::ToBytes as AsBytesWithLifetime>::Out {
@@ -107,8 +107,8 @@ impl RedbValue for DynamicCollection {
 }
 
 impl DynamicCollection {
-    fn new(data: Vec<u8>) -> Self {
-        Self { data }
+    fn new(data: &[u8]) -> &Self {
+        unsafe { mem::transmute(data) }
     }
 
     fn collection_type(&self) -> DynamicCollectionType {
@@ -133,13 +133,13 @@ impl DynamicCollection {
     }
 
     fn iter<'a, V: RedbKey + ?Sized>(
-        &self,
+        &'a self,
         mem: &'a TransactionalMemory,
     ) -> MultimapValueIter<'a, V> {
         match self.collection_type() {
             Inline => {
                 let leaf_iter = LeafKeyIter::new(
-                    self.as_inline().to_vec(), // TODO: optimize out this copy
+                    self.as_inline(),
                     V::fixed_width(),
                     <() as RedbValue>::fixed_width(),
                 );
@@ -157,7 +157,7 @@ impl DynamicCollection {
     }
 
     fn iter_free_on_drop<'a, V: RedbKey + ?Sized>(
-        &self,
+        &'a self,
         pages: Vec<PageNumber>,
         freed_pages: Rc<RefCell<Vec<PageNumber>>>,
         mem: &'a TransactionalMemory,
@@ -165,7 +165,7 @@ impl DynamicCollection {
         match self.collection_type() {
             Inline => {
                 let leaf_iter = LeafKeyIter::new(
-                    self.as_inline().to_vec(), // TODO: optimize out this copy
+                    self.as_inline(),
                     V::fixed_width(),
                     <() as RedbValue>::fixed_width(),
                 );
@@ -179,39 +179,30 @@ impl DynamicCollection {
         }
     }
 
-    fn new_inline(data: &[u8]) -> Self {
+    fn make_inline_data(data: &[u8]) -> Vec<u8> {
         let mut result = vec![Inline.into()];
         result.extend_from_slice(data);
 
-        Self::new(result)
+        result
     }
 
-    fn new_subtree(root: PageNumber, checksum: Checksum) -> Self {
+    fn make_subtree_data(root: PageNumber, checksum: Checksum) -> Vec<u8> {
         let mut result = vec![Subtree.into()];
         result.extend_from_slice(&root.to_le_bytes());
         result.extend_from_slice(checksum.as_bytes().as_ref());
 
-        Self::new(result)
+        result
     }
 }
 
-enum ValueIterState<'a> {
-    Subtree(Box<dyn DoubleEndedIterator<Item = EntryAccessor<'a>> + 'a>),
-    InlineLeaf(LeafKeyIter),
-}
-
-impl<'a> ValueIterState<'a> {
-    fn reverse(self) -> Self {
-        match self {
-            ValueIterState::Subtree(iter) => ValueIterState::Subtree(Box::new(iter.rev())),
-            ValueIterState::InlineLeaf(iter) => ValueIterState::InlineLeaf(iter.reverse()),
-        }
-    }
+enum ValueIterState<'a, V: RedbKey + ?Sized + 'a> {
+    Subtree(BtreeRangeIter<'a, V, ()>),
+    InlineLeaf(LeafKeyIter<'a>),
 }
 
 #[doc(hidden)]
 pub struct MultimapValueIter<'a, V: RedbKey + ?Sized + 'a> {
-    inner: ValueIterState<'a>,
+    inner: ValueIterState<'a, V>,
     freed_pages: Option<Rc<RefCell<Vec<PageNumber>>>>,
     free_on_drop: Vec<PageNumber>,
     mem: Option<&'a TransactionalMemory>,
@@ -221,7 +212,7 @@ pub struct MultimapValueIter<'a, V: RedbKey + ?Sized + 'a> {
 impl<'a, V: RedbKey + ?Sized + 'a> MultimapValueIter<'a, V> {
     fn new_subtree(inner: BtreeRangeIter<'a, V, ()>) -> Self {
         Self {
-            inner: ValueIterState::Subtree(Box::new(inner)),
+            inner: ValueIterState::Subtree(inner),
             freed_pages: None,
             free_on_drop: vec![],
             mem: None,
@@ -236,7 +227,7 @@ impl<'a, V: RedbKey + ?Sized + 'a> MultimapValueIter<'a, V> {
         mem: &'a TransactionalMemory,
     ) -> Self {
         Self {
-            inner: ValueIterState::Subtree(Box::new(inner)),
+            inner: ValueIterState::Subtree(inner),
             freed_pages: Some(freed_pages),
             free_on_drop: pages,
             mem: Some(mem),
@@ -244,7 +235,7 @@ impl<'a, V: RedbKey + ?Sized + 'a> MultimapValueIter<'a, V> {
         }
     }
 
-    fn new_inline(inner: LeafKeyIter) -> Self {
+    fn new_inline(inner: LeafKeyIter<'a>) -> Self {
         Self {
             inner: ValueIterState::InlineLeaf(inner),
             freed_pages: None,
@@ -253,10 +244,12 @@ impl<'a, V: RedbKey + ?Sized + 'a> MultimapValueIter<'a, V> {
             _value_type: Default::default(),
         }
     }
+}
 
-    // TODO: implement Iter when GATs are stable
-    #[allow(clippy::should_implement_trait)]
-    pub fn next(&mut self) -> Option<<<V as RedbValue>::View as WithLifetime>::Out> {
+impl<'a, V: RedbKey + ?Sized> Iterator for MultimapValueIter<'a, V> {
+    type Item = <<V as RedbValue>::View as WithLifetime<'a>>::Out;
+
+    fn next(&mut self) -> Option<Self::Item> {
         match self.inner {
             ValueIterState::Subtree(ref mut iter) => iter.next().map(|e| V::from_bytes(e.key())),
             ValueIterState::InlineLeaf(ref mut iter) => {
@@ -264,23 +257,29 @@ impl<'a, V: RedbKey + ?Sized + 'a> MultimapValueIter<'a, V> {
             }
         }
     }
+}
 
-    pub fn rev(mut self) -> Self {
-        // TODO: remove this hack
-        let mut dummy_state =
-            ValueIterState::InlineLeaf(LeafKeyIter::new(vec![1, 0, 0, 0], None, None));
-        mem::swap(&mut self.inner, &mut dummy_state);
-        self.inner = dummy_state.reverse();
-
-        self
+impl<'a, V: RedbKey + ?Sized> DoubleEndedIterator for MultimapValueIter<'a, V> {
+    fn next_back(&mut self) -> Option<Self::Item> {
+        match self.inner {
+            ValueIterState::Subtree(ref mut iter) => {
+                iter.next_back().map(|e| V::from_bytes(e.key()))
+            }
+            ValueIterState::InlineLeaf(ref mut iter) => {
+                iter.next_key_back().map(|key| V::from_bytes(key))
+            }
+        }
     }
 }
 
 impl<'a, V: RedbKey + ?Sized> Drop for MultimapValueIter<'a, V> {
     fn drop(&mut self) {
         // Drop our references to the pages that are about to be freed
-        let mut dummy_state =
-            ValueIterState::InlineLeaf(LeafKeyIter::new(vec![1, 0, 0, 0], None, None));
+        let mut dummy_state = ValueIterState::InlineLeaf(LeafKeyIter::new(
+            b"\x01\x00\x00\x00".as_slice(),
+            None,
+            None,
+        ));
         mem::swap(&mut self.inner, &mut dummy_state);
         drop(dummy_state);
         for page in self.free_on_drop.iter() {
@@ -300,7 +299,7 @@ impl<'a, V: RedbKey + ?Sized> Drop for MultimapValueIter<'a, V> {
 
 #[doc(hidden)]
 pub struct MultimapRangeIter<'a, K: RedbKey + ?Sized + 'a, V: RedbKey + ?Sized + 'a> {
-    inner: Box<dyn DoubleEndedIterator<Item = EntryAccessor<'a>> + 'a>,
+    inner: BtreeRangeIter<'a, K, DynamicCollection>,
     mem: &'a TransactionalMemory,
     _key_type: PhantomData<K>,
     _value_type: PhantomData<V>,
@@ -309,23 +308,23 @@ pub struct MultimapRangeIter<'a, K: RedbKey + ?Sized + 'a, V: RedbKey + ?Sized +
 impl<'a, K: RedbKey + ?Sized + 'a, V: RedbKey + ?Sized + 'a> MultimapRangeIter<'a, K, V> {
     fn new(inner: BtreeRangeIter<'a, K, DynamicCollection>, mem: &'a TransactionalMemory) -> Self {
         Self {
-            inner: Box::new(inner),
+            inner,
             mem,
             _key_type: Default::default(),
             _value_type: Default::default(),
         }
     }
+}
+
+impl<'a, K: RedbKey + ?Sized + 'a, V: RedbKey + ?Sized + 'a> Iterator
+    for MultimapRangeIter<'a, K, V>
+{
+    type Item = (
+        <<K as RedbValue>::View as WithLifetime<'a>>::Out,
+        MultimapValueIter<'a, V>,
+    );
 
-    // TODO: Simplify this when GATs are stable
-    #[allow(clippy::type_complexity)]
-    // TODO: implement Iter when GATs are stable
-    #[allow(clippy::should_implement_trait)]
-    pub fn next(
-        &mut self,
-    ) -> Option<(
-        <<K as RedbValue>::View as WithLifetime>::Out,
-        MultimapValueIter<V>,
-    )> {
+    fn next(&mut self) -> Option<Self::Item> {
         let entry = self.inner.next()?;
         let key = K::from_bytes(entry.key());
         let collection = DynamicCollection::from_bytes(entry.value());
@@ -333,14 +332,18 @@ impl<'a, K: RedbKey + ?Sized + 'a, V: RedbKey + ?Sized + 'a> MultimapRangeIter<'
 
         Some((key, iter))
     }
+}
 
-    pub fn rev(self) -> Self {
-        Self {
-            inner: Box::new(self.inner.rev()),
-            mem: self.mem,
-            _key_type: Default::default(),
-            _value_type: Default::default(),
-        }
+impl<'a, K: RedbKey + ?Sized + 'a, V: RedbKey + ?Sized + 'a> DoubleEndedIterator
+    for MultimapRangeIter<'a, K, V>
+{
+    fn next_back(&mut self) -> Option<Self::Item> {
+        let entry = self.inner.next_back()?;
+        let key = K::from_bytes(entry.key());
+        let collection = DynamicCollection::from_bytes(entry.value());
+        let iter = collection.iter(self.mem);
+
+        Some((key, iter))
     }
 }
 
@@ -425,9 +428,10 @@ impl<'db, 'txn, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultimapTable<'db, 'tx
                             builder.append(value.as_bytes().as_ref(), ().as_bytes().as_ref());
                         }
                         drop(builder);
+                        let inline_data = DynamicCollection::make_inline_data(&data);
                         unsafe {
                             self.tree
-                                .insert(key, &DynamicCollection::new_inline(&data))?
+                                .insert(key, DynamicCollection::new(&inline_data))?
                         };
                     } else {
                         // convert into a subtree
@@ -448,11 +452,11 @@ impl<'db, 'txn, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultimapTable<'db, 'tx
                         let existed = unsafe { subtree.insert(value, &())?.is_some() };
                         assert_eq!(existed, found);
                         let (new_root, new_checksum) = subtree.get_root().unwrap();
+                        let subtree_data =
+                            DynamicCollection::make_subtree_data(new_root, new_checksum);
                         unsafe {
-                            self.tree.insert(
-                                key,
-                                &DynamicCollection::new_subtree(new_root, new_checksum),
-                            )?
+                            self.tree
+                                .insert(key, DynamicCollection::new(&subtree_data))?
                         };
                     }
 
@@ -466,9 +470,10 @@ impl<'db, 'txn, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultimapTable<'db, 'tx
                     // and we borrow &mut self.
                     let existed = unsafe { subtree.insert(value, &())?.is_some() };
                     let (new_root, new_checksum) = subtree.get_root().unwrap();
+                    let subtree_data = DynamicCollection::make_subtree_data(new_root, new_checksum);
                     unsafe {
                         self.tree
-                            .insert(key, &DynamicCollection::new_subtree(new_root, new_checksum))?
+                            .insert(key, DynamicCollection::new(&subtree_data))?
                     };
 
                     existed
@@ -488,9 +493,10 @@ impl<'db, 'txn, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultimapTable<'db, 'tx
                 );
                 builder.append(value.as_bytes().as_ref(), ().as_bytes().as_ref());
                 drop(builder);
+                let inline_data = DynamicCollection::make_inline_data(&data);
                 unsafe {
                     self.tree
-                        .insert(key, &DynamicCollection::new_inline(&data))?
+                        .insert(key, DynamicCollection::new(&inline_data))?
                 };
             } else {
                 let mut subtree = BtreeMut::new(None, self.mem, self.freed_pages.clone());
@@ -499,9 +505,10 @@ impl<'db, 'txn, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultimapTable<'db, 'tx
                 // and we borrow &mut self.
                 unsafe { subtree.insert(value, &())? };
                 let (new_root, new_checksum) = subtree.get_root().unwrap();
+                let subtree_data = DynamicCollection::make_subtree_data(new_root, new_checksum);
                 unsafe {
                     self.tree
-                        .insert(key, &DynamicCollection::new_subtree(new_root, new_checksum))?
+                        .insert(key, DynamicCollection::new(&subtree_data))?
                 };
             }
             false
@@ -552,9 +559,10 @@ impl<'db, 'txn, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultimapTable<'db, 'tx
                             }
                             drop(builder);
 
+                            let inline_data = DynamicCollection::make_inline_data(&new_data);
                             unsafe {
                                 self.tree
-                                    .insert(key, &DynamicCollection::new_inline(&new_data))?
+                                    .insert(key, DynamicCollection::new(&inline_data))?
                             };
                         }
                         true
@@ -581,11 +589,11 @@ impl<'db, 'txn, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultimapTable<'db, 'tx
                                 );
                                 let len = accessor.total_length();
                                 if len < self.mem.get_page_size() / 2 {
+                                    let inline_data =
+                                        DynamicCollection::make_inline_data(&page.memory()[..len]);
                                     unsafe {
-                                        self.tree.insert(
-                                            key,
-                                            &DynamicCollection::new_inline(&page.memory()[..len]),
-                                        )?
+                                        self.tree
+                                            .insert(key, DynamicCollection::new(&inline_data))?
                                     };
                                     drop(page);
                                     unsafe {
@@ -594,20 +602,24 @@ impl<'db, 'txn, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultimapTable<'db, 'tx
                                         }
                                     }
                                 } else {
+                                    let subtree_data = DynamicCollection::make_subtree_data(
+                                        new_root,
+                                        new_checksum,
+                                    );
                                     unsafe {
-                                        self.tree.insert(
-                                            key,
-                                            &DynamicCollection::new_subtree(new_root, new_checksum),
-                                        )?
+                                        self.tree
+                                            .insert(key, DynamicCollection::new(&subtree_data))?
                                     };
                                 }
                             }
                             BRANCH => {
                                 unsafe {
-                                    self.tree.insert(
-                                        key,
-                                        &DynamicCollection::new_subtree(new_root, new_checksum),
-                                    )?
+                                    let subtree_data = DynamicCollection::make_subtree_data(
+                                        new_root,
+                                        new_checksum,
+                                    );
+                                    self.tree
+                                        .insert(key, DynamicCollection::new(&subtree_data))?
                                 };
                             }
                             _ => unreachable!(),
@@ -634,12 +646,19 @@ impl<'db, 'txn, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultimapTable<'db, 'tx
         // Tables can only be opened mutably in one location (see Error::TableAlreadyOpen),
         // and we borrow &mut self.
         let iter = if let Some(collection) = unsafe { self.tree.remove(key)? } {
-            let mut pages = vec![];
-            if matches!(
-                collection.to_value().collection_type(),
-                DynamicCollectionType::Subtree
-            ) {
-                let root = collection.to_value().as_subtree().0;
+            // TODO: optimize out this copy. The .remove() above should be replaced with
+            // .remove_retain_uncommitted, which should return the PageNumber so we can free it ourselves
+            let len = collection.to_value().as_bytes().len();
+            let mut tmp_page = self.mem.allocate(len)?;
+            let tmp_page_number = tmp_page.get_page_number();
+            tmp_page.memory_mut()[..len].copy_from_slice(collection.to_value().as_bytes());
+            let mut pages = vec![tmp_page_number];
+            drop(tmp_page);
+            let tmp_page = self.mem.get_page(tmp_page_number);
+            let collection = DynamicCollection::new(&tmp_page.memory_full_lifetime()[..len]);
+
+            if matches!(collection.collection_type(), DynamicCollectionType::Subtree) {
+                let root = collection.as_subtree().0;
                 let all_pages = AllPageNumbersBtreeIter::new(
                     root,
                     V::fixed_width(),
@@ -650,9 +669,7 @@ impl<'db, 'txn, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultimapTable<'db, 'tx
                     pages.push(page);
                 }
             }
-            collection
-                .to_value()
-                .iter_free_on_drop(pages, self.freed_pages.clone(), self.mem)
+            collection.iter_free_on_drop(pages, self.freed_pages.clone(), self.mem)
         } else {
             MultimapValueIter::new_subtree(BtreeRangeIter::new::<RangeFull, &V>(.., None, self.mem))
         };
@@ -686,9 +703,8 @@ impl<'db, 'txn, K: RedbKey + ?Sized, V: RedbKey + ?Sized> ReadableMultimapTable<
 
     /// Returns the number of key-value pairs in the table
     fn len(&self) -> Result<usize> {
-        let mut iter: MultimapRangeIter<K, V> = self.range(..)?;
         let mut count = 0;
-        while let Some((_, mut values)) = iter.next() {
+        for (_, mut values) in self.range(..)? {
             while values.next().is_some() {
                 count += 1;
             }
@@ -764,9 +780,8 @@ impl<'txn, K: RedbKey + ?Sized, V: RedbKey + ?Sized> ReadableMultimapTable<K, V>
     }
 
     fn len(&self) -> Result<usize> {
-        let mut iter: MultimapRangeIter<K, V> = self.range(..)?;
         let mut count = 0;
-        while let Some((_, mut values)) = iter.next() {
+        for (_, mut values) in self.range(..)? {
             while values.next().is_some() {
                 count += 1;
             }
diff --git a/src/tree_store/btree_base.rs b/src/tree_store/btree_base.rs
index 52598a7..0a7160a 100644
--- a/src/tree_store/btree_base.rs
+++ b/src/tree_store/btree_base.rs
@@ -528,22 +528,21 @@ impl<'a> LeafAccessor<'a> {
     }
 }
 
-pub(crate) struct LeafKeyIter {
-    data: Vec<u8>,
+pub(crate) struct LeafKeyIter<'a> {
+    data: &'a [u8],
     fixed_key_size: Option<usize>,
     fixed_value_size: Option<usize>,
     start_entry: isize, // inclusive
     end_entry: isize,   // inclusive
-    reverse: bool,
 }
 
-impl LeafKeyIter {
+impl<'a> LeafKeyIter<'a> {
     pub(crate) fn new(
-        data: Vec<u8>,
+        data: &'a [u8],
         fixed_key_size: Option<usize>,
         fixed_value_size: Option<usize>,
     ) -> Self {
-        let accessor = LeafAccessor::new(&data, fixed_key_size, fixed_value_size);
+        let accessor = LeafAccessor::new(data, fixed_key_size, fixed_value_size);
         let end_entry = isize::try_from(accessor.num_pairs()).unwrap() - 1;
         Self {
             data,
@@ -551,37 +550,29 @@ impl LeafKeyIter {
             fixed_value_size,
             start_entry: 0,
             end_entry,
-            reverse: false,
         }
     }
 
-    pub(crate) fn next_key(&mut self) -> Option<&[u8]> {
+    pub(crate) fn next_key(&mut self) -> Option<&'a [u8]> {
         if self.end_entry < self.start_entry {
             return None;
         }
-        let accessor = LeafAccessor::new(&self.data, self.fixed_key_size, self.fixed_value_size);
-        if self.reverse {
-            self.end_entry -= 1;
-            accessor
-                .entry((self.end_entry + 1).try_into().unwrap())
-                .map(|e| e.key())
-        } else {
-            self.start_entry += 1;
-            accessor
-                .entry((self.start_entry - 1).try_into().unwrap())
-                .map(|e| e.key())
-        }
+        let accessor = LeafAccessor::new(self.data, self.fixed_key_size, self.fixed_value_size);
+        self.start_entry += 1;
+        accessor
+            .entry((self.start_entry - 1).try_into().unwrap())
+            .map(|e| e.key())
     }
 
-    pub(crate) fn reverse(self) -> Self {
-        Self {
-            data: self.data,
-            fixed_key_size: self.fixed_key_size,
-            fixed_value_size: self.fixed_value_size,
-            start_entry: self.start_entry,
-            end_entry: self.end_entry,
-            reverse: !self.reverse,
+    pub(crate) fn next_key_back(&mut self) -> Option<&'a [u8]> {
+        if self.end_entry < self.start_entry {
+            return None;
         }
+        let accessor = LeafAccessor::new(self.data, self.fixed_key_size, self.fixed_value_size);
+        self.end_entry -= 1;
+        accessor
+            .entry((self.end_entry + 1).try_into().unwrap())
+            .map(|e| e.key())
     }
 }
 
diff --git a/src/tree_store/mod.rs b/src/tree_store/mod.rs
index 3f9286f..2e4d337 100644
--- a/src/tree_store/mod.rs
+++ b/src/tree_store/mod.rs
@@ -9,9 +9,7 @@ pub(crate) use btree::{Btree, BtreeMut, RawBtree};
 pub use btree_base::AccessGuard;
 pub(crate) use btree_base::AccessGuardMut;
 pub(crate) use btree_base::Checksum;
-pub(crate) use btree_base::{
-    EntryAccessor, LeafAccessor, LeafKeyIter, RawLeafBuilder, BRANCH, LEAF,
-};
+pub(crate) use btree_base::{LeafAccessor, LeafKeyIter, RawLeafBuilder, BRANCH, LEAF};
 pub(crate) use btree_iters::{AllPageNumbersBtreeIter, BtreeRangeIter};
 pub use page_store::Savepoint;
 pub(crate) use page_store::{get_db_size, Page, PageNumber, TransactionalMemory};
diff --git a/src/tree_store/page_store/mmap/unix.rs b/src/tree_store/page_store/mmap/unix.rs
index dfa58ab..8195009 100644
--- a/src/tree_store/page_store/mmap/unix.rs
+++ b/src/tree_store/page_store/mmap/unix.rs
@@ -92,7 +92,7 @@ impl MmapInner {
     }
 
     #[inline]
-    pub(super) fn flush(&self, owner: &Mmap) -> Result {
+    pub(super) fn flush(&self, #[allow(unused_variables)] owner: &Mmap) -> Result {
         // Disable fsync when fuzzing, since it doesn't test crash consistency
         #[cfg(not(fuzzing))]
         {


Test patch:
diff --git a/tests/basic_tests.rs b/tests/basic_tests.rs
index 1fb70be..08083a8 100644
--- a/tests/basic_tests.rs
+++ b/tests/basic_tests.rs
@@ -459,6 +459,7 @@ fn str_type() {
     let hello = "hello".to_string();
     assert_eq!("world", table.get(&hello).unwrap().unwrap());
 
+    // TODO: is there a way to avoid having to annotate the return type for RangeFull queries?
     let mut iter: RangeIter<str, str> = table.range::<RangeFull, &str>(..).unwrap();
     assert_eq!(iter.next().unwrap().1, "world");
     assert!(iter.next().is_none());
diff --git a/tests/multimap_tests.rs b/tests/multimap_tests.rs
index 2162ff4..003a8e8 100644
--- a/tests/multimap_tests.rs
+++ b/tests/multimap_tests.rs
@@ -118,6 +118,12 @@ fn range_query() {
         }
     }
     assert!(iter.next().is_none());
+
+    let mut total: u64 = 0;
+    for (_, values) in table.range(start..=end).unwrap() {
+        total += values.sum::<u64>();
+    }
+    assert_eq!(total, 45);
 }
 
 #[test]