Cleanup
This commit is contained in:
parent
c0a3ee08b8
commit
1618bffb85
6 changed files with 22 additions and 152 deletions
|
|
@ -69,11 +69,12 @@ pub fn decode_sequence<T: Decode>(len: usize, bytes: &[u8]) -> Result<Vec<T>, bi
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
#[allow(dead_code)]
|
||||||
fn example_encoding_decoding() {
|
fn example_encoding_decoding() {
|
||||||
let xs: Vec<u32> = vec![123, 250, 256, 123, 123, 123];
|
let _xs: Vec<u32> = vec![123, 250, 256, 123, 123, 123];
|
||||||
let xs: Vec<u32> = vec![];
|
let _xs: Vec<u32> = vec![];
|
||||||
let xs: Vec<u32> = vec![123];
|
let _xs: Vec<u32> = vec![123];
|
||||||
let xs: Vec<u32> = vec![123, 250];
|
let _xs: Vec<u32> = vec![123, 250];
|
||||||
|
|
||||||
let xs: Vec<String> = vec!["foo".to_string(), "bar".to_string()];
|
let xs: Vec<String> = vec!["foo".to_string(), "bar".to_string()];
|
||||||
|
|
||||||
|
|
@ -87,4 +88,3 @@ fn example_encoding_decoding() {
|
||||||
let dxs = decode_vector::<String>(&exs[..]).unwrap();
|
let dxs = decode_vector::<String>(&exs[..]).unwrap();
|
||||||
println!("decoded {:?}", dxs);
|
println!("decoded {:?}", dxs);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -49,8 +49,7 @@ pub struct AppendOnlyCursor<T> {
|
||||||
|
|
||||||
// ===Traits===
|
// ===Traits===
|
||||||
#[async_trait]
|
#[async_trait]
|
||||||
// TODO: Make this private
|
pub(crate) trait PrimitiveCursor<T> {
|
||||||
pub trait PrimitiveCursor<T> {
|
|
||||||
fn file(&mut self) -> &mut File;
|
fn file(&mut self) -> &mut File;
|
||||||
fn eof_file_position(&self) -> FilePosition;
|
fn eof_file_position(&self) -> FilePosition;
|
||||||
|
|
||||||
|
|
@ -100,7 +99,7 @@ pub trait PrimitiveCursor<T> {
|
||||||
}
|
}
|
||||||
|
|
||||||
#[async_trait]
|
#[async_trait]
|
||||||
pub trait PrimitiveWriteCursor<T>: PrimitiveCursor<T> {
|
pub(crate) trait PrimitiveWriteCursor<T>: PrimitiveCursor<T> {
|
||||||
async fn write_bytes(&mut self, bytes: &[u8]) -> Result<usize> {
|
async fn write_bytes(&mut self, bytes: &[u8]) -> Result<usize> {
|
||||||
Ok(self.file().write(bytes).await?)
|
Ok(self.file().write(bytes).await?)
|
||||||
}
|
}
|
||||||
|
|
@ -536,8 +535,6 @@ impl <'cursor, T> ReadCursor<'cursor, T> {
|
||||||
|
|
||||||
|
|
||||||
impl <'cursor, T> WriteCursor<'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
|
// 'store lives atleast as long as 'cursor
|
||||||
pub async fn new<'store: 'cursor>(store: &'store mut Store<T>) -> Result<Self>
|
pub async fn new<'store: 'cursor>(store: &'store mut Store<T>) -> Result<Self>
|
||||||
|
|
@ -566,31 +563,6 @@ impl <'cursor, T> WriteCursor<'cursor, T>
|
||||||
Ok(cursor)
|
Ok(cursor)
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn connect<'header: 'cursor, 'indexes: 'cursor>(path_to_rows: &str, header: &'header mut StoreHeader, indexes: &'indexes mut StoreIndexes<T>) -> Result<Self>
|
|
||||||
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===
|
// ===Entry Header Manipulation===
|
||||||
// assumes we are at the start of valid entry.
|
// assumes we are at the start of valid entry.
|
||||||
async fn set_new_entry_header(&mut self, entry_header: EntryHeader) -> Result<()>
|
async fn set_new_entry_header(&mut self, entry_header: EntryHeader) -> Result<()>
|
||||||
|
|
@ -707,7 +679,7 @@ impl <'cursor, T> WriteCursor<'cursor, T>
|
||||||
|
|
||||||
// ===Garbage Collection===
|
// ===Garbage Collection===
|
||||||
async fn attempt_garbage_collection_if_necessary(&mut self) -> Result<()>
|
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 {
|
if self.header.deleted_count > GARBAGE_COLLECTION_TRIGGER {
|
||||||
println!("=======START GARBAGE COLLETOR====");
|
println!("=======START GARBAGE COLLETOR====");
|
||||||
|
|
@ -717,7 +689,7 @@ impl <'cursor, T> WriteCursor<'cursor, T>
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn initiate_garbage_collection(&mut self) -> Result<usize>
|
pub async fn initiate_garbage_collection(&mut self) -> Result<usize>
|
||||||
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?;
|
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
|
// Since garbage collection changes FilePositions of live entries, we need to update the
|
||||||
|
|
|
||||||
|
|
@ -8,7 +8,6 @@ pub enum Error {
|
||||||
IndexIsStoringEofFilePosition(Column),
|
IndexIsStoringEofFilePosition(Column),
|
||||||
ColumnAlreadyIndexed(Column),
|
ColumnAlreadyIndexed(Column),
|
||||||
IoError(std::io::Error),
|
IoError(std::io::Error),
|
||||||
InvalidStoreHeader,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
|
|
@ -49,5 +48,3 @@ impl From<std::io::Error> for Error {
|
||||||
Self::IoError(err)
|
Self::IoError(err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -9,7 +9,7 @@ mod store_header;
|
||||||
|
|
||||||
use crate::entry::{Entry, EntryDetailed};
|
use crate::entry::{Entry, EntryDetailed};
|
||||||
use crate::storage_engine::{Store, FilePosition};
|
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;
|
type Data = u32;
|
||||||
|
|
||||||
|
|
@ -18,7 +18,7 @@ const TABLE_PATH: &'static str = "test_table";
|
||||||
type Result<T> = std::result::Result<T, std::io::Error>;
|
type Result<T> = std::result::Result<T, std::io::Error>;
|
||||||
|
|
||||||
async fn create_store() -> Result<Store<Data>> {
|
async fn create_store() -> Result<Store<Data>> {
|
||||||
let mut store: Store<Data> = Store::new(TABLE_PATH, 5, 0).await.map_err(|e| e.to_io_or_panic())?;
|
let store: Store<Data> = Store::new(TABLE_PATH, 5, 0).await.map_err(|e| e.to_io_or_panic())?;
|
||||||
println!("CREATED");
|
println!("CREATED");
|
||||||
// println!("THE STORE: {:?}", store);
|
// println!("THE STORE: {:?}", store);
|
||||||
// println!("THE BYTES: {:?}", store.read_all_bytes().await?);
|
// println!("THE BYTES: {:?}", store.read_all_bytes().await?);
|
||||||
|
|
@ -27,7 +27,7 @@ async fn create_store() -> Result<Store<Data>> {
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn connect_store() -> Result<Store<Data>> {
|
async fn connect_store() -> Result<Store<Data>> {
|
||||||
let mut store: Store<Data> = Store::connect(TABLE_PATH).await.map_err(|e| e.to_io_or_panic())?;
|
let store: Store<Data> = Store::connect(TABLE_PATH).await.map_err(|e| e.to_io_or_panic())?;
|
||||||
println!("CONNECTED");
|
println!("CONNECTED");
|
||||||
// println!("THE STORE: {:?}", store);
|
// println!("THE STORE: {:?}", store);
|
||||||
// println!("THE BYTES: {:?}", store.read_all_bytes().await?);
|
// println!("THE BYTES: {:?}", store.read_all_bytes().await?);
|
||||||
|
|
@ -56,10 +56,9 @@ async fn append_entry(cursor: &mut WriteCursor<'_, Data>, entry: Entry<Data>) ->
|
||||||
|
|
||||||
async fn read_entry(cursor: &mut ReadCursor<'_, Data>, file_position: FilePosition) -> Result<Option<EntryDetailed<Data>>> {
|
async fn read_entry(cursor: &mut ReadCursor<'_, Data>, file_position: FilePosition) -> Result<Option<EntryDetailed<Data>>> {
|
||||||
println!("READING ENTRY at file_position={}", file_position);
|
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())?;
|
let entry = cursor.read_entry_at(file_position).await.map_err(|e| e.to_io_or_panic())?;
|
||||||
// println!("ENTRY: {:?}", entry);
|
println!("ENTRY: {:?}", entry);
|
||||||
// Ok(entry)
|
Ok(entry)
|
||||||
todo!()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn append_bunch_of_entries(store: &mut Store<Data>) -> Result<()> {
|
async fn append_bunch_of_entries(store: &mut Store<Data>) -> Result<()> {
|
||||||
|
|
@ -88,7 +87,7 @@ async fn test_garbage_collection(store: &mut Store<Data>) -> Result<()> {
|
||||||
// 1. mark a bunch of entries as deleted
|
// 1. mark a bunch of entries as deleted
|
||||||
let column = 0;
|
let column = 0;
|
||||||
let value = 1;
|
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;
|
// let value = 50;
|
||||||
// cursor.delete_entries_where_eq(column, &value).await.map_err(|e| e.to_io_or_panic())?;
|
// 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?;
|
test_garbage_collection(&mut store).await?;
|
||||||
|
|
||||||
{
|
// {
|
||||||
let mut cursor = store.read_cursor().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())?;
|
// cursor.read_entries().await.map_err(|e| e.to_io_or_panic())?;
|
||||||
}
|
// }
|
||||||
|
|
||||||
|
|
||||||
// {
|
// {
|
||||||
|
|
|
||||||
|
|
@ -2,7 +2,6 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||||
use tokio::fs::{File, OpenOptions, DirBuilder};
|
use tokio::fs::{File, OpenOptions, DirBuilder};
|
||||||
use tokio::fs;
|
use tokio::fs;
|
||||||
use std::path::{Path, PathBuf};
|
use std::path::{Path, PathBuf};
|
||||||
use std::marker::PhantomData;
|
|
||||||
use bincode::{Decode, Encode};
|
use bincode::{Decode, Encode};
|
||||||
|
|
||||||
use crate::error::Error;
|
use crate::error::Error;
|
||||||
|
|
@ -25,9 +24,6 @@ pub struct Store<T> {
|
||||||
|
|
||||||
pub type StoreIndexes<T> = Vec<Option<Index<T, FilePosition>>>;
|
pub type StoreIndexes<T> = Vec<Option<Index<T, FilePosition>>>;
|
||||||
|
|
||||||
pub type PositionOfValue = FilePosition;
|
|
||||||
pub type PositionOfRow = FilePosition;
|
|
||||||
|
|
||||||
|
|
||||||
//===Store===
|
//===Store===
|
||||||
pub async fn store_exists(table_folder: &str) -> Result<bool> {
|
pub async fn store_exists(table_folder: &str) -> Result<bool> {
|
||||||
|
|
@ -197,6 +193,7 @@ impl <T>Store<T> {
|
||||||
}
|
}
|
||||||
|
|
||||||
// For debugging.
|
// For debugging.
|
||||||
|
#[allow(dead_code)]
|
||||||
pub async fn read_all_bytes(&mut self) -> std::result::Result<Vec<u8>, std::io::Error>
|
pub async fn read_all_bytes(&mut self) -> std::result::Result<Vec<u8>, std::io::Error>
|
||||||
where T: Send + Sync
|
where T: Send + Sync
|
||||||
{
|
{
|
||||||
|
|
@ -205,99 +202,3 @@ impl <T>Store<T> {
|
||||||
Ok(bytes)
|
Ok(bytes)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// ===Store Header===
|
|
||||||
|
|
||||||
// ====Entry====
|
|
||||||
|
|
||||||
|
|
||||||
// impl StorageEngine for ColumnStore {
|
|
||||||
// async fn append(&mut self, id: Index, entry: Row<T>) -> Result<???, Error>
|
|
||||||
|
|
||||||
// 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<dyn Iterator<Item = RestrictedRow> + 'a + Send>,
|
|
||||||
// ),
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
// #[async_trait]
|
|
||||||
// trait StorageEngine<T>
|
|
||||||
// where T: Encode + Decode
|
|
||||||
// {
|
|
||||||
// async fn append(&mut self, id: Index, entry: Row<T>) -> Result<???, Error>
|
|
||||||
|
|
||||||
// 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<Entry>;
|
|
||||||
// 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";
|
|
||||||
|
|
|
||||||
|
|
@ -35,6 +35,7 @@ impl StoreHeader {
|
||||||
pub const DELETED_COUNT_OFFSET: usize = Self::NUMBER_OF_COLUMNS_OFFSET + Self::NUMBER_OF_COLUMNS_SIZE;
|
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 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;
|
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;
|
pub const INDEXED_COLUMNS_OFFSET: usize = Self::PRIMARY_COLUMN_OFFSET + Self::PRIMARY_COLUMN_SIZE;
|
||||||
|
|
||||||
fn indexed_columns_size(number_of_columns: usize) -> usize {
|
fn indexed_columns_size(number_of_columns: usize) -> usize {
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue