add key abstraction and prepare for MVCC (#28)

* add key abstraction and prepare for MVCC

Signed-off-by: Alex Chi <iskyzh@gmail.com>

* a little bit type exercise

Signed-off-by: Alex Chi <iskyzh@gmail.com>

* refactor tests

Signed-off-by: Alex Chi <iskyzh@gmail.com>

* fix clippy warnings

Signed-off-by: Alex Chi <iskyzh@gmail.com>

* refactor starter code

Signed-off-by: Alex Chi <iskyzh@gmail.com>

* final touch docs

Signed-off-by: Alex Chi <iskyzh@gmail.com>

---------

Signed-off-by: Alex Chi <iskyzh@gmail.com>

mini-lsm/src/iterators/concat_iterator.rs

@@ -2,7 +2,10 @@ use std::sync::Arc;
 
 use anyhow::Result;
 
-use crate::table::{SsTable, SsTableIterator};
+use crate::{
+    key::KeySlice,
+    table::{SsTable, SsTableIterator},
+};
 
 use super::StorageIterator;
 
@@ -46,10 +49,10 @@ impl SstConcatIterator {
         Ok(iter)
     }
 
-    pub fn create_and_seek_to_key(sstables: Vec<Arc<SsTable>>, key: &[u8]) -> Result<Self> {
+    pub fn create_and_seek_to_key(sstables: Vec<Arc<SsTable>>, key: KeySlice) -> Result<Self> {
         Self::check_sst_valid(&sstables);
         let idx: usize = sstables
-            .partition_point(|table| table.first_key() <= key)
+            .partition_point(|table| table.first_key().as_key_slice() <= key)
             .saturating_sub(1);
         if idx >= sstables.len() {
             return Ok(Self {
@@ -89,7 +92,9 @@ impl SstConcatIterator {
 }
 
 impl StorageIterator for SstConcatIterator {
-    fn key(&self) -> &[u8] {
+    type KeyType<'a> = KeySlice<'a>;
+
+    fn key(&self) -> KeySlice {
         self.current.as_ref().unwrap().key()
     }
 

mini-lsm/src/iterators/merge_iterator.rs

@@ -4,6 +4,8 @@ use std::collections::BinaryHeap;
 
 use anyhow::Result;
 
+use crate::key::KeySlice;
+
 use super::StorageIterator;
 
 struct HeapWrapper<I: StorageIterator>(pub usize, pub Box<I>);
@@ -19,7 +21,7 @@ impl<I: StorageIterator> Eq for HeapWrapper<I> {}
 impl<I: StorageIterator> PartialOrd for HeapWrapper<I> {
     #[allow(clippy::non_canonical_partial_ord_impl)]
     fn partial_cmp(&self, other: &Self) -> Option<cmp::Ordering> {
-        match self.1.key().cmp(other.1.key()) {
+        match self.1.key().cmp(&other.1.key()) {
             cmp::Ordering::Greater => Some(cmp::Ordering::Greater),
             cmp::Ordering::Less => Some(cmp::Ordering::Less),
             cmp::Ordering::Equal => self.0.partial_cmp(&other.0),
@@ -75,8 +77,12 @@ impl<I: StorageIterator> MergeIterator<I> {
     }
 }
 
-impl<I: StorageIterator> StorageIterator for MergeIterator<I> {
-    fn key(&self) -> &[u8] {
+impl<I: 'static + for<'a> StorageIterator<KeyType<'a> = KeySlice<'a>>> StorageIterator
+    for MergeIterator<I>
+{
+    type KeyType<'a> = KeySlice<'a>;
+
+    fn key(&self) -> KeySlice {
         self.current.as_ref().unwrap().1.key()
     }
 

mini-lsm/src/iterators/two_merge_iterator.rs

@@ -1,5 +1,7 @@
 use anyhow::Result;
 
+use crate::key::KeySlice;
+
 use super::StorageIterator;
 
 /// Merges two iterators of different types into one. If the two iterators have the same key, only
@@ -10,7 +12,11 @@ pub struct TwoMergeIterator<A: StorageIterator, B: StorageIterator> {
     choose_a: bool,
 }
 
-impl<A: StorageIterator, B: StorageIterator> TwoMergeIterator<A, B> {
+impl<
+        A: 'static + for<'a> StorageIterator<KeyType<'a> = KeySlice<'a>>,
+        B: 'static + for<'a> StorageIterator<KeyType<'a> = KeySlice<'a>>,
+    > TwoMergeIterator<A, B>
+{
     fn choose_a(a: &A, b: &B) -> bool {
         if !a.is_valid() {
             return false;
@@ -40,8 +46,14 @@ impl<A: StorageIterator, B: StorageIterator> TwoMergeIterator<A, B> {
     }
 }
 
-impl<A: StorageIterator, B: StorageIterator> StorageIterator for TwoMergeIterator<A, B> {
-    fn key(&self) -> &[u8] {
+impl<
+        A: 'static + for<'a> StorageIterator<KeyType<'a> = KeySlice<'a>>,
+        B: 'static + for<'a> StorageIterator<KeyType<'a> = KeySlice<'a>>,
+    > StorageIterator for TwoMergeIterator<A, B>
+{
+    type KeyType<'a> = KeySlice<'a>;
+
+    fn key(&self) -> KeySlice {
         if self.choose_a {
             self.a.key()
         } else {