diff --git a/storage_engine/src/binary_coding.rs b/storage_engine/src/binary_coding.rs index 5e6378d..4e66623 100644 --- a/storage_engine/src/binary_coding.rs +++ b/storage_engine/src/binary_coding.rs @@ -69,11 +69,12 @@ pub fn decode_sequence(len: usize, bytes: &[u8]) -> Result, bi } +#[allow(dead_code)] fn example_encoding_decoding() { - let xs: Vec = vec![123, 250, 256, 123, 123, 123]; - let xs: Vec = vec![]; - let xs: Vec = vec![123]; - let xs: Vec = vec![123, 250]; + let _xs: Vec = vec![123, 250, 256, 123, 123, 123]; + let _xs: Vec = vec![]; + let _xs: Vec = vec![123]; + let _xs: Vec = vec![123, 250]; let xs: Vec = vec!["foo".to_string(), "bar".to_string()]; @@ -87,4 +88,3 @@ fn example_encoding_decoding() { let dxs = decode_vector::(&exs[..]).unwrap(); println!("decoded {:?}", dxs); } - diff --git a/storage_engine/src/cursor.rs b/storage_engine/src/cursor.rs index 8b6bfbb..5acbfbe 100644 --- a/storage_engine/src/cursor.rs +++ b/storage_engine/src/cursor.rs @@ -49,8 +49,7 @@ pub struct AppendOnlyCursor { // ===Traits=== #[async_trait] -// TODO: Make this private -pub trait PrimitiveCursor { +pub(crate) trait PrimitiveCursor { fn file(&mut self) -> &mut File; fn eof_file_position(&self) -> FilePosition; @@ -100,7 +99,7 @@ pub trait PrimitiveCursor { } #[async_trait] -pub trait PrimitiveWriteCursor: PrimitiveCursor { +pub(crate) trait PrimitiveWriteCursor: PrimitiveCursor { async fn write_bytes(&mut self, bytes: &[u8]) -> Result { Ok(self.file().write(bytes).await?) } @@ -536,8 +535,6 @@ impl <'cursor, T> ReadCursor<'cursor, T> { impl <'cursor, T> WriteCursor<'cursor, T> -// TODO: Consider adding this manually to wher eit is really needed - where T: Sync { // 'store lives atleast as long as 'cursor pub async fn new<'store: 'cursor>(store: &'store mut Store) -> Result @@ -566,31 +563,6 @@ impl <'cursor, T> WriteCursor<'cursor, T> Ok(cursor) } - pub async fn connect<'header: 'cursor, 'indexes: 'cursor>(path_to_rows: &str, header: &'header mut StoreHeader, indexes: &'indexes mut StoreIndexes) -> Result - where T: Send - { - let file: File = - OpenOptions::new() - .read(true) - .write(true) - .open(path_to_rows) - .await?; - - let mut cursor = Self { - header, - file, - indexes, - - eof_file_position: 0, - }; - let eof_file_position: FilePosition = cursor.seek_to_end().await?; - cursor.eof_file_position = eof_file_position; - - cursor.seek_to_start_of_data().await?; - - Ok(cursor) - } - // ===Entry Header Manipulation=== // assumes we are at the start of valid entry. async fn set_new_entry_header(&mut self, entry_header: EntryHeader) -> Result<()> @@ -707,7 +679,7 @@ impl <'cursor, T> WriteCursor<'cursor, T> // ===Garbage Collection=== async fn attempt_garbage_collection_if_necessary(&mut self) -> Result<()> - where T: Send + Decode + Encode + Clone + Ord + where T: Send + Sync + Decode + Encode + Clone + Ord { if self.header.deleted_count > GARBAGE_COLLECTION_TRIGGER { println!("=======START GARBAGE COLLETOR===="); @@ -717,7 +689,7 @@ impl <'cursor, T> WriteCursor<'cursor, T> } pub async fn initiate_garbage_collection(&mut self) -> Result - where T: Send + Decode + Encode + Clone + Ord + where T: Send + Sync + Decode + Encode + Clone + Ord { let mut cursor_to_intermediate = self.spawn_cursor_to_intermediate_file().await?; // Since garbage collection changes FilePositions of live entries, we need to update the diff --git a/storage_engine/src/error.rs b/storage_engine/src/error.rs index 4adf004..47bf5cc 100644 --- a/storage_engine/src/error.rs +++ b/storage_engine/src/error.rs @@ -8,7 +8,6 @@ pub enum Error { IndexIsStoringEofFilePosition(Column), ColumnAlreadyIndexed(Column), IoError(std::io::Error), - InvalidStoreHeader, } #[derive(Debug)] @@ -49,5 +48,3 @@ impl From for Error { Self::IoError(err) } } - - diff --git a/storage_engine/src/main.rs b/storage_engine/src/main.rs index f3a32e0..4681cb2 100644 --- a/storage_engine/src/main.rs +++ b/storage_engine/src/main.rs @@ -9,7 +9,7 @@ mod store_header; use crate::entry::{Entry, EntryDetailed}; use crate::storage_engine::{Store, FilePosition}; -use crate::cursor::{ReadCursor, WriteCursor, CursorWithStoreHeader, CursorWithWriteStoreHeader, CursorWithWriteAccessToIndex, CursorWithAccessToIndex}; +use crate::cursor::{ReadCursor, WriteCursor, CursorWithStoreHeader, CursorWithWriteAccessToIndex}; type Data = u32; @@ -18,7 +18,7 @@ const TABLE_PATH: &'static str = "test_table"; type Result = std::result::Result; async fn create_store() -> Result> { - let mut store: Store = Store::new(TABLE_PATH, 5, 0).await.map_err(|e| e.to_io_or_panic())?; + let store: Store = Store::new(TABLE_PATH, 5, 0).await.map_err(|e| e.to_io_or_panic())?; println!("CREATED"); // println!("THE STORE: {:?}", store); // println!("THE BYTES: {:?}", store.read_all_bytes().await?); @@ -27,7 +27,7 @@ async fn create_store() -> Result> { } async fn connect_store() -> Result> { - let mut store: Store = Store::connect(TABLE_PATH).await.map_err(|e| e.to_io_or_panic())?; + let store: Store = Store::connect(TABLE_PATH).await.map_err(|e| e.to_io_or_panic())?; println!("CONNECTED"); // println!("THE STORE: {:?}", store); // println!("THE BYTES: {:?}", store.read_all_bytes().await?); @@ -56,10 +56,9 @@ async fn append_entry(cursor: &mut WriteCursor<'_, Data>, entry: Entry) -> async fn read_entry(cursor: &mut ReadCursor<'_, Data>, file_position: FilePosition) -> Result>> { println!("READING ENTRY at file_position={}", file_position); - // let entry = cursor.read_entry_at(file_position).await.map_err(|e| e.to_io_or_panic())?; - // println!("ENTRY: {:?}", entry); - // Ok(entry) - todo!() + let entry = cursor.read_entry_at(file_position).await.map_err(|e| e.to_io_or_panic())?; + println!("ENTRY: {:?}", entry); + Ok(entry) } async fn append_bunch_of_entries(store: &mut Store) -> Result<()> { @@ -88,7 +87,7 @@ async fn test_garbage_collection(store: &mut Store) -> Result<()> { // 1. mark a bunch of entries as deleted let column = 0; let value = 1; - // cursor.delete_entries_where_eq(column, &value, true).await.map_err(|e| e.to_io_or_panic())?; + cursor.delete_entries_where_eq(column, &value, true).await.map_err(|e| e.to_io_or_panic())?; // let value = 50; // cursor.delete_entries_where_eq(column, &value).await.map_err(|e| e.to_io_or_panic())?; @@ -133,10 +132,10 @@ async fn main() -> Result<()> { test_garbage_collection(&mut store).await?; - { - let mut cursor = store.read_cursor().await.map_err(|e| e.to_io_or_panic())?; - cursor.read_entries().await.map_err(|e| e.to_io_or_panic())?; - } + // { + // let mut cursor = store.read_cursor().await.map_err(|e| e.to_io_or_panic())?; + // cursor.read_entries().await.map_err(|e| e.to_io_or_panic())?; + // } // { diff --git a/storage_engine/src/storage_engine.rs b/storage_engine/src/storage_engine.rs index 94ac5bd..74ba9ad 100644 --- a/storage_engine/src/storage_engine.rs +++ b/storage_engine/src/storage_engine.rs @@ -2,7 +2,6 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::fs::{File, OpenOptions, DirBuilder}; use tokio::fs; use std::path::{Path, PathBuf}; -use std::marker::PhantomData; use bincode::{Decode, Encode}; use crate::error::Error; @@ -25,9 +24,6 @@ pub struct Store { pub type StoreIndexes = Vec>>; -pub type PositionOfValue = FilePosition; -pub type PositionOfRow = FilePosition; - //===Store=== pub async fn store_exists(table_folder: &str) -> Result { @@ -197,6 +193,7 @@ impl Store { } // For debugging. + #[allow(dead_code)] pub async fn read_all_bytes(&mut self) -> std::result::Result, std::io::Error> where T: Send + Sync { @@ -205,99 +202,3 @@ impl Store { Ok(bytes) } } - -// ===Store Header=== - -// ====Entry==== - - -// impl StorageEngine for ColumnStore { -// async fn append(&mut self, id: Index, entry: Row) -> Result - -// async fn get_all(&self) -> ??? -// async fn get_eq(&self, column: Column, value: T) -> ??? - -// async fn delete_all(&mut self) -// async fn delete_eq(&mut self, column: Column, value: T) -> ??? -// } - -// struct Error { -// } - - -// Selected( -// &'a TableSchema, -// ColumnSelection, -// TODO: Don't do the Box(dyn Iterator<...>) -// you'll have a concrete implementation of Iterator, and that's what you'll use -// Box + 'a + Send>, -// ), - - - -// #[async_trait] -// trait StorageEngine -// where T: Encode + Decode -// { -// async fn append(&mut self, id: Index, entry: Row) -> Result - -// async fn get_all(&self) -> ??? -// async fn get_eq(&self, column: Column, value: T) -> ??? - -// async fn delete_all(&mut self) -// async fn delete_eq(&mut self, column: Column, value: T) -> ??? -// } - -// #[cfg(test)] -// mod tests { -// #[test] -// fn hello_test() { -// assert!(true); -// } -// } - -// let sroage_engine = STorageEngine::new("users") -// let mut next_position = 0 - - -// type FilePosition = usize; - - -// type StoreFile = Vec; -// type IndexFile = ??? - -// struct IndexEntry { - -// } - - -// #00000 [false, u26, "Arnold", "schwarzenegger", "gettothechoppa@yahoo.com"] #5120000 [true, u27, "Arnold", "Vosloo", "avosloo@aol.com"] -// #00000 [true, u27, "Arnold", "Vosloo", "avosloo@aol.com"] - - -// at #00000 512 kb deleted, -// ... - - - - -// [(u26, [#00000]), (u27, [#5120000])] -// [("Arnold", [#000000, #5120000]), ("Arnfsdaf", []), ("Adasdsd", []), ("Bdsad", [])] -// // basically always keep indexes in memory and on write always sync on disk - - - - -// CREATE INDEX usersname ON "users" (name); - -// INSERT INTO users (id, name, surname, email) VALUES (u26, "Arnold", "schwarzenegger", "gettothechoppa@yahoo.com"); -// INSERT INTO users (id, name, surname, email) VALUES (u27, "Arnold", "Vosloo", "avosloo@aol.com"); - - -// SELECT * FROM users WHERE id=u26; - -// SELECT * FROM users WHERE name="Arnold"; - - -// SELECT * FROM cars; -// DELETE FROM users WHERE name="Arnold"; diff --git a/storage_engine/src/store_header.rs b/storage_engine/src/store_header.rs index 004e26a..890b198 100644 --- a/storage_engine/src/store_header.rs +++ b/storage_engine/src/store_header.rs @@ -35,6 +35,7 @@ impl StoreHeader { pub const DELETED_COUNT_OFFSET: usize = Self::NUMBER_OF_COLUMNS_OFFSET + Self::NUMBER_OF_COLUMNS_SIZE; pub const TOTAL_COUNT_OFFSET: usize = Self::DELETED_COUNT_OFFSET + Self::DELETED_COUNT_SIZE; pub const PRIMARY_COLUMN_OFFSET: usize = Self::TOTAL_COUNT_OFFSET + Self::TOTAL_COUNT_SIZE; + #[allow(dead_code)] pub const INDEXED_COLUMNS_OFFSET: usize = Self::PRIMARY_COLUMN_OFFSET + Self::PRIMARY_COLUMN_SIZE; fn indexed_columns_size(number_of_columns: usize) -> usize {