feat: in-memory index

storage_engine/src/error.rs

@@ -15,6 +15,7 @@ pub enum DecodeErrorKind {
     EntryData,
     EntryIsDeleted,
     EntryHeaderWithDataSizes,
+    CorruptedData,
 }
 
 // ===Errors===

storage_engine/src/index.rs

@@ -1,18 +1,20 @@
 use std::marker::PhantomData;
-use tokio::io::{AsyncReadExt, AsyncWriteExt, AsyncSeekExt, SeekFrom};
-use tokio::fs::{File, OpenOptions, DirBuilder};
 use std::path::Path;
+use tokio::fs::{DirBuilder, File, OpenOptions};
+use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt, SeekFrom};
 
-use std::collections::{BTreeMap};
 use async_trait::async_trait;
+use std::collections::{BTreeMap, HashSet};
+use std::hash::Hash;
 
+use crate::binary_coding::{decode, decode_sequence, encode, encode_sequence};
 use bincode;
 use bincode::{Decode, Encode};
-use crate::binary_coding::{encode, decode, encode_sequence, decode_sequence};
 use tokio::fs;
 
-use crate::error::Error;
+use crate::error::{DecodeErrorKind, Error};
 
+use bincode::error::DecodeError;
 use std::mem::size_of;
 
 type Result<T> = std::result::Result<T, Error>;
@@ -23,67 +25,126 @@ type Result<T> = std::result::Result<T, Error>;
 #[derive(Debug)]
 pub struct Index<K, V> {
     file: File,
-    // None means index is asleep on disk.
+    data: BTreeMap<K, HashSet<V>>,
-    in_memory: Option<BTreeMap<K, V>>,
-    header: IndexHeader,
     key_type: PhantomData<K>,
     value_type: PhantomData<V>,
 }
 
 #[derive(Debug)]
-pub struct IndexHeader {
+pub struct IndexHeader {}
-}
 
-use crate::storage_engine::FilePosition;
+impl<K, V> Index<K, V>
+where
+    K: Encode + Decode + Ord,
+    V: Encode + Decode + Clone + Eq + Hash,
+{
+    pub async fn new(file_name: &str) -> Result<Index<K, V>> {
+        let file: File = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .create(true)
+            .open(file_name)
+            .await?;
 
-#[async_trait]
+        let data = BTreeMap::new();
-pub trait SomethingSupportingLeq {
-    async fn less_than_eq(&mut self, file_position0: FilePosition, file_position1: FilePosition) -> std::result::Result<bool, Error>;
-}
 
-impl <K, V>Index<K, V> {
+        Ok(Index {
-    // TODO: delete
+            file,
-    // pub async fn new<F, Fut, Store>(file_name: &str, less_than_eq: &F) -> Result<Index<K, V>> 
+            data,
-    //     where F: Fn(&mut Store, K, K) -> Fut,
+            key_type: PhantomData::<K>,
-    //           Store: SomethingSupportingLeq,
+            value_type: PhantomData::<V>,
-    //           Fut: Future<Output=std::result::Result<bool, Error>>,
+        })
-    // {
-    //     todo!()
-    // }
-    pub async fn new(file_name: &str) -> Result<Index<K, V>> 
-    {
-        todo!()
     }
 
     pub async fn connect(file_name: &str) -> Result<Index<K, V>> {
-        todo!()
+        let mut file: File = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .create(true)
+            .open(file_name)
+            .await?;
+
+        let mut bytes = vec![];
+        file.read_to_end(&mut bytes).await?;
+
+        let data = Index::decode_tree(&bytes)
+            .map_err(|e| Error::DecodeError(DecodeErrorKind::CorruptedData, e))?;
+
+        Ok(Index {
+            file,
+            data,
+            key_type: PhantomData::<K>,
+            value_type: PhantomData::<V>,
+        })
     }
 
-    // Saves the in-memory index to disk and deallocates.
+    pub async fn insert(&mut self, k: K, v: V) -> Result<()> {
-    pub async fn sleep() -> Result<Index<K, V>> {
+        self.data.entry(k).or_insert_with(HashSet::new).insert(v);
-        todo!()
+        Ok(())
     }
 
-    // Loads the index into memory
+    pub async fn lookup(&self, k: K) -> Result<Option<HashSet<V>>> {
-    pub async fn wake() -> Result<Index<K, V>> {
+        let hashset = self.data.get(&k).unwrap();
-        todo!()
+        Ok(Some(hashset.clone()))
     }
 
-    pub async fn insert() -> Result<()>
+    pub async fn delete(&mut self, k: K, v: V) -> Result<Option<bool>> {
-        where K: Encode, V: Encode
+        Ok(Some(
-    {
+            self.data.entry(k).or_insert_with(HashSet::new).remove(&v),
-        todo!()
+        ))
     }
 
-    pub async fn lookup<Store>(&mut self, store: &mut Store, k: K) -> Result<Option<V>> 
+    fn encode(&self) -> Result<Vec<u8>> {
-        where K: Encode + Decode,
+        let mut encoded = Vec::new();
-              Store: SomethingSupportingLeq,
+        encoded.extend(encode(&self.data)?);
-    {
+        Ok(encoded)
-        let x = store.less_than_eq(123, 123).await?;
-        todo!()
     }
 
-    pub async fn delete(&mut self, k: K) -> Result<Option<V>> {
+    fn decode_tree(data: &[u8]) -> std::result::Result<BTreeMap<K, HashSet<V>>, DecodeError> {
-        todo!()
+        let data: BTreeMap<K, HashSet<V>> = decode(data)?.0;
+        Ok(data)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn encode_decode() {
+        let mut index: Index<String, u32> = Index {
+            file: File::from_std(std::fs::File::create("test").unwrap()),
+            data: BTreeMap::new(),
+            key_type: PhantomData::<String>,
+            value_type: PhantomData::<u32>,
+        };
+
+        index.insert("foo".to_string(), 123).await.unwrap();
+        index.insert("foo".to_string(), 124).await.unwrap();
+        index.insert("bar".to_string(), 125).await.unwrap();
+        index.insert("bar".to_string(), 126).await.unwrap();
+
+        let lookup = index.lookup("foo".to_string()).await.unwrap().unwrap();
+        assert_eq!(lookup.len(), 2);
+        assert!(lookup.contains(&123));
+        assert!(lookup.contains(&124));
+        println!("lookup {:?}", lookup);
+
+        let encoded = index.encode().unwrap();
+        let decoded = Index::<String, u32>::decode_tree(&encoded).unwrap();
+        let decoded = Index {
+            file: File::from_std(std::fs::File::create("test").unwrap()),
+            data: decoded,
+            key_type: PhantomData::<String>,
+            value_type: PhantomData::<u32>,
+        };
+
+        let lookup = decoded.lookup("foo".to_string()).await.unwrap().unwrap();
+        assert_eq!(lookup.len(), 2);
+        assert!(lookup.contains(&123));
+        assert!(lookup.contains(&124));
+        println!("lookup {:?}", lookup);
+
+        println!("{encoded:?}")
     }
 }

storage_engine/src/storage_engine.rs

@@ -53,15 +53,6 @@ pub async fn less_than_eq<T>(store: &mut Store<T>, file_position0: FilePosition,
     todo!()
 }
 
-#[async_trait]
-impl <T>SomethingSupportingLeq for Store<T> 
-    where T: Send
-{
-    async fn less_than_eq(&mut self, file_position0: FilePosition, file_position1: FilePosition) -> std::result::Result<bool, Error> {
-        Ok(true)
-    }
-}
-
 pub const ROWS_FILE_NAME: &'static str = "rows";
 
 impl <T>Store<T> {