Repository: cberner/redb
Pull request #181 (2022-02-20 18:06:02)

Simplify iterators and unify range() & range_reversed()


Patch:
diff --git a/src/multimap_table.rs b/src/multimap_table.rs
index 3cce4b4..9fbf30f 100644
--- a/src/multimap_table.rs
+++ b/src/multimap_table.rs
@@ -8,7 +8,7 @@ use std::cmp::Ordering;
 use std::collections::Bound;
 use std::convert::TryInto;
 use std::marker::PhantomData;
-use std::ops::{RangeBounds, RangeInclusive};
+use std::ops::RangeBounds;
 use std::rc::Rc;
 
 #[derive(Eq, PartialEq)]
@@ -215,23 +215,12 @@ fn make_bound<'a, K: RedbKey + ?Sized + 'a, V: RedbKey + ?Sized + 'a>(
     }
 }
 
-pub struct MultimapValueIter<
-    'a,
-    T: RangeBounds<MultimapKVPair<K, V>>,
-    K: RedbKey + ?Sized + 'a,
-    V: RedbKey + ?Sized + 'a,
-> {
-    inner: BtreeRangeIter<'a, T, MultimapKVPair<K, V>, MultimapKVPair<K, V>, [u8]>,
+pub struct MultimapValueIter<'a, K: RedbKey + ?Sized + 'a, V: RedbKey + ?Sized + 'a> {
+    inner: BtreeRangeIter<'a, MultimapKVPair<K, V>, [u8]>,
 }
 
-impl<
-        'a,
-        T: RangeBounds<MultimapKVPair<K, V>>,
-        K: RedbKey + ?Sized + 'a,
-        V: RedbKey + ?Sized + 'a,
-    > MultimapValueIter<'a, T, K, V>
-{
-    fn new(inner: BtreeRangeIter<'a, T, MultimapKVPair<K, V>, MultimapKVPair<K, V>, [u8]>) -> Self {
+impl<'a, K: RedbKey + ?Sized + 'a, V: RedbKey + ?Sized + 'a> MultimapValueIter<'a, K, V> {
+    fn new(inner: BtreeRangeIter<'a, MultimapKVPair<K, V>, [u8]>) -> Self {
         Self { inner }
     }
 
@@ -249,25 +238,18 @@ impl<
             None
         }
     }
+
+    pub fn rev(self) -> Self {
+        Self::new(self.inner.reverse())
+    }
 }
 
-pub struct MultimapRangeIter<
-    'a,
-    T: RangeBounds<MultimapKVPair<K, V>>,
-    K: RedbKey + ?Sized + 'a,
-    V: RedbKey + ?Sized + 'a,
-> {
-    inner: BtreeRangeIter<'a, T, MultimapKVPair<K, V>, MultimapKVPair<K, V>, [u8]>,
+pub struct MultimapRangeIter<'a, K: RedbKey + ?Sized + 'a, V: RedbKey + ?Sized + 'a> {
+    inner: BtreeRangeIter<'a, MultimapKVPair<K, V>, [u8]>,
 }
 
-impl<
-        'a,
-        T: RangeBounds<MultimapKVPair<K, V>>,
-        K: RedbKey + ?Sized + 'a,
-        V: RedbKey + ?Sized + 'a,
-    > MultimapRangeIter<'a, T, K, V>
-{
-    fn new(inner: BtreeRangeIter<'a, T, MultimapKVPair<K, V>, MultimapKVPair<K, V>, [u8]>) -> Self {
+impl<'a, K: RedbKey + ?Sized + 'a, V: RedbKey + ?Sized + 'a> MultimapRangeIter<'a, K, V> {
+    fn new(inner: BtreeRangeIter<'a, MultimapKVPair<K, V>, [u8]>) -> Self {
         Self { inner }
     }
 
@@ -294,6 +276,10 @@ impl<
             None
         }
     }
+
+    pub fn rev(self) -> Self {
+        Self::new(self.inner.reverse())
+    }
 }
 
 pub struct MultimapTable<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> {
@@ -360,7 +346,7 @@ impl<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultimapTable<'s, K, V> {
         Ok(found.is_some())
     }
 
-    pub fn remove_all(&mut self, key: &K) -> Result<MultimapGetIterType<K, V>, Error> {
+    pub fn remove_all(&mut self, key: &K) -> Result<MultimapValueIter<K, V>, Error> {
         // Match only on the key, so that we can remove all the associated values
         let key_only = make_serialized_key_with_op(key, MultimapKeyCompareOp::KeyOnly);
         let original_root = self.table_root.get();
@@ -396,7 +382,7 @@ impl<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> MultimapTable<'s, K, V> {
 impl<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> ReadableMultimapTable<K, V>
     for MultimapTable<'s, K, V>
 {
-    fn get<'a>(&'a self, key: &'a K) -> Result<MultimapGetIterType<'a, K, V>, Error> {
+    fn get<'a>(&'a self, key: &'a K) -> Result<MultimapValueIter<'a, K, V>, Error> {
         let lower_bytes = make_serialized_key_with_op(key, MultimapKeyCompareOp::KeyMinusEpsilon);
         let upper_bytes = make_serialized_key_with_op(key, MultimapKeyCompareOp::KeyPlusEpsilon);
         let lower = MultimapKVPair::<K, V>::new(lower_bytes);
@@ -409,7 +395,7 @@ impl<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> ReadableMultimapTable<K, V>
     fn range<'a, T: RangeBounds<&'a K> + 'a>(
         &'a self,
         range: T,
-    ) -> Result<MultimapGetRangeIterType<'a, K, V>, Error> {
+    ) -> Result<MultimapRangeIter<'a, K, V>, Error> {
         let (start_bytes, end_bytes) = make_inclusive_query_range(range);
         let start_kv = start_bytes.map(MultimapKVPair::<K, V>::new);
         let end_kv = end_bytes.map(MultimapKVPair::<K, V>::new);
@@ -421,21 +407,6 @@ impl<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> ReadableMultimapTable<K, V>
             .map(MultimapRangeIter::new)
     }
 
-    fn range_reversed<'a, T: RangeBounds<&'a K> + 'a>(
-        &'a self,
-        range: T,
-    ) -> Result<MultimapGetRangeIterType<'a, K, V>, Error> {
-        let (start_bytes, end_bytes) = make_inclusive_query_range(range);
-        let start_kv = start_bytes.map(MultimapKVPair::<K, V>::new);
-        let end_kv = end_bytes.map(MultimapKVPair::<K, V>::new);
-        let start = make_bound(start_kv);
-        let end = make_bound(end_kv);
-
-        self.storage
-            .get_range_reversed((start, end), self.table_root.get())
-            .map(MultimapRangeIter::new)
-    }
-
     fn len(&self) -> Result<usize, Error> {
         self.storage.len(self.table_root.get())
     }
@@ -445,24 +416,13 @@ impl<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> ReadableMultimapTable<K, V>
     }
 }
 
-type MultimapGetIterType<'a, K, V> =
-    MultimapValueIter<'a, RangeInclusive<MultimapKVPair<K, V>>, K, V>;
-type MultimapGetRangeIterType<'a, K, V> =
-    MultimapRangeIter<'a, (Bound<MultimapKVPair<K, V>>, Bound<MultimapKVPair<K, V>>), K, V>;
-
 pub trait ReadableMultimapTable<K: RedbKey + ?Sized, V: RedbKey + ?Sized> {
-    fn get<'a>(&'a self, key: &'a K) -> Result<MultimapGetIterType<'a, K, V>, Error>;
+    fn get<'a>(&'a self, key: &'a K) -> Result<MultimapValueIter<'a, K, V>, Error>;
 
     fn range<'a, T: RangeBounds<&'a K> + 'a>(
         &'a self,
         range: T,
-    ) -> Result<MultimapGetRangeIterType<'a, K, V>, Error>;
-
-    // TODO: remove this method and make the iterator returned from range() double ended
-    fn range_reversed<'a, T: RangeBounds<&'a K> + 'a>(
-        &'a self,
-        range: T,
-    ) -> Result<MultimapGetRangeIterType<'a, K, V>, Error>;
+    ) -> Result<MultimapRangeIter<'a, K, V>, Error>;
 
     fn len(&self) -> Result<usize, Error>;
 
@@ -493,7 +453,7 @@ impl<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> ReadOnlyMultimapTable<'s, K,
 impl<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> ReadableMultimapTable<K, V>
     for ReadOnlyMultimapTable<'s, K, V>
 {
-    fn get<'a>(&'a self, key: &'a K) -> Result<MultimapGetIterType<'a, K, V>, Error> {
+    fn get<'a>(&'a self, key: &'a K) -> Result<MultimapValueIter<'a, K, V>, Error> {
         let lower_bytes = make_serialized_key_with_op(key, MultimapKeyCompareOp::KeyMinusEpsilon);
         let upper_bytes = make_serialized_key_with_op(key, MultimapKeyCompareOp::KeyPlusEpsilon);
         let lower = MultimapKVPair::<K, V>::new(lower_bytes);
@@ -506,7 +466,7 @@ impl<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> ReadableMultimapTable<K, V>
     fn range<'a, T: RangeBounds<&'a K> + 'a>(
         &'a self,
         range: T,
-    ) -> Result<MultimapGetRangeIterType<'a, K, V>, Error> {
+    ) -> Result<MultimapRangeIter<'a, K, V>, Error> {
         let (start_bytes, end_bytes) = make_inclusive_query_range(range);
         let start_kv = start_bytes.map(MultimapKVPair::<K, V>::new);
         let end_kv = end_bytes.map(MultimapKVPair::<K, V>::new);
@@ -518,21 +478,6 @@ impl<'s, K: RedbKey + ?Sized, V: RedbKey + ?Sized> ReadableMultimapTable<K, V>
             .map(MultimapRangeIter::new)
     }
 
-    fn range_reversed<'a, T: RangeBounds<&'a K> + 'a>(
-        &'a self,
-        range: T,
-    ) -> Result<MultimapGetRangeIterType<'a, K, V>, Error> {
-        let (start_bytes, end_bytes) = make_inclusive_query_range(range);
-        let start_kv = start_bytes.map(MultimapKVPair::<K, V>::new);
-        let end_kv = end_bytes.map(MultimapKVPair::<K, V>::new);
-        let start = make_bound(start_kv);
-        let end = make_bound(end_kv);
-
-        self.storage
-            .get_range_reversed((start, end), self.table_root)
-            .map(MultimapRangeIter::new)
-    }
-
     fn len(&self) -> Result<usize, Error> {
         self.storage.len(self.table_root)
     }
diff --git a/src/table.rs b/src/table.rs
index 78329a5..dd83903 100644
--- a/src/table.rs
+++ b/src/table.rs
@@ -99,24 +99,15 @@ impl<'s, K: RedbKey + ?Sized, V: RedbValue + ?Sized> ReadableTable<K, V> for Tab
             .get::<K, V>(key.as_bytes().as_ref(), self.table_root.get())
     }
 
-    fn range<'a, T: RangeBounds<KR> + 'a, KR: Borrow<K>>(
+    fn range<'a, T: RangeBounds<KR>, KR: Borrow<K> + 'a>(
         &'a self,
         range: T,
-    ) -> Result<RangeIter<T, KR, K, V>, Error> {
+    ) -> Result<RangeIter<K, V>, Error> {
         self.storage
             .get_range(range, self.table_root.get())
             .map(RangeIter::new)
     }
 
-    fn range_reversed<'a, T: RangeBounds<KR> + 'a, KR: Borrow<K>>(
-        &'a self,
-        range: T,
-    ) -> Result<RangeIter<T, KR, K, V>, Error> {
-        self.storage
-            .get_range_reversed(range, self.table_root.get())
-            .map(RangeIter::new)
-    }
-
     fn len(&self) -> Result<usize, Error> {
         self.storage.len(self.table_root.get())
     }
@@ -129,16 +120,10 @@ impl<'s, K: RedbKey + ?Sized, V: RedbValue + ?Sized> ReadableTable<K, V> for Tab
 pub trait ReadableTable<K: RedbKey + ?Sized, V: RedbValue + ?Sized> {
     fn get(&self, key: &K) -> Result<Option<<<V as RedbValue>::View as WithLifetime>::Out>, Error>;
 
-    fn range<'a, T: RangeBounds<KR> + 'a, KR: Borrow<K>>(
-        &'a self,
-        range: T,
-    ) -> Result<RangeIter<T, KR, K, V>, Error>;
-
-    // TODO: remove this method and make the iterator returned from range() double ended
-    fn range_reversed<'a, T: RangeBounds<KR> + 'a, KR: Borrow<K>>(
+    fn range<'a, T: RangeBounds<KR>, KR: Borrow<K> + 'a>(
         &'a self,
         range: T,
-    ) -> Result<RangeIter<T, KR, K, V>, Error>;
+    ) -> Result<RangeIter<K, V>, Error>;
 
     fn len(&self) -> Result<usize, Error>;
 
@@ -174,24 +159,15 @@ impl<'s, K: RedbKey + ?Sized, V: RedbValue + ?Sized> ReadableTable<K, V>
             .get::<K, V>(key.as_bytes().as_ref(), self.table_root)
     }
 
-    fn range<'a, T: RangeBounds<KR> + 'a, KR: Borrow<K>>(
+    fn range<'a, T: RangeBounds<KR>, KR: Borrow<K> + 'a>(
         &'a self,
         range: T,
-    ) -> Result<RangeIter<T, KR, K, V>, Error> {
+    ) -> Result<RangeIter<K, V>, Error> {
         self.storage
             .get_range(range, self.table_root)
             .map(RangeIter::new)
     }
 
-    fn range_reversed<'a, T: RangeBounds<KR> + 'a, KR: Borrow<K>>(
-        &'a self,
-        range: T,
-    ) -> Result<RangeIter<T, KR, K, V>, Error> {
-        self.storage
-            .get_range_reversed(range, self.table_root)
-            .map(RangeIter::new)
-    }
-
     fn len(&self) -> Result<usize, Error> {
         self.storage.len(self.table_root)
     }
@@ -201,25 +177,12 @@ impl<'s, K: RedbKey + ?Sized, V: RedbValue + ?Sized> ReadableTable<K, V>
     }
 }
 
-pub struct RangeIter<
-    'a,
-    T: RangeBounds<KR>,
-    KR: Borrow<K>,
-    K: RedbKey + ?Sized + 'a,
-    V: RedbValue + ?Sized + 'a,
-> {
-    inner: BtreeRangeIter<'a, T, KR, K, V>,
+pub struct RangeIter<'a, K: RedbKey + ?Sized + 'a, V: RedbValue + ?Sized + 'a> {
+    inner: BtreeRangeIter<'a, K, V>,
 }
 
-impl<
-        'a,
-        T: RangeBounds<KR>,
-        KR: Borrow<K>,
-        K: RedbKey + ?Sized + 'a,
-        V: RedbValue + ?Sized + 'a,
-    > RangeIter<'a, T, KR, K, V>
-{
-    fn new(inner: BtreeRangeIter<'a, T, KR, K, V>) -> Self {
+impl<'a, K: RedbKey + ?Sized + 'a, V: RedbValue + ?Sized + 'a> RangeIter<'a, K, V> {
+    fn new(inner: BtreeRangeIter<'a, K, V>) -> Self {
         Self { inner }
     }
 
@@ -241,6 +204,10 @@ impl<
             None
         }
     }
+
+    pub fn rev(self) -> Self {
+        Self::new(self.inner.reverse())
+    }
 }
 
 #[cfg(test)]
diff --git a/src/tree_store/btree_iters.rs b/src/tree_store/btree_iters.rs
index c87d48c..d015bf9 100644
--- a/src/tree_store/btree_iters.rs
+++ b/src/tree_store/btree_iters.rs
@@ -9,50 +9,17 @@ use std::collections::Bound;
 use std::marker::PhantomData;
 use std::ops::RangeBounds;
 
-fn bound_contains_key<
-    'a,
-    T: RangeBounds<KR>,
-    KR: Borrow<K> + ?Sized + 'a,
-    K: RedbKey + ?Sized + 'a,
->(
-    range: &T,
-    key: &[u8],
-) -> bool {
-    if let Bound::Included(start) = range.start_bound() {
-        if K::compare(key, start.borrow().as_bytes().as_ref()).is_lt() {
-            return false;
-        }
-    } else if let Bound::Excluded(start) = range.start_bound() {
-        if K::compare(key, start.borrow().as_bytes().as_ref()).is_le() {
-            return false;
-        }
-    }
-    if let Bound::Included(end) = range.end_bound() {
-        if K::compare(key, end.borrow().as_bytes().as_ref()).is_gt() {
-            return false;
-        }
-    } else if let Bound::Excluded(end) = range.end_bound() {
-        if K::compare(key, end.borrow().as_bytes().as_ref()).is_ge() {
-            return false;
-        }
-    }
-
-    true
-}
-
 #[derive(Debug)]
 pub enum RangeIterState<'a> {
     Leaf {
         page: PageImpl<'a>,
         entry: usize,
         parent: Option<Box<RangeIterState<'a>>>,
-        reversed: bool,
     },
     Internal {
         page: PageImpl<'a>,
         child: usize,
         parent: Option<Box<RangeIterState<'a>>>,
-        reversed: bool,
     },
 }
 
@@ -64,23 +31,21 @@ impl<'a> RangeIterState<'a> {
         }
     }
 
