finish impl memtable

src/blocks/block_builder.rs

@@ -23,6 +23,7 @@ pub const BLOCK_SIZE: usize = 4 * 1024;
 /// +----------------------------------------------------------------------------------------------------------------------+
 /// ```
 /// Every record will be encoded in this format and save the raw bytes into the data field.
+/// If a records with a `zero length` value pay-load, it means a deleting operation by client
 
 pub struct BlockBuilder {
     data: Vec<u8>,
@@ -50,8 +51,10 @@ impl BlockBuilder {
         // the block limits, current block will be extended
         if Self::evaluate_record_encoded_length(key, value) + self.amount > BLOCK_SIZE {
             use anyhow::Error;
-            return Err(Error::from(Box::new(io::Error::new(io::ErrorKind::Other, "your message here"))));
-
+            return Err(Error::from(Box::new(io::Error::new(
+                io::ErrorKind::Other,
+                "your message here",
+            ))));
         }
 
         // write the offset of current record
@@ -94,7 +97,7 @@ impl BlockBuilder {
         self.offsets.clear();
         self.amount = 2;
     }
-    pub fn is_empty(&self) ->bool{
+    pub fn is_empty(&self) -> bool {
         self.amount == 2
     }
 }

src/memtable.rs

@@ -31,7 +31,7 @@ impl MemTable {
         log_record_builder.add(OperationType::PUT, key, value)?;
         let ret = self
             .logger
-            .log_and_sync(log_record_builder.build().as_slice());
+            .log_and_sync(log_record_builder.build());
 
         assert!(ret.is_ok());
         self.table
@@ -44,8 +44,10 @@ impl MemTable {
             Some(value) => Some(value.clone()),
         }
     }
+
+    /// delete is composed by putting a new record with zero length value portion
     pub fn delete(&mut self, key: &[u8]) -> Result<()> {
-        self.put(key,"".as_bytes())
+        self.put(key, "".as_bytes())
     }
     pub fn recover(file: &FileObject) -> Self {
         todo!()

src/memtable/logger.rs

@@ -1,6 +1,7 @@
 use crate::util::env::{logfile_path, FileObject};
 use anyhow::Result;
-use bytes::Bytes;
+use bytes::{BufMut, Bytes, BytesMut};
+use std::thread::sleep;
 
 pub struct MemTableLogger {
     seq: u64,
@@ -11,12 +12,13 @@ impl MemTableLogger {
         let file_obj = FileObject::create(logfile_path(seq as usize).as_str()).unwrap();
         MemTableLogger { seq, file_obj }
     }
-    pub fn log_and_sync(&self, log_records: &[LoggerRecord]) -> Result<()> {
-        todo!()
+    pub fn log_and_sync(& mut self, log_records: &[u8]) -> Result<()> {
+        self.file_obj.write(log_records)?;
+        self.file_obj.sync()
     }
 }
 
-struct LoggerRecord {
+pub struct LoggerRecord {
     opt_type: OperationType,
     key: Bytes,
     value: Bytes,
@@ -35,15 +37,45 @@ pub enum OperationType {
     DELETE,
 }
 
-pub struct LogRecordsBuilder {}
+/// LogRecords is a sequence of `Entity`
+/// `Entity` format is described below
+///
+/// ``` text
+/// +----------------------------------------------------------------------------------------------------------------------+
+/// | Key Length (2 bytes) |  Key PayLoads(key-length bytes) | Value Length (2 bytes) | Value PayLoad (value-length bytes) |
+/// +----------------------------------------------------------------------------------------------------------------------+
+/// ```
+/// It is same as the block entry, if the `Value Length` portion is zero, yields a delete operation
+/// on that key
+pub struct LogRecordsBuilder {
+    data: Vec<u8>,
+}
 impl LogRecordsBuilder {
     pub fn new() -> Self {
-        todo!()
+        LogRecordsBuilder { data: Vec::new() }
     }
     pub fn add(&mut self, opt: OperationType, key: &[u8], value: &[u8]) -> Result<()> {
-        todo!()
+        let mut buf = BytesMut::new();
+        //encoding key portion of the records
+        buf.put_u16(key.len() as u16);
+        buf.put_slice(key);
+        // encoding value portion of the records
+        match opt {
+            OperationType::PUT => {
+                buf.put_u16(value.len() as u16);
+                buf.put_slice(value);
+            }
+            OperationType::DELETE => {
+                buf.put_u16(0);
+            }
+        }
+        self.data.put_slice(buf.as_ref());
+        Ok(())
     }
-    pub fn build(&self) -> Vec<LoggerRecord> {
-        todo!()
+    pub fn build(&self) -> &[u8] {
+        self.data.as_slice()
+    }
+    pub fn cleanup(&mut self) {
+        self.data.clear();
     }
 }

src/memtable/tests.rs

@@ -42,6 +42,6 @@ fn test_memtable_remove() {
     assert_eq!(memtable.get("key".as_bytes()),None);
     let ret = memtable.delete("key1".as_bytes());
     assert!(ret.is_ok());
-    assert_eq!(memtable.get("key1".as_bytes()),None);
+    assert_eq!(memtable.get("key1".as_bytes()),Some("".into()));
 
 }