test: add tests

src/block/builder.rs

@@ -103,3 +103,151 @@ fn compress_key(first_key: &KeyVec, key: KeySlice, buffer: &mut Vec<u8>) {
         buffer.extend_from_slice(&key[common_prefix..]);
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::block::{Block, BlockBuilder, BlockIterator};
+    use crate::key::{KeySlice, KeyVec};
+    use bytes::Bytes;
+    use nom::AsBytes;
+    use std::sync::Arc;
+
+    #[test]
+    fn test_block_build_single_key() {
+        let mut builder = BlockBuilder::new(16);
+        assert!(builder.add(KeySlice::for_testing_from_slice_no_ts(b"233"), b"233333"));
+        builder.build();
+    }
+
+    #[test]
+    fn test_block_build_full() {
+        let mut builder = BlockBuilder::new(16);
+        assert!(builder.add(KeySlice::for_testing_from_slice_no_ts(b"11"), b"11"));
+        assert!(!builder.add(KeySlice::for_testing_from_slice_no_ts(b"22"), b"22"));
+        builder.build();
+    }
+
+    #[test]
+    fn test_block_build_large_1() {
+        let mut builder = BlockBuilder::new(16);
+        assert!(builder.add(
+            KeySlice::for_testing_from_slice_no_ts(b"11"),
+            &b"1".repeat(100)
+        ));
+        builder.build();
+    }
+
+    #[test]
+    fn test_block_build_large_2() {
+        let mut builder = BlockBuilder::new(16);
+        assert!(builder.add(KeySlice::for_testing_from_slice_no_ts(b"11"), b"1"));
+        assert!(!builder.add(
+            KeySlice::for_testing_from_slice_no_ts(b"11"),
+            &b"1".repeat(100)
+        ));
+    }
+
+    fn key_of(idx: usize) -> KeyVec {
+        KeyVec::for_testing_from_vec_no_ts(format!("key_{:03}", idx * 5).into_bytes())
+    }
+
+    fn value_of(idx: usize) -> Vec<u8> {
+        format!("value_{:010}", idx).into_bytes()
+    }
+
+    fn num_of_keys() -> usize {
+        100
+    }
+
+    fn generate_block() -> Block {
+        let mut builder = BlockBuilder::new(10000);
+        for idx in 0..num_of_keys() {
+            let key = key_of(idx);
+            let value = value_of(idx);
+            assert!(builder.add(key.as_key_slice(), &value[..]));
+        }
+        builder.build()
+    }
+
+    #[test]
+    fn test_block_build_all() {
+        generate_block();
+    }
+
+    #[test]
+    fn test_block_encode() {
+        let block = generate_block();
+        block.encode();
+    }
+
+    #[test]
+    fn test_block_decode() {
+        let block = generate_block();
+        let encoded = block.encode();
+        let decoded_block = Block::decode(&encoded);
+        assert_eq!(block.offsets, decoded_block.offsets);
+        assert_eq!(block.data, decoded_block.data);
+    }
+
+    fn as_bytes(x: &[u8]) -> Bytes {
+        Bytes::copy_from_slice(x)
+    }
+
+    #[test]
+    fn test_block_iterator() {
+        let block = Arc::new(generate_block());
+        for _ in 0..5 {
+            let mut iter = BlockIterator::create_and_seek_to_first(block.clone());
+            for i in 0..num_of_keys() {
+                let entry = iter.next().unwrap().unwrap();
+                let key = entry.key.as_bytes();
+                let value = entry.value.as_bytes();
+                assert_eq!(
+                    key,
+                    key_of(i).for_testing_key_ref(),
+                    "expected key: {:?}, actual key: {:?}",
+                    as_bytes(key_of(i).for_testing_key_ref()),
+                    as_bytes(key)
+                );
+                assert_eq!(
+                    value,
+                    value_of(i),
+                    "expected value: {:?}, actual value: {:?}",
+                    as_bytes(&value_of(i)),
+                    as_bytes(value)
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn test_block_seek_key() {
+        let block = Arc::new(generate_block());
+        let mut iter = BlockIterator::create_and_seek_to_key(block, key_of(0).as_key_slice());
+        for offset in 1..=5 {
+            for i in 0..num_of_keys() {
+                let entry = iter.next().unwrap().unwrap();
+                let key = entry.key.as_bytes();
+                let value = entry.value.as_bytes();
+                assert_eq!(
+                    key,
+                    key_of(i).for_testing_key_ref(),
+                    "expected key: {:?}, actual key: {:?}",
+                    as_bytes(key_of(i).for_testing_key_ref()),
+                    as_bytes(key)
+                );
+                assert_eq!(
+                    value,
+                    value_of(i),
+                    "expected value: {:?}, actual value: {:?}",
+                    as_bytes(&value_of(i)),
+                    as_bytes(value)
+                );
+                iter.seek_to_key(KeySlice::for_testing_from_slice_no_ts(
+                    &format!("key_{:03}", i * 5 + offset).into_bytes(),
+                ));
+            }
+            iter.seek_to_key(KeySlice::for_testing_from_slice_no_ts(b"k"));
+        }
+    }
+}

src/iterators/lsm.rs

@@ -29,7 +29,7 @@ type LsmIteratorInner<'a, File> = TwoMergeIterator<
 
 #[derive(new)]
 pub struct LockedLsmIter<'a, P: Persistent> {
-    state: &'a LsmStorageStateInner<P>,
+    state: arc_swap::Guard<Arc<LsmStorageStateInner<P>>>,
     lower: Bound<&'a [u8]>,
     upper: Bound<&'a [u8]>,
 }

src/iterators/merge.rs

@@ -107,11 +107,14 @@ where
 #[cfg(test)]
 mod test {
     use std::fmt::Debug;
+    use std::vec;
 
     use futures::{stream, StreamExt};
 
     use crate::entry::Entry;
     use crate::iterators::merge::MergeIteratorInner;
+    use crate::iterators::utils::build_stream;
+    use crate::iterators::{create_merge_iter, eq};
 
     #[tokio::test]
     async fn test_empty() {
@@ -164,4 +167,53 @@ mod test {
         let expect: Vec<_> = expect.into_iter().collect();
         assert_eq!(expect, merged);
     }
+
+    #[tokio::test]
+    async fn test_task2_merge_1() {
+        let build_sub_stream = || {
+            let i1 = build_stream([("a", "1.1"), ("b", "2.1"), ("c", "3.1"), ("e", "")]).map(Ok);
+            let i2 = build_stream([("a", "1.2"), ("b", "2.2"), ("c", "3.2"), ("d", "4.2")]).map(Ok);
+            let i3 = build_stream([("b", "2.3"), ("c", "3.3"), ("d", "4.3")]).map(Ok);
+            (i1, i2, i3)
+        };
+
+        let (i1, i2, i3) = build_sub_stream();
+
+        assert!(
+            eq(
+                create_merge_iter(stream::iter([i1, i2, i3]))
+                    .await
+                    .map(Result::unwrap),
+                build_stream([
+                    ("a", "1.1"),
+                    ("b", "2.1"),
+                    ("c", "3.1"),
+                    ("d", "4.2"),
+                    ("e", ""),
+                ]),
+            )
+            .await
+        );
+
+        let (i1, i2, i3) = build_sub_stream();
+
+        assert!(
+            eq(
+                create_merge_iter(stream::iter([i3, i1, i2]))
+                    .await
+                    .map(Result::unwrap),
+                build_stream([
+                    ("a", "1.1"),
+                    ("b", "2.3"),
+                    ("c", "3.3"),
+                    ("d", "4.3"),
+                    ("e", ""),
+                ]),
+            )
+            .await
+        );
+    }
+
+    #[tokio::test]
+    async fn test_task2_merge_2() {}
 }

src/iterators/mod.rs

@@ -15,4 +15,7 @@ pub use ok_iter::OkIter;
 pub use two_merge::{create_two_merge_iter, TwoMergeIterator};
 pub use utils::iter_fut_iter_to_stream;
 pub use utils::split_first;
-pub use utils::{iter_fut_to_stream, transpose_try_iter};
+pub use utils::{eq, iter_fut_to_stream, transpose_try_iter};
+
+#[cfg(test)]
+pub use utils::build_stream;

src/iterators/utils.rs

@@ -1,12 +1,14 @@
 use std::future::Future;
-use std::iter;
 use std::iter::Map;
 use std::iter::Once;
+use std::pin::pin;
+use std::{iter, vec};
 
+use crate::entry::Entry;
 use either::Either;
 use futures::future::IntoStream;
 use futures::stream::{FlatMap, Flatten, Iter};
-use futures::{stream, FutureExt, StreamExt};
+use futures::{stream, FutureExt, Stream, StreamExt};
 
 pub fn transpose_try_iter<I, T, E>(iterator: Result<I, E>) -> Either<I, Once<Result<T, E>>>
 where
@@ -53,3 +55,57 @@ where
 {
     stream::iter(iterator.map(FutureExt::into_stream as fn(_) -> _)).flatten()
 }
+
+pub async fn eq<S1, S2>(s1: S1, s2: S2) -> bool
+where
+    S1: Stream,
+    S2: Stream,
+    S1::Item: PartialEq<S2::Item>,
+{
+    let mut s1 = pin!(s1);
+    let mut s2 = pin!(s2);
+    loop {
+        match (s1.next().await, s2.next().await) {
+            (Some(x1), Some(x2)) => {
+                if x1 != x2 {
+                    return false;
+                }
+            }
+            (Some(_), None) | (None, Some(_)) => return false,
+            (None, None) => return true,
+        }
+    }
+}
+
+#[cfg(test)]
+pub type EntryStream = Iter<vec::IntoIter<Entry>>;
+
+#[cfg(test)]
+pub fn build_stream<'a>(source: impl IntoIterator<Item = (&'a str, &'a str)>) -> EntryStream {
+    let s: Vec<_> = source
+        .into_iter()
+        .map(|(key, value)| Entry::from_slice(key.as_bytes(), value.as_bytes()))
+        .collect();
+    stream::iter(s)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::eq;
+    use futures::stream;
+
+    #[tokio::test]
+    async fn test_eq() {
+        assert!(iter_eq([1], [1],).await);
+        assert!(!iter_eq([1], [2],).await);
+        assert!(iter_eq([] as [i32; 0], [],).await);
+        assert!(!iter_eq([1, 2, 3], [1, 2],).await);
+    }
+
+    async fn iter_eq<T: PartialEq>(
+        s1: impl IntoIterator<Item = T>,
+        s2: impl IntoIterator<Item = T>,
+    ) -> bool {
+        eq(stream::iter(s1), stream::iter(s2)).await
+    }
+}

src/memtable/iterator.rs

@@ -35,11 +35,15 @@ pub type MaybeEmptyMemTableIterRef<'a> = MaybeEmptyStream<Entry, MemTableIterato
 #[cfg(test)]
 mod test {
     use crate::entry::Entry;
-    use crate::iterators::create_merge_iter_from_non_empty_iters;
+    use crate::iterators::{
+        create_merge_iter, create_merge_iter_from_non_empty_iters, eq, MergeIterator,
+    };
     use crate::memtable::MemTable;
+    use bytes::Bytes;
     use futures::{stream, Stream, StreamExt};
     use nom::AsBytes;
     use std::collections::Bound;
+    use std::vec;
 
     #[tokio::test]
     async fn test_task1_memtable_iter() {
@@ -102,6 +106,25 @@ mod test {
         }
     }
 
+    #[tokio::test]
+    async fn test_task1_empty_memtable_iter() {
+        use std::ops::Bound;
+        let memtable = MemTable::create(0);
+
+        let _ = memtable
+            .for_testing_scan_slice(Bound::Excluded(b"key1"), Bound::Excluded(b"key3"))
+            .await
+            .unwrap();
+        let _ = memtable
+            .for_testing_scan_slice(Bound::Included(b"key1"), Bound::Included(b"key2"))
+            .await
+            .unwrap();
+        let _ = memtable
+            .for_testing_scan_slice(Bound::Unbounded, Bound::Unbounded)
+            .await
+            .unwrap();
+    }
+
     async fn get_memtable_iter<'a>(
         memtable: &'a MemTable,
         lower: Bound<&'a [u8]>,