-    fn next(self, manager: &'a TransactionalMemory) -> Option<RangeIterState> {
+    fn next(self, reverse: bool, manager: &'a TransactionalMemory) -> Option<RangeIterState> {
         match self {
             Leaf {
                 page,
                 entry,
                 parent,
-                reversed,
             } => {
                 let accessor = LeafAccessor::new(&page);
-                let direction = if reversed { -1 } else { 1 };
+                let direction = if reverse { -1 } else { 1 };
                 let next_entry = entry as isize + direction;
                 if 0 <= next_entry && next_entry < accessor.num_pairs() as isize {
                     Some(Leaf {
                         page,
                         entry: next_entry as usize,
                         parent,
-                        reversed,
                     })
                 } else {
                     parent.map(|x| *x)
@@ -90,25 +55,23 @@ impl<'a> RangeIterState<'a> {
                 page,
                 child,
                 mut parent,
-                reversed,
             } => {
                 let accessor = InternalAccessor::new(&page);
                 let child_page = accessor.child_page(child).unwrap();
                 let child_page = manager.get_page(child_page);
-                let direction = if reversed { -1 } else { 1 };
+                let direction = if reverse { -1 } else { 1 };
                 let next_child = child as isize + direction;
                 if 0 <= next_child && next_child < accessor.count_children() as isize {
                     parent = Some(Box::new(Internal {
                         page,
                         child: next_child as usize,
                         parent,
-                        reversed,
                     }));
                 }
                 match child_page.memory()[0] {
                     LEAF => {
                         let child_accessor = LeafAccessor::new(&child_page);
-                        let entry = if reversed {
+                        let entry = if reverse {
                             child_accessor.num_pairs() - 1
                         } else {
                             0
@@ -117,12 +80,11 @@ impl<'a> RangeIterState<'a> {
                             page: child_page,
                             entry,
                             parent,
-                            reversed,
                         })
                     }
                     INTERNAL => {
                         let child_accessor = InternalAccessor::new(&child_page);
-                        let child = if reversed {
+                        let child = if reverse {
                             child_accessor.count_children() - 1
                         } else {
                             0
@@ -131,7 +93,6 @@ impl<'a> RangeIterState<'a> {
                             page: child_page,
                             child,
                             parent,
-                            reversed,
                         })
                     }
                     _ => unreachable!(),
@@ -178,7 +139,7 @@ impl<'a> Iterator for AllPageNumbersBtreeIter<'a> {
         let value = state.page_number();
         // Only return each page number the first time we visit it
         loop {
-            if let Some(next_state) = state.next(self.manager) {
+            if let Some(next_state) = state.next(false, self.manager) {
                 state = next_state;
             } else {
                 self.next = None;
@@ -210,125 +171,205 @@ pub(crate) fn page_numbers_iter_start_state(page: PageImpl) -> RangeIterState {
             page,
             entry: 0,
             parent: None,
-            reversed: false,
         },
         INTERNAL => Internal {
             page,
             child: 0,
             parent: None,
-            reversed: false,
         },
         _ => unreachable!(),
     }
 }
 
-pub struct BtreeRangeIter<
-    'a,
-    T: RangeBounds<KR>,
-    KR: Borrow<K> + ?Sized + 'a,
-    K: RedbKey + ?Sized + 'a,
-    V: RedbValue + ?Sized + 'a,
-> {
-    // whether we've returned the value for the self.next state. We don't want to advance the initial
-    // state, until it has been returned
-    consumed: bool,
-    next: Option<RangeIterState<'a>>,
-    query_range: T,
+pub struct BtreeRangeIter<'a, K: RedbKey + ?Sized + 'a, V: RedbValue + ?Sized + 'a> {
+    left: Option<RangeIterState<'a>>, // Exclusive. The previous element returned
+    right: Option<RangeIterState<'a>>, // Exclusive. The previous element returned
+    include_left: bool,               // left is inclusive, instead of exclusive
+    include_right: bool,              // right is inclusive, instead of exclusive
     reversed: bool,
     manager: &'a TransactionalMemory,
     _key_type: PhantomData<K>,
-    _key_ref_type: PhantomData<KR>,
     _value_type: PhantomData<V>,
 }
 
-impl<
-        'a,
-        T: RangeBounds<KR>,
-        KR: Borrow<K> + ?Sized + 'a,
-        K: RedbKey + ?Sized + 'a,
-        V: RedbValue + ?Sized + 'a,
-    > BtreeRangeIter<'a, T, KR, K, V>
-{
-    pub(crate) fn new(
-        state: Option<RangeIterState<'a>>,
+impl<'a, K: RedbKey + ?Sized + 'a, V: RedbValue + ?Sized + 'a> BtreeRangeIter<'a, K, V> {
+    pub(crate) fn new<T: RangeBounds<KR>, KR: Borrow<K> + ?Sized + 'a>(
         query_range: T,
+        table_root: Option<PageNumber>,
         manager: &'a TransactionalMemory,
     ) -> Self {
-        Self {
-            consumed: false,
-            next: state,
-            query_range,
-            reversed: false,
-            manager,
-            _key_type: Default::default(),
-            _key_ref_type: Default::default(),
-            _value_type: Default::default(),
+        if let Some(root) = table_root {
+            let (include_left, left) = match query_range.start_bound() {
+                Bound::Included(k) => find_iter_left::<K>(
+                    manager.get_page(root),
+                    None,
+                    k.borrow().as_bytes().as_ref(),
+                    true,
+                    manager,
+                ),
+                Bound::Excluded(k) => find_iter_left::<K>(
+                    manager.get_page(root),
+                    None,
+                    k.borrow().as_bytes().as_ref(),
+                    false,
+                    manager,
+                ),
+                Bound::Unbounded => {
+                    let state = find_iter_unbounded_left(manager.get_page(root), None, manager);
+                    (true, state)
+                }
+            };
+            let (include_right, right) = match query_range.end_bound() {
+                Bound::Included(k) => find_iter_right::<K>(
+                    manager.get_page(root),
+                    None,
+                    k.borrow().as_bytes().as_ref(),
+                    true,
+                    manager,
+                ),
+                Bound::Excluded(k) => find_iter_right::<K>(
+                    manager.get_page(root),
+                    None,
+                    k.borrow().as_bytes().as_ref(),
+                    false,
+                    manager,
+                ),
+                Bound::Unbounded => {
+                    let state = find_iter_unbounded_right(manager.get_page(root), None, manager);
+                    (true, state)
+                }
+            };
+            Self {
+                left,
+                right,
+                include_left,
+                include_right,
+                reversed: false,
+                manager,
+                _key_type: Default::default(),
+                _value_type: Default::default(),
+            }
+        } else {
+            Self {
+                left: None,
+                right: None,
+                include_left: false,
+                include_right: false,
+                reversed: false,
+                manager,
+                _key_type: Default::default(),
+                _value_type: Default::default(),
+            }
         }
     }
 
-    pub(crate) fn new_reversed(
-        state: Option<RangeIterState<'a>>,
-        query_range: T,
-        manager: &'a TransactionalMemory,
-    ) -> Self {
+    pub(crate) fn reverse(self) -> Self {
         Self {
-            consumed: false,
-            next: state,
-            query_range,
-            reversed: true,
-            manager,
+            left: self.left,
+            right: self.right,
+            include_left: self.include_left,
+            include_right: self.include_right,
+            reversed: !self.reversed,
+            manager: self.manager,
             _key_type: Default::default(),
-            _key_ref_type: Default::default(),
             _value_type: Default::default(),
         }
     }
 
     // TODO: we need generic-associated-types to implement Iterator
     pub fn next(&mut self) -> Option<EntryAccessor> {
+        if let (
+            Some(Leaf {
+                page: left_page,
+                entry: left_entry,
+                parent: _,
+            }),
+            Some(Leaf {
+                page: right_page,
+                entry: right_entry,
+                parent: _,
+            }),
+        ) = (&self.left, &self.right)
+        {
+            if left_page.get_page_number() == right_page.get_page_number()
+                && left_entry >= right_entry
+                && !(self.include_left && self.include_right)
+            {
+                return None;
+            }
+        }
+
         loop {
-            if self.consumed {
-                let state = self.next.take()?;
-                self.next = state.next(self.manager);
+            if !self.reversed {
+                if !self.include_left {
+                    self.left = self.left.take()?.next(self.reversed, self.manager);
+                }
                 // Return None if the next state is None
-                self.next.as_ref()?;
-            }
+                self.left.as_ref()?;
 
-            self.next.as_ref()?;
+                if let (
+                    Some(Leaf {
+                        page: left_page,
+                        entry: left_entry,
+                        parent: _,
+                    }),
+                    Some(Leaf {
+                        page: right_page,
+                        entry: right_entry,
+                        parent: _,
+                    }),
+                ) = (&self.left, &self.right)
+                {
+                    if left_page.get_page_number() == right_page.get_page_number()
+                        && left_entry >= right_entry
+                        && !self.include_right
+                    {
+                        return None;
+                    }
+                }
 
-            self.consumed = true;
-            if let Some(entry) = self.next.as_ref().unwrap().get_entry() {
-                if bound_contains_key::<T, KR, K>(&self.query_range, entry.key()) {
-                    return self.next.as_ref().map(|s| s.get_entry().unwrap());
-                } else {
-                    #[allow(clippy::collapsible_else_if)]
-                    if self.reversed {
-                        if let Bound::Included(start) = self.query_range.start_bound() {
-                            if K::compare(entry.key(), start.borrow().as_bytes().as_ref()).is_lt() {
-                                self.next = None;
-                            }
-                        } else if let Bound::Excluded(start) = self.query_range.start_bound() {
-                            if K::compare(entry.key(), start.borrow().as_bytes().as_ref()).is_le() {
-                                self.next = None;
-                            }
-                        }
-                    } else {
-                        if let Bound::Included(end) = self.query_range.end_bound() {
-                            if K::compare(entry.key(), end.borrow().as_bytes().as_ref()).is_gt() {
-                                self.next = None;
-                            }
-                        } else if let Bound::Excluded(end) = self.query_range.end_bound() {
-                            if K::compare(entry.key(), end.borrow().as_bytes().as_ref()).is_ge() {
-                                self.next = None;
-                            }
-                        }
-                    };
+                self.include_left = false;
+                if self.left.as_ref().unwrap().get_entry().is_some() {
+                    return self.left.as_ref().map(|s| s.get_entry().unwrap());
+                }
+            } else {
+                if !self.include_right {
+                    self.right = self.right.take()?.next(self.reversed, self.manager);
+                }
+                // Return None if the next state is None
+                self.right.as_ref()?;
+
+                if let (
+                    Some(Leaf {
+                        page: left_page,
+                        entry: left_entry,
+                        parent: _,
+                    }),
+                    Some(Leaf {
+                        page: right_page,
+                        entry: right_entry,
+                        parent: _,
+                    }),
+                ) = (&self.left, &self.right)
+                {
+                    if left_page.get_page_number() == right_page.get_page_number()
+                        && left_entry >= right_entry
+                        && !self.include_left
+                    {
+                        return None;
+                    }
+                }
+
+                self.include_right = false;
+                if self.right.as_ref().unwrap().get_entry().is_some() {
+                    return self.right.as_ref().map(|s| s.get_entry().unwrap());
                 }
             }
         }
     }
 }
 
-pub(crate) fn find_iter_unbounded_start<'a>(
+fn find_iter_unbounded_left<'a>(
     page: PageImpl<'a>,
     mut parent: Option<Box<RangeIterState<'a>>>,
     manager: &'a TransactionalMemory,
@@ -339,7 +380,6 @@ pub(crate) fn find_iter_unbounded_start<'a>(
             page,
             entry: 0,
             parent,
-            reversed: false,
         }),
         INTERNAL => {
             let accessor = InternalAccessor::new(&page);
@@ -349,15 +389,14 @@ pub(crate) fn find_iter_unbounded_start<'a>(
                 page,
                 child: 1,
                 parent,
-                reversed: false,
             }));
-            find_iter_unbounded_start(child_page, parent, manager)
+            find_iter_unbounded_left(child_page, parent, manager)
         }
         _ => unreachable!(),
     }
 }
 
-pub(crate) fn find_iter_unbounded_reversed<'a>(
+fn find_iter_unbounded_right<'a>(
     page: PageImpl<'a>,
     mut parent: Option<Box<RangeIterState<'a>>>,
     manager: &'a TransactionalMemory,
@@ -371,7 +410,6 @@ pub(crate) fn find_iter_unbounded_reversed<'a>(
                 page,
                 entry,
                 parent,
-                reversed: true,
             })
         }
         INTERNAL => {
@@ -384,32 +422,32 @@ pub(crate) fn find_iter_unbounded_reversed<'a>(
                     page,
                     child: child_index - 1,
                     parent,
-                    reversed: true,
                 }));
             }
-            find_iter_unbounded_reversed(child_page, parent, manager)
+            find_iter_unbounded_right(child_page, parent, manager)
         }
         _ => unreachable!(),
     }
 }
 
-pub(crate) fn find_iter_start<'a, K: RedbKey + ?Sized>(
+fn find_iter_left<'a, K: RedbKey + ?Sized>(
     page: PageImpl<'a>,
     mut parent: Option<Box<RangeIterState<'a>>>,
     query: &[u8],
+    include_query: bool,
     manager: &'a TransactionalMemory,
-) -> Option<RangeIterState<'a>> {
+) -> (bool, Option<RangeIterState<'a>>) {
     let node_mem = page.memory();
     match node_mem[0] {
         LEAF => {
             let accessor = LeafAccessor::new(&page);
-            let (position, _) = accessor.position::<K>(query);
-            Some(Leaf {
+            let (position, found) = accessor.position::<K>(query);
+            let result = Leaf {
                 page,
                 entry: position,
                 parent,
-                reversed: false,
-            })
+            };
+            (include_query || !found, Some(result))
         }
         INTERNAL => {
             let accessor = InternalAccessor::new(&page);
@@ -420,32 +458,32 @@ pub(crate) fn find_iter_start<'a, K: RedbKey + ?Sized>(
                     page,
                     child: child_index + 1,
                     parent,
-                    reversed: false,
                 }));
             }
-            find_iter_start::<K>(child_page, parent, query, manager)
+            find_iter_left::<K>(child_page, parent, query, include_query, manager)
         }
         _ => unreachable!(),
     }
 }
 
-pub(crate) fn find_iter_start_reversed<'a, K: RedbKey + ?Sized>(
+fn find_iter_right<'a, K: RedbKey + ?Sized>(
     page: PageImpl<'a>,
     mut parent: Option<Box<RangeIterState<'a>>>,
     query: &[u8],
+    include_query: bool,
     manager: &'a TransactionalMemory,
-) -> Option<RangeIterState<'a>> {
+) -> (bool, Option<RangeIterState<'a>>) {
     let node_mem = page.memory();
     match node_mem[0] {
         LEAF => {
             let accessor = LeafAccessor::new(&page);
-            let (position, _) = accessor.position::<K>(query);
-            Some(Leaf {
+            let (position, found) = accessor.position::<K>(query);
+            let result = Leaf {
                 page,
                 entry: position,
                 parent,
-                reversed: true,
-            })
+            };
+            (include_query && found, Some(result))
         }
         INTERNAL => {
             let accessor = InternalAccessor::new(&page);
@@ -456,10 +494,9 @@ pub(crate) fn find_iter_start_reversed<'a, K: RedbKey + ?Sized>(
                     page,
                     child: child_index - 1,
                     parent,
-                    reversed: true,
                 }));
             }
-            find_iter_start_reversed::<K>(child_page, parent, query, manager)
+            find_iter_right::<K>(child_page, parent, query, include_query, manager)
         }
         _ => unreachable!(),
     }
diff --git a/src/tree_store/storage.rs b/src/tree_store/storage.rs
index e22c13b..abd5191 100644
--- a/src/tree_store/storage.rs
+++ b/src/tree_store/storage.rs
@@ -1,8 +1,5 @@
 use crate::tree_store::btree_base::AccessGuard;
-use crate::tree_store::btree_iters::{
-    find_iter_start, find_iter_start_reversed, find_iter_unbounded_reversed,
-    find_iter_unbounded_start, page_numbers_iter_start_state, AllPageNumbersBtreeIter,
-};
+use crate::tree_store::btree_iters::{page_numbers_iter_start_state, AllPageNumbersBtreeIter};
 use crate::tree_store::btree_utils::{
     find_key, fragmented_bytes, make_mut_single_leaf, overhead_bytes, print_tree, stored_bytes,
     tree_delete, tree_height, tree_insert,
@@ -14,10 +11,10 @@ use crate::Error;
 use memmap2::MmapRaw;
 use std::borrow::Borrow;
 use std::cmp::{max, min};
-use std::collections::{BTreeSet, Bound};
+use std::collections::BTreeSet;
 use std::convert::TryInto;
 use std::mem::size_of;
-use std::ops::{RangeBounds, RangeFull, RangeTo};
+use std::ops::{RangeBounds, RangeFull};
 use std::panic;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Mutex;
@@ -141,7 +138,7 @@ impl TableHeader {
 }
 
 pub struct TableNameIter<'a> {
-    inner: BtreeRangeIter<'a, RangeFull, &'static str, str, [u8]>,
+    inner: BtreeRangeIter<'a, str, [u8]>,
     table_type: TableType,
 }
 
@@ -182,9 +179,8 @@ impl Storage {
             let mut all_pages_iter: Box<dyn Iterator<Item = PageNumber>> =
                 Box::new(AllPageNumbersBtreeIter::new(start, &mem));
 
-            let start_state = find_iter_unbounded_start(mem.get_page(root), None, &mem);
-            let mut iter: BtreeRangeIter<RangeFull, [u8], [u8], [u8]> =
-                BtreeRangeIter::new(start_state, .., &mem);
+            let mut iter: BtreeRangeIter<[u8], [u8]> =
+                BtreeRangeIter::new::<RangeFull, [u8]>(.., Some(root), &mem);
 
             while let Some(entry) = iter.next() {
                 let definition = TableHeader::from_bytes(entry.value());
@@ -286,7 +282,7 @@ impl Storage {
         table_type: TableType,
         master_root_page: Option<PageNumber>,
     ) -> Result<TableNameIter, Error> {
-        let iter = self.get_range(.., master_root_page)?;
+        let iter = self.get_range::<RangeFull, str, str, [u8]>(.., master_root_page)?;
         Ok(TableNameIter {
             inner: iter,
             table_type,
@@ -431,8 +427,8 @@ impl Storage {
     }
 
     pub(crate) fn len(&self, root_page: Option<PageNumber>) -> Result<usize, Error> {
-        let mut iter: BtreeRangeIter<RangeFull, [u8], [u8], [u8]> =
-            self.get_range(.., root_page)?;
+        let mut iter: BtreeRangeIter<[u8], [u8]> =
+            self.get_range::<RangeFull, [u8], [u8], [u8]>(.., root_page)?;
         let mut count = 0;
         while iter.next().is_some() {
             count += 1;
@@ -532,7 +528,7 @@ impl Storage {
             .get_table(FREED_TABLE, TableType::Normal, master_root)?
             .unwrap();
         #[allow(clippy::type_complexity)]
-        let mut iter: BtreeRangeIter<'_, RangeTo<&[u8]>, &[u8], [u8], [u8]> =
+        let mut iter: BtreeRangeIter<'_, [u8], [u8]> =
             self.get_range(..lookup_key.as_ref(), freed_table.get_root())?;
         while let Some(entry) = iter.next() {
             to_remove.push(entry.key().to_vec());
@@ -673,8 +669,8 @@ impl Storage {
             .get_root_page_number()
             .map(|p| fragmented_bytes(self.mem.get_page(p), &self.mem))
             .unwrap_or(0);
-        let mut iter: BtreeRangeIter<RangeFull, [u8], [u8], [u8]> =
-            self.get_range(.., self.get_root_page_number())?;
+        let mut iter: BtreeRangeIter<[u8], [u8]> =
+            self.get_range::<RangeFull, [u8], [u8], [u8]>(.., self.get_root_page_number())?;
         while let Some(entry) = iter.next() {
             let definition = TableHeader::from_bytes(entry.value());
             if let Some(table_root) = definition.get_root() {
@@ -701,8 +697,9 @@ impl Storage {
             eprintln!("Master tree:");
             print_tree(self.mem.get_page(page), &self.mem);
 
-            let mut iter: BtreeRangeIter<RangeFull, [u8], [u8], [u8]> =
-                self.get_range(.., Some(page)).unwrap();
+            let mut iter: BtreeRangeIter<[u8], [u8]> = self
+                .get_range::<RangeFull, [u8], [u8], [u8]>(.., Some(page))
+                .unwrap();
 
             while let Some(entry) = iter.next() {
                 eprintln!("{} tree:", String::from_utf8_lossy(entry.key()));
@@ -741,60 +738,8 @@ impl Storage {
         &'a self,
         range: T,
         root_page: Option<PageNumber>,
-    ) -> Result<BtreeRangeIter<T, KR, K, V>, Error> {
-        if let Some(root) = root_page {
-            match range.start_bound() {
-                Bound::Included(k) | Bound::Excluded(k) => {
-                    let start_state = find_iter_start::<K>(
-                        self.mem.get_page(root),
-                        None,
-                        k.borrow().as_bytes().as_ref(),
-                        &self.mem,
-                    );
-                    Ok(BtreeRangeIter::new(start_state, range, &self.mem))
-                }
-                Bound::Unbounded => {
-                    let start_state =
-                        find_iter_unbounded_start(self.mem.get_page(root), None, &self.mem);
-                    Ok(BtreeRangeIter::new(start_state, range, &self.mem))
-                }
-            }
-        } else {
-            Ok(BtreeRangeIter::new(None, range, &self.mem))
-        }
-    }
-
-    pub(crate) fn get_range_reversed<
-        'a,
-        T: RangeBounds<KR>,
-        KR: Borrow<K> + ?Sized + 'a,
-        K: RedbKey + ?Sized + 'a,
-        V: RedbValue + ?Sized + 'a,
-    >(
-        &'a self,
-        range: T,
-        root_page: Option<PageNumber>,
-    ) -> Result<BtreeRangeIter<T, KR, K, V>, Error> {
-        if let Some(root) = root_page {
-            match range.end_bound() {
-                Bound::Included(k) | Bound::Excluded(k) => {
-                    let start_state = find_iter_start_reversed::<K>(
-                        self.mem.get_page(root),
-                        None,
-                        k.borrow().as_bytes().as_ref(),
-                        &self.mem,
-                    );
-                    Ok(BtreeRangeIter::new_reversed(start_state, range, &self.mem))
-                }
-                Bound::Unbounded => {
-                    let start_state =
-                        find_iter_unbounded_reversed(self.mem.get_page(root), None, &self.mem);
-                    Ok(BtreeRangeIter::new_reversed(start_state, range, &self.mem))
-                }
-            }
-        } else {
-            Ok(BtreeRangeIter::new_reversed(None, range, &self.mem))
-        }
+    ) -> Result<BtreeRangeIter<K, V>, Error> {
+        Ok(BtreeRangeIter::new(range, root_page, &self.mem))
     }
 
     // Returns the new root page, and a bool indicating whether the entry existed


Test patch:
diff --git a/tests/basic_tests.rs b/tests/basic_tests.rs
index a114597..5e39fd1 100644
--- a/tests/basic_tests.rs
+++ b/tests/basic_tests.rs
@@ -298,7 +298,7 @@ fn i128_type() {
     let read_txn = db.begin_read().unwrap();
     let table = read_txn.open_table(&definition).unwrap();
     assert_eq!(-2, table.get(&-1).unwrap().unwrap());
-    let mut iter: RangeIter<RangeFull, i128, i128, i128> = table.range(..).unwrap();
+    let mut iter: RangeIter<i128, i128> = table.range::<RangeFull, i128>(..).unwrap();
     for i in -11..10 {
         assert_eq!(iter.next().unwrap().1, i);
     }
@@ -339,16 +339,15 @@ fn str_type() {
     let hello = "hello".to_string();
     assert_eq!("world", table.get(&hello).unwrap().unwrap());
 
-    let mut iter: RangeIter<RangeFull, &str, str, str> = table.range(..).unwrap();
+    let mut iter: RangeIter<str, str> = table.range::<RangeFull, &str>(..).unwrap();
     assert_eq!(iter.next().unwrap().1, "world");
     assert!(iter.next().is_none());
 
-    let mut iter: RangeIter<Range<String>, String, str, str> =
-        table.range("a".to_string().."z".to_string()).unwrap();
+    let mut iter: RangeIter<str, str> = table.range("a".to_string().."z".to_string()).unwrap();
     assert_eq!(iter.next().unwrap().1, "world");
     assert!(iter.next().is_none());
 
-    let mut iter: RangeIter<Range<&str>, &str, str, str> = table.range("a".."z").unwrap();
+    let mut iter: RangeIter<str, str> = table.range("a".."z").unwrap();
     assert_eq!(iter.next().unwrap().1, "world");
     assert!(iter.next().is_none());
 }
@@ -372,17 +371,18 @@ fn owned_get_signatures() {
 
     assert_eq!(2, table.get(&1).unwrap().unwrap());
 
-    let mut iter: RangeIter<RangeFull, u32, u32, u32> = table.range(..).unwrap();
+    let mut iter: RangeIter<u32, u32> = table.range::<RangeFull, u32>(..).unwrap();
     for i in 0..10 {
         assert_eq!(iter.next().unwrap().1, i + 1);
     }
     assert!(iter.next().is_none());
-    let mut iter: RangeIter<Range<u32>, u32, u32, u32> = table.range(0..10).unwrap();
+    let mut iter: RangeIter<u32, u32> = table.range(0..10).unwrap();
     for i in 0..10 {
         assert_eq!(iter.next().unwrap().1, i + 1);
     }
     assert!(iter.next().is_none());
-    let mut iter: RangeIter<Range<&u32>, u32, u32, u32> = table.range(&0..&10).unwrap();
+    // TODO: is there a way to remove the type hint on range()?
+    let mut iter: RangeIter<u32, u32> = table.range::<Range<&u32>, &u32>(&0..&10).unwrap();
     for i in 0..10 {
         assert_eq!(iter.next().unwrap().1, i + 1);
     }
@@ -410,14 +410,14 @@ fn ref_get_signatures() {
 
     let start = vec![0u8];
     let end = vec![10u8];
-    let mut iter: RangeIter<RangeFull, &[u8], [u8], [u8]> = table.range(..).unwrap();
+    // TODO: is there a way to make RangeFull queries work without the type hint?
+    let mut iter: RangeIter<[u8], [u8]> = table.range::<RangeFull, &[u8]>(..).unwrap();
     for i in 0..10 {
         assert_eq!(iter.next().unwrap().1, &[i + 1]);
     }
     assert!(iter.next().is_none());
 
-    let mut iter: RangeIter<Range<&[u8]>, &[u8], [u8], [u8]> =
-        table.range(start.as_slice()..&end).unwrap();
+    let mut iter: RangeIter<[u8], [u8]> = table.range(start.as_slice()..&end).unwrap();
     for i in 0..10 {
         assert_eq!(iter.next().unwrap().1, &[i + 1]);
     }
diff --git a/tests/integration_tests.rs b/tests/integration_tests.rs
index fafdb93..740d93f 100644
--- a/tests/integration_tests.rs
+++ b/tests/integration_tests.rs
@@ -441,22 +441,27 @@ fn range_query() {
     let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
     let db = unsafe { Database::create(tmpfile.path(), 1024 * 1024).unwrap() };
     let write_txn = db.begin_write().unwrap();
-    let mut table = write_txn.open_table(&SLICE_TABLE).unwrap();
-    for i in 0..10u8 {
-        let key = vec![i];
-        table.insert(&key, b"value").unwrap();
+    let mut table = write_txn.open_table(&U64_TABLE).unwrap();
+    for i in 0..10 {
+        table.insert(&i, &i).unwrap();
     }
     write_txn.commit().unwrap();
 
     let read_txn = db.begin_read().unwrap();
-    let table = read_txn.open_table(&SLICE_TABLE).unwrap();
-    let start = vec![3u8];
-    let end = vec![7u8];
-    let mut iter = table.range(start..end).unwrap();
-    for i in 3..7u8 {
+    let table = read_txn.open_table(&U64_TABLE).unwrap();
+    let mut iter = table.range(3..7).unwrap();
+    for i in 3..7u64 {
         let (key, value) = iter.next().unwrap();
-        assert_eq!(&[i], key);
-        assert_eq!(b"value", value);
+        assert_eq!(i, key);
+        assert_eq!(i, value);
+    }
+    assert!(iter.next().is_none());
+
+    let mut iter = table.range(3..=7).unwrap();
+    for i in 3..=7u64 {
+        let (key, value) = iter.next().unwrap();
+        assert_eq!(i, key);
+        assert_eq!(i, value);
     }
     assert!(iter.next().is_none());
 }
@@ -466,26 +471,38 @@ fn range_query_reversed() {
     let tmpfile: NamedTempFile = NamedTempFile::new().unwrap();
     let db = unsafe { Database::create(tmpfile.path(), 1024 * 1024).unwrap() };
     let write_txn = db.begin_write().unwrap();
-    let mut table = write_txn.open_table(&SLICE_TABLE).unwrap();
-    for i in 0..10u8 {
-        let key = vec![i];
-        table.insert(&key, b"value").unwrap();
+    let mut table = write_txn.open_table(&U64_TABLE).unwrap();
+    for i in 0..10 {
+        table.insert(&i, &i).unwrap();
     }
     write_txn.commit().unwrap();
 
     let read_txn = db.begin_read().unwrap();
-    let table = read_txn.open_table(&SLICE_TABLE).unwrap();
-    let start = vec![3u8];
-    let end = vec![7u8];
-    let mut iter = table
-        .range_reversed(start.as_slice()..end.as_slice())
-        .unwrap();
-    for i in (3..7u8).rev() {
+    let table = read_txn.open_table(&U64_TABLE).unwrap();
+    let mut iter = table.range(3..7).unwrap().rev();
+    for i in (3..7u64).rev() {
         let (key, value) = iter.next().unwrap();
-        assert_eq!(&[i], key);
-        assert_eq!(b"value", value);
+        assert_eq!(i, key);
+        assert_eq!(i, value);
     }
     assert!(iter.next().is_none());
+
+    // Test reversing multiple times
+    let mut iter = table.range(3..7).unwrap();
+    let (key, _) = iter.next().unwrap();
+    assert_eq!(3, key);
+
+    iter = iter.rev();
+    let (key, _) = iter.next().unwrap();
+    assert_eq!(6, key);
+    let (key, _) = iter.next().unwrap();
+    assert_eq!(5, key);
+
+    iter = iter.rev();
+    let (key, _) = iter.next().unwrap();
+    assert_eq!(4, key);
+
+    assert!(iter.next().is_none());
 }
 
 #[test]