add blob header

Author: marvin-j97

src/segment/merge.rs

@@ -1,5 +1,5 @@
 use crate::{id::SegmentId, SegmentReader};
-use std::cmp::Reverse;
+use std::{cmp::Reverse, sync::Arc};
 
 // TODO: replace with MinHeap...
 use min_max_heap::MinMaxHeap;
@@ -9,8 +9,8 @@ type IteratorIndex = usize;
 #[derive(Debug)]
 struct IteratorValue {
     index: IteratorIndex,
-    key: Vec<u8>,
-    value: Vec<u8>,
+    key: Arc<[u8]>,
+    value: Arc<[u8]>,
     segment_id: SegmentId,
 }
 
@@ -77,7 +77,7 @@ impl MergeReader {
 }
 
 impl Iterator for MergeReader {
-    type Item = crate::Result<(Vec<u8>, Vec<u8>, SegmentId)>;
+    type Item = crate::Result<(Arc<[u8]>, Arc<[u8]>, SegmentId)>;
 
     fn next(&mut self) -> Option<Self::Item> {
         if self.heap.is_empty() {

src/segment/mod.rs

@@ -34,8 +34,8 @@ impl Segment {
     /// # Errors
     ///
     /// Will return `Err` if an IO error occurs.
-    pub fn scan(&self) -> std::io::Result<reader::Reader> {
-        reader::Reader::new(&self.path, self.id, self.meta.item_count)
+    pub fn scan(&self) -> crate::Result<reader::Reader> {
+        reader::Reader::new(&self.path, self.id)
     }
 
     /// Always returns `false` because a segment is never empty.

src/segment/multi_writer.rs

@@ -1,4 +1,4 @@
-use super::writer::Writer;
+use super::writer::{Writer, BLOB_HEADER_MAGIC};
 use crate::{
     compression::Compressor,
     id::{IdGenerator, SegmentId},
@@ -87,7 +87,10 @@ impl MultiWriter {
         self.get_active_writer().offset()
         // NOTE: Point to the value record, not the key
         // The key is not really needed when dereferencing a value handle
-        + std::mem::size_of::<u16>() as u64 + key.len() as u64
+            + (BLOB_HEADER_MAGIC.len()
+            + std::mem::size_of::<u16>()
+            + key.len()
+        ) as u64
     }
 
     #[must_use]

src/segment/reader.rs

@@ -1,16 +1,18 @@
+use super::{meta::METADATA_HEADER_MAGIC, writer::BLOB_HEADER_MAGIC};
 use crate::id::SegmentId;
 use byteorder::{BigEndian, ReadBytesExt};
 use std::{
     fs::File,
     io::{BufReader, Read},
     path::PathBuf,
+    sync::Arc,
 };
 
 /// Reads through a segment in order.
 pub struct Reader {
     pub(crate) segment_id: SegmentId,
     inner: BufReader<File>,
-    item_count: u64,
+    is_terminated: bool,
 }
 
 impl Reader {
@@ -19,43 +21,58 @@ impl Reader {
     /// # Errors
     ///
     /// Will return `Err` if an IO error occurs.
-    pub fn new<P: Into<PathBuf>>(
-        path: P,
-        segment_id: SegmentId,
-        item_count: u64,
-    ) -> std::io::Result<Self> {
+    pub fn new<P: Into<PathBuf>>(path: P, segment_id: SegmentId) -> crate::Result<Self> {
         let path = path.into();
         let file_reader = BufReader::new(File::open(path)?);
 
         Ok(Self {
             segment_id,
             inner: file_reader,
-            item_count,
+            is_terminated: false,
         })
     }
 }
 
 impl Iterator for Reader {
-    type Item = std::io::Result<(Vec<u8>, Vec<u8>)>;
+    type Item = crate::Result<(Arc<[u8]>, Arc<[u8]>)>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        if self.item_count == 0 {
+        if self.is_terminated {
             return None;
         }
 
+        {
+            let mut buf = [0; BLOB_HEADER_MAGIC.len()];
+
+            if let Err(e) = self.inner.read_exact(&mut buf) {
+                return Some(Err(e.into()));
+            };
+
+            if buf == METADATA_HEADER_MAGIC {
+                self.is_terminated = true;
+                return None;
+            }
+
+            if buf != BLOB_HEADER_MAGIC {
+                return Some(Err(crate::Error::Deserialize(
+                    crate::serde::DeserializeError::InvalidHeader("Blob"),
+                )));
+            }
+        }
+
         let key_len = match self.inner.read_u16::<BigEndian>() {
             Ok(v) => v,
             Err(e) => {
                 if e.kind() == std::io::ErrorKind::UnexpectedEof {
                     return None;
                 }
-                return Some(Err(e));
+                return Some(Err(e.into()));
             }
         };
 
         let mut key = vec![0; key_len.into()];
         if let Err(e) = self.inner.read_exact(&mut key) {
-            return Some(Err(e));
+            return Some(Err(e.into()));
         };
 
         // TODO: handle crc
@@ -65,7 +82,7 @@ impl Iterator for Reader {
                 if e.kind() == std::io::ErrorKind::UnexpectedEof {
                     return None;
                 }
-                return Some(Err(e));
+                return Some(Err(e.into()));
             }
         };
 
@@ -75,17 +92,15 @@ impl Iterator for Reader {
                 if e.kind() == std::io::ErrorKind::UnexpectedEof {
                     return None;
                 }
-                return Some(Err(e));
+                return Some(Err(e.into()));
             }
         };
 
         let mut val = vec![0; val_len as usize];
         if let Err(e) = self.inner.read_exact(&mut val) {
-            return Some(Err(e));
+            return Some(Err(e.into()));
         };
 
-        self.item_count -= 1;
-
-        Some(Ok((key, val)))
+        Some(Ok((key.into(), val.into())))
     }
 }

src/segment/trailer.rs

@@ -26,8 +26,6 @@ impl SegmentFileTrailer {
         // Get metadata ptr
         let metadata_ptr = reader.read_u64::<BigEndian>()?;
 
-        eprintln!("load metadata @ {metadata_ptr}");
-
         // IMPORTANT: Subtract sizeof(meta_ptr) ------v
         let remaining_padding = TRAILER_SIZE - std::mem::size_of::<u64>() - TRAILER_MAGIC.len();
         reader.seek_relative(remaining_padding as i64)?;

src/segment/writer.rs

@@ -11,6 +11,8 @@ use std::{
     sync::Arc,
 };
 
+pub const BLOB_HEADER_MAGIC: &[u8] = &[b'V', b'L', b'G', b'B', b'L', b'O', b'B', b'1'];
+
 /// Segment writer
 pub struct Writer {
     pub path: PathBuf,
@@ -104,11 +106,13 @@ impl Writer {
             None => value.to_vec(),
         };
 
-        // Write CRC
         let mut hasher = crc32fast::Hasher::new();
         hasher.update(&value);
         let crc = hasher.finalize();
 
+        // Write header
+        self.active_writer.write_all(BLOB_HEADER_MAGIC)?;
+
         // Write key
 
         // NOTE: Truncation is okay and actually needed
@@ -117,6 +121,7 @@ impl Writer {
             .write_u16::<BigEndian>(key.len() as u16)?;
         self.active_writer.write_all(key)?;
 
+        // Write CRC
         self.active_writer.write_u32::<BigEndian>(crc)?;
 
         // Write value
@@ -127,6 +132,9 @@ impl Writer {
             .write_u32::<BigEndian>(value.len() as u32)?;
         self.active_writer.write_all(&value)?;
 
+        // Header
+        self.offset += BLOB_HEADER_MAGIC.len() as u64;
+
         // CRC
         self.offset += std::mem::size_of::<u32>() as u64;
 

src/value_log.rs

@@ -85,6 +85,13 @@ impl ValueLog {
         }
     }
 
+    #[doc(hidden)]
+    pub fn verify(&self) -> crate::Result<usize> {
+        let _lock = self.rollover_guard.lock().expect("lock is poisoned");
+
+        Ok(0)
+    }
+
     /// Creates a new empty value log in a directory.
     pub(crate) fn create_new<P: Into<PathBuf>>(path: P, config: Config) -> crate::Result<Self> {
         let path = absolute_path(path.into());
@@ -427,13 +434,13 @@ impl ValueLog {
     }
 
     #[doc(hidden)]
-    pub fn get_reader(&self) -> std::io::Result<MergeReader> {
+    pub fn get_reader(&self) -> crate::Result<MergeReader> {
         let segments = self.manifest.segments.read().expect("lock is poisoned");
 
         let readers = segments
             .values()
             .map(|x| x.scan())
-            .collect::<std::io::Result<Vec<_>>>()?;
+            .collect::<crate::Result<Vec<_>>>()?;
 
         Ok(MergeReader::new(readers))
     }
@@ -481,7 +488,7 @@ impl ValueLog {
         let readers = segments
             .into_iter()
             .map(|x| x.scan())
-            .collect::<std::io::Result<Vec<_>>>()?;
+            .collect::<crate::Result<Vec<_>>>()?;
 
         let reader = MergeReader::new(readers);