Fix delete bug
This commit is contained in:
parent
3bf04ae2d6
commit
a345bf99c6
4 changed files with 107 additions and 62 deletions
|
|
@ -10,12 +10,13 @@ use bincode::{Decode, Encode};
|
||||||
use crate::binary_coding::encode;
|
use crate::binary_coding::encode;
|
||||||
|
|
||||||
use crate::entry::{Entry, EntryDetailed};
|
use crate::entry::{Entry, EntryDetailed};
|
||||||
use crate::entry_header::EntryHeaderWithDataSize;
|
use crate::entry_header::{EntryHeaderWithDataSize, EntryHeader};
|
||||||
use crate::store_header::StoreHeader;
|
use crate::store_header::StoreHeader;
|
||||||
use crate::storage_engine::{Store, FilePosition, Result, ROWS_FILE_NAME};
|
use crate::storage_engine::{Store, FilePosition, Result, ROWS_FILE_NAME};
|
||||||
|
|
||||||
#[async_trait]
|
#[async_trait]
|
||||||
trait PrimitiveCursor<T> {
|
// TODO: Make this private
|
||||||
|
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;
|
||||||
|
|
||||||
|
|
@ -30,22 +31,23 @@ trait PrimitiveCursor<T> {
|
||||||
Ok(result)
|
Ok(result)
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn seek_to(&mut self, file_position: FilePosition) -> Result<()> {
|
async fn seek_to(&mut self, file_position: FilePosition) -> Result<FilePosition> {
|
||||||
self.file().seek(SeekFrom::Start(file_position)).await?;
|
let file_position = self.file().seek(SeekFrom::Start(file_position)).await?;
|
||||||
Ok(())
|
Ok(file_position)
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn seek_to_start(&mut self) -> Result<()> {
|
// Start of the file i.e. the Header, not the entries.
|
||||||
self.file().seek(SeekFrom::Start(0)).await?;
|
async fn seek_to_start(&mut self) -> Result<FilePosition> {
|
||||||
Ok(())
|
let file_position = self.file().seek(SeekFrom::Start(0)).await?;
|
||||||
|
Ok(file_position)
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn seek_to_end(&mut self) -> Result<()> {
|
async fn seek_to_end(&mut self) -> Result<FilePosition> {
|
||||||
self.file().seek(SeekFrom::End(0)).await?;
|
let file_position = self.file().seek(SeekFrom::End(0)).await?;
|
||||||
Ok(())
|
Ok(file_position)
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn seek_to_start_of_data(&mut self) -> Result<()> {
|
async fn seek_to_start_of_data(&mut self) -> Result<FilePosition> {
|
||||||
self.seek_to(StoreHeader::SIZE as u64).await
|
self.seek_to(StoreHeader::SIZE as u64).await
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -55,7 +57,10 @@ trait PrimitiveCursor<T> {
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn is_at_eof(&mut self) -> Result<bool> {
|
async fn is_at_eof(&mut self) -> Result<bool> {
|
||||||
Ok(self.current_file_position().await? == self.eof_file_position())
|
let current_file_position = self.current_file_position().await?;
|
||||||
|
let eof_file_position = self.eof_file_position();
|
||||||
|
println!("IN EOF: current={}, eof_file_position={}", current_file_position, eof_file_position);
|
||||||
|
Ok(current_file_position == eof_file_position)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -90,9 +95,12 @@ pub trait CursorWithStoreHeader<T>: PrimitiveCursor<T> {
|
||||||
async fn next(&mut self) -> Result<Option<EntryDetailed<T>>>
|
async fn next(&mut self) -> Result<Option<EntryDetailed<T>>>
|
||||||
where T: Decode
|
where T: Decode
|
||||||
{
|
{
|
||||||
|
println!("are we at eof?");
|
||||||
if self.is_at_eof().await? {
|
if self.is_at_eof().await? {
|
||||||
|
println!("YES");
|
||||||
return Ok(None)
|
return Ok(None)
|
||||||
}
|
}
|
||||||
|
println!("NO");
|
||||||
|
|
||||||
let entry_header = self.read_entry_header().await?;
|
let entry_header = self.read_entry_header().await?;
|
||||||
|
|
||||||
|
|
@ -118,6 +126,7 @@ pub trait CursorWithStoreHeader<T>: PrimitiveCursor<T> {
|
||||||
|
|
||||||
async fn read_all_bytes(&mut self) -> std::result::Result<Vec<u8>, std::io::Error> {
|
async fn read_all_bytes(&mut self) -> std::result::Result<Vec<u8>, std::io::Error> {
|
||||||
let mut bytes: Vec<u8> = vec![];
|
let mut bytes: Vec<u8> = vec![];
|
||||||
|
self.seek_to_start().await.map_err(|e| e.to_io_or_panic())?;
|
||||||
self.file().read_to_end(&mut bytes).await?;
|
self.file().read_to_end(&mut bytes).await?;
|
||||||
Ok(bytes)
|
Ok(bytes)
|
||||||
}
|
}
|
||||||
|
|
@ -191,8 +200,11 @@ impl <T> ReadCursor<T> {
|
||||||
file,
|
file,
|
||||||
data_type: store.data_type,
|
data_type: store.data_type,
|
||||||
|
|
||||||
eof_file_position: 0, // This will be overwriten by the seek_to_start_of_data
|
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?;
|
cursor.seek_to_start_of_data().await?;
|
||||||
|
|
||||||
Ok(cursor)
|
Ok(cursor)
|
||||||
|
|
@ -221,8 +233,11 @@ impl <'cursor, T> WriteCursor<'cursor, T> {
|
||||||
file,
|
file,
|
||||||
data_type: store.data_type,
|
data_type: store.data_type,
|
||||||
|
|
||||||
eof_file_position: 0, // This will be overwriten by the seek_to_start_of_data
|
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?;
|
cursor.seek_to_start_of_data().await?;
|
||||||
|
|
||||||
Ok(cursor)
|
Ok(cursor)
|
||||||
|
|
@ -255,12 +270,10 @@ impl <'cursor, T> WriteCursor<'cursor, T> {
|
||||||
}
|
}
|
||||||
|
|
||||||
// ===Entry Header Manipulation===
|
// ===Entry Header Manipulation===
|
||||||
// assumes we are at the start of the valid entry.
|
// assumes we are at the start of valid entry.
|
||||||
async fn set_entry_is_deleted_to(&mut self, is_deleted: bool) -> Result<()>
|
async fn set_new_entry_header(&mut self, entry_header: EntryHeader) -> Result<()> {
|
||||||
where T: Send
|
let bytes: Vec<u8> = entry_header.encode()?;
|
||||||
{
|
self.write_bytes(&bytes).await?;
|
||||||
self.seek_to(EntryHeaderWithDataSize::IS_DELETED_OFFSET as u64).await?;
|
|
||||||
self.write_bytes(&encode::<bool>(&is_deleted)?).await?;
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -274,8 +287,7 @@ impl <'cursor, T> WriteCursor<'cursor, T> {
|
||||||
self.increment_total_count().await?;
|
self.increment_total_count().await?;
|
||||||
|
|
||||||
let encoded_entry: Vec<u8> = entry.encode()?;
|
let encoded_entry: Vec<u8> = entry.encode()?;
|
||||||
self.seek_to_end().await?;
|
let file_position = self.seek_to_end().await?;
|
||||||
let file_position: FilePosition = self.current_file_position().await?;
|
|
||||||
self.write_bytes(&encoded_entry).await?;
|
self.write_bytes(&encoded_entry).await?;
|
||||||
|
|
||||||
let eof_file_position: FilePosition = self.current_file_position().await?;
|
let eof_file_position: FilePosition = self.current_file_position().await?;
|
||||||
|
|
@ -289,13 +301,19 @@ impl <'cursor, T> WriteCursor<'cursor, T> {
|
||||||
where T: Send
|
where T: Send
|
||||||
{
|
{
|
||||||
self.seek_to(file_position).await?;
|
self.seek_to(file_position).await?;
|
||||||
let entry_header = self.read_entry_header().await?;
|
println!("Attempting to read the entry");
|
||||||
|
let mut entry_header = self.read_entry_header().await?;
|
||||||
|
println!("Entry Header == {:?}", entry_header);
|
||||||
if entry_header.is_deleted {
|
if entry_header.is_deleted {
|
||||||
|
println!("Already deleted");
|
||||||
Ok(())
|
Ok(())
|
||||||
} else {
|
} else {
|
||||||
|
println!("Marking as deleted");
|
||||||
self.increment_deleted_count().await?;
|
self.increment_deleted_count().await?;
|
||||||
self.seek_to(file_position).await?;
|
self.seek_to(file_position).await?;
|
||||||
self.set_entry_is_deleted_to(true).await?;
|
|
||||||
|
entry_header.is_deleted = true;
|
||||||
|
self.set_new_entry_header(entry_header.into()).await?;
|
||||||
|
|
||||||
self.attempt_garbage_collection_if_necessary().await?;
|
self.attempt_garbage_collection_if_necessary().await?;
|
||||||
Ok(())
|
Ok(())
|
||||||
|
|
|
||||||
|
|
@ -1,6 +1,6 @@
|
||||||
use bincode::{Decode, Encode};
|
use bincode::{Decode, Encode};
|
||||||
|
|
||||||
use crate::binary_coding::{encode, encode_sequence, encode_sequence_with_sizes, decode_sequence};
|
use crate::binary_coding::{encode_sequence, encode_sequence_with_sizes, decode_sequence};
|
||||||
use crate::storage_engine::Result;
|
use crate::storage_engine::Result;
|
||||||
use crate::error::{Error, DecodeErrorKind};
|
use crate::error::{Error, DecodeErrorKind};
|
||||||
use crate::entry_header::{EntryHeader, EntryHeaderWithDataSize};
|
use crate::entry_header::{EntryHeader, EntryHeaderWithDataSize};
|
||||||
|
|
@ -17,13 +17,6 @@ pub struct EntryDetailed<T> {
|
||||||
data: Vec<T>,
|
data: Vec<T>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl EntryHeader {
|
|
||||||
fn encode(self: &EntryHeader) -> Result<Vec<u8>> {
|
|
||||||
let result: Vec<u8> = encode(&self.is_deleted)?;
|
|
||||||
Ok(result)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl <T>Entry<T> {
|
impl <T>Entry<T> {
|
||||||
pub fn new(data: Vec<T>) -> Self {
|
pub fn new(data: Vec<T>) -> Self {
|
||||||
Self { header: EntryHeader { is_deleted: false }, data }
|
Self { header: EntryHeader { is_deleted: false }, data }
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,4 @@
|
||||||
use crate::binary_coding::{decode, decode_sequence};
|
use crate::binary_coding::{decode, encode, decode_sequence};
|
||||||
use crate::storage_engine::Result;
|
use crate::storage_engine::Result;
|
||||||
use crate::error::{Error, DecodeErrorKind};
|
use crate::error::{Error, DecodeErrorKind};
|
||||||
use std::mem::size_of;
|
use std::mem::size_of;
|
||||||
|
|
@ -14,6 +14,20 @@ pub struct EntryHeaderWithDataSize {
|
||||||
pub data_sizes: Vec<usize>, // vec![5, 6, 20] means that column 0 stores 5 bytes, column 1 stores 6
|
pub data_sizes: Vec<usize>, // vec![5, 6, 20] means that column 0 stores 5 bytes, column 1 stores 6
|
||||||
// bytes etc
|
// bytes etc
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl EntryHeader {
|
||||||
|
pub fn encode(self: &EntryHeader) -> Result<Vec<u8>> {
|
||||||
|
let result: Vec<u8> = encode(&self.is_deleted)?;
|
||||||
|
Ok(result)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl From<EntryHeaderWithDataSize> for EntryHeader {
|
||||||
|
fn from(entry: EntryHeaderWithDataSize) -> Self {
|
||||||
|
Self { is_deleted: entry.is_deleted, }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
impl EntryHeaderWithDataSize {
|
impl EntryHeaderWithDataSize {
|
||||||
pub const IS_DELETED_OFFSET: usize = 0;
|
pub const IS_DELETED_OFFSET: usize = 0;
|
||||||
pub const IS_DELETED_SIZE: usize = size_of::<bool>();
|
pub const IS_DELETED_SIZE: usize = size_of::<bool>();
|
||||||
|
|
|
||||||
|
|
@ -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};
|
use crate::cursor::{ReadCursor, WriteCursor, CursorWithStoreHeader, PrimitiveCursor};
|
||||||
|
|
||||||
type Data = u32;
|
type Data = u32;
|
||||||
|
|
||||||
|
|
@ -20,8 +20,8 @@ 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 mut 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?);
|
||||||
|
|
||||||
Ok(store)
|
Ok(store)
|
||||||
}
|
}
|
||||||
|
|
@ -29,8 +29,8 @@ 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 mut 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?);
|
||||||
Ok(store)
|
Ok(store)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -60,14 +60,7 @@ async fn read_entry(cursor: &mut ReadCursor<Data>, file_position: FilePosition)
|
||||||
todo!()
|
todo!()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn append_bunch_of_entries(store: &mut Store<Data>) -> Result<()> {
|
||||||
#[tokio::main]
|
|
||||||
async fn main() -> Result<()> {
|
|
||||||
println!("STOOOOOOOOOOOORAAAAAAAAAAAGE");
|
|
||||||
|
|
||||||
let mut store: Store<Data> = create_or_connect().await?;
|
|
||||||
|
|
||||||
{
|
|
||||||
let mut cursor = store.write_cursor().await.map_err(|e| e.to_io_or_panic())?;
|
let mut cursor = store.write_cursor().await.map_err(|e| e.to_io_or_panic())?;
|
||||||
let entry0: Entry<u32> = Entry::new(vec![1, 2, 3, 4, 5]);
|
let entry0: Entry<u32> = Entry::new(vec![1, 2, 3, 4, 5]);
|
||||||
append_entry(&mut cursor, &entry0).await?;
|
append_entry(&mut cursor, &entry0).await?;
|
||||||
|
|
@ -81,28 +74,55 @@ async fn main() -> Result<()> {
|
||||||
|
|
||||||
let entry3: Entry<u32> = Entry::new(vec![50,50,50,50,50]);
|
let entry3: Entry<u32> = Entry::new(vec![50,50,50,50,50]);
|
||||||
append_entry(&mut cursor, &entry3).await?;
|
append_entry(&mut cursor, &entry3).await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::main]
|
||||||
|
async fn main() -> Result<()> {
|
||||||
|
println!("STOOOOOOOOOOOORAAAAAAAAAAAGE");
|
||||||
|
|
||||||
|
let mut store: Store<Data> = create_or_connect().await?;
|
||||||
|
|
||||||
|
if store.header.total_count == 0 {
|
||||||
|
println!("INSERTING!");
|
||||||
|
append_bunch_of_entries(&mut store).await?;
|
||||||
|
}
|
||||||
|
|
||||||
|
{
|
||||||
|
let mut cursor = store.write_cursor().await.map_err(|e| e.to_io_or_panic())?;
|
||||||
|
|
||||||
|
let entry: Entry<u32> = Entry::new(vec![60, 50, 40, 30, 20]);
|
||||||
|
// let file_position = append_entry(&mut cursor, &entry).await?;
|
||||||
|
// let file_position = 215;
|
||||||
|
// cursor.seek_to(file_position).await.map_err(|e| e.to_io_or_panic())?;
|
||||||
|
|
||||||
|
// let entry_header = cursor.read_entry_header().await.map_err(|e| e.to_io_or_panic())?;
|
||||||
|
// println!("entry header = {:?}", entry_header);
|
||||||
|
|
||||||
|
// println!("FILE POSITION == {}", file_position);
|
||||||
|
// cursor.mark_deleted_at(file_position).await.map_err(|e| e.to_io_or_panic())?;
|
||||||
|
// let entry_header = cursor.read_entry_header().await.map_err(|e| e.to_io_or_panic())?;
|
||||||
|
// println!("entry header after delete = {:?}", entry_header);
|
||||||
}
|
}
|
||||||
|
|
||||||
// println!("{:?}", store);
|
// println!("{:?}", store);
|
||||||
// println!("{:?}", store.read_all_bytes().await?);
|
// println!("{:?}", store.read_all_bytes().await?);
|
||||||
|
|
||||||
// let entry0: Entry<u32> = Entry::new(vec![99, 98, 97, 96, 95]);
|
let mut cursor = store.read_cursor().await.map_err(|e| e.to_io_or_panic())?;
|
||||||
// append_entry(&mut cursor, &entry0).await?;
|
|
||||||
|
|
||||||
// let entry1: Entry<u32> = Entry::new(vec![50,50,50,50,50]);
|
// let x = cursor.next().await.map_err(|e| e.to_io_or_panic())?;
|
||||||
// let file_position = append_entry(&mut cursor, &entry1).await?;
|
// println!("{:?}", x);
|
||||||
// println!("CURRENT FILE_POSITION = {}", file_position);
|
// let x = cursor.next().await.map_err(|e| e.to_io_or_panic())?;
|
||||||
|
// println!("{:?}", x);
|
||||||
|
// let x = cursor.next().await.map_err(|e| e.to_io_or_panic())?;
|
||||||
|
// println!("{:?}", x);
|
||||||
|
// let x = cursor.next().await.map_err(|e| e.to_io_or_panic())?;
|
||||||
|
// println!("{:?}", x);
|
||||||
|
// let x = cursor.next().await.map_err(|e| e.to_io_or_panic())?;
|
||||||
|
// println!("{:?}", x);
|
||||||
|
|
||||||
// Now file_position point to entry1.
|
cursor.read_entries().await.map_err(|e| e.to_io_or_panic())?;
|
||||||
// cursor.mark_deleted_at(file_position).await.map_err(|e| e.to_io_or_panic())?;
|
|
||||||
// cursor.seek_to(file_position).await.map_err(|e| e.to_io_or_panic())?;
|
|
||||||
|
|
||||||
// let entry2: StoreEntry<u32> = StoreEntry::new_deleted(vec![3, 2, 1]);
|
|
||||||
// let cursor2 = store.append_entry(&entry2).await.map_err(|e| e.to_io_or_panic())?;
|
|
||||||
// println!("cursor2 = {}", cursor2);
|
|
||||||
|
|
||||||
println!("{:?}", store);
|
|
||||||
println!("{:?}", store.read_all_bytes().await?);
|
|
||||||
|
|
||||||
|
|
||||||
println!("DONE");
|
println!("DONE");
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue