Rewrite forward reader, WIP everything else

5 files changed, 337 insertions, 137 deletions

diff --git a/log_db/src/common.rs b/log_db/src/common.rs
index c08f212..448226d 100644
--- a/log_db/src/common.rs
+++ b/log_db/src/common.rs
@@ -4,6 +4,7 @@ use std::fmt::Display;
 use std::fs::{metadata, File};
 use std::io::{self};
 use std::path::{Path, PathBuf};
+use uuid::Uuid;
 
 // For Unix-like systems
 #[cfg(unix)]
@@ -17,48 +18,8 @@ pub const ACTIVE_SYMLINK_FILENAME: &str = "active";
 pub const METADATA_FILE_HEADER_SIZE: usize = 24;
 pub const EXCL_LOCK_REQUEST_FILENAME: &str = "excl_lock_req";
 pub const DEFAULT_READ_BUF_SIZE: usize = 1024 * 1024; // 1 MB
-pub const FIELD_SEPARATOR: u8 = b'\x1C';
-pub const ESCAPE_CHARACTER: u8 = b'\x1D';
 pub const TEST_RESOURCES_DIR: &str = "tests/resources";
 
-// Special sequences. Note: these must have the same length!
-// Since the log is read both forwards and backwards, we must have a signal
-// character (ESCAPE_CHARACTER) on both sides of the special sequence.
-pub const SEQ_RECORD_SEP: &[u8] = &[
-    ESCAPE_CHARACTER,
-    FIELD_SEPARATOR,
-    FIELD_SEPARATOR,
-    ESCAPE_CHARACTER,
-];
-pub const SEQ_LIT_ESCAPE: &[u8] = &[
-    ESCAPE_CHARACTER,
-    ESCAPE_CHARACTER,
-    ESCAPE_CHARACTER,
-    ESCAPE_CHARACTER,
-];
-pub const SEQ_LIT_FIELD_SEP: &[u8] = &[
-    ESCAPE_CHARACTER,
-    ESCAPE_CHARACTER,
-    FIELD_SEPARATOR,
-    ESCAPE_CHARACTER,
-];
-
-/// There are three special sequences that need to be handled:
-/// Here: SC = escape char, FS = field separator.
-/// - SC FS FS SC  -> actual record separator
-/// - SC SC FS SC  -> literal FS
-/// - SC SC SC SC  -> literal SC
-///
-/// Returns SpecialSequence or None if not valid.
-pub fn validate_special(buf: &[u8]) -> Option<SpecialSequence> {
-    match buf {
-        SEQ_RECORD_SEP => Some(SpecialSequence::RecordSeparator),
-        SEQ_LIT_FIELD_SEP => Some(SpecialSequence::LiteralFieldSeparator),
-        SEQ_LIT_ESCAPE => Some(SpecialSequence::LiteralEscape),
-        _ => None,
-    }
-}
-
 #[derive(Debug, Eq, PartialEq)]
 pub enum SpecialSequence {
     RecordSeparator,
@@ -155,6 +116,11 @@ impl Ord for LogKeySet {
     }
 }
 
+pub struct MetadataHeader {
+    pub version: u8,
+    pub uuid: Uuid,
+}
+
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub enum WriteDurability {
     /// Changes are written to an application-level write buffer without flushing to the OS write buffer or syncing to disk.
diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs
index 868d0c9..b0de03f 100644
--- a/log_db/src/lib.rs
+++ b/log_db/src/lib.rs
@@ -17,7 +17,9 @@ use memtable_secondary::SecondaryMemtable;
 use std::collections::BTreeMap;
 use std::fmt::Debug;
 use std::fs::{self};
-use std::io::{self, Write};
+use std::io::Seek;
+use std::io::SeekFrom;
+use std::io::{self, Read, Write};
 use std::os::unix::fs::MetadataExt;
 use std::path::{Path, PathBuf};
 use std::thread;
@@ -141,8 +143,8 @@ struct Config<Field: Eq + Clone> {
 
 pub struct DB<Field: Eq + Clone + Debug> {
     config: Config<Field>,
-    log_path: PathBuf,
-    log_file: fs::File,
+    active_metadata_file: fs::File,
+    active_data_file: fs::File,
     primary_key_index: usize,
     primary_memtable: PrimaryMemtable,
     secondary_memtables: Vec<SecondaryMemtable>,
@@ -224,32 +226,33 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
             .collect();
 
         let active_symlink = Path::new(&config.data_dir).join(ACTIVE_SYMLINK_FILENAME);
-        fs::read_dir(&config.data_dir)?.for_each(|entry| if let Ok(entry) = entry {});
 
         let active_target = fs::read_link(&active_symlink)?;
-        let log_path = Path::new(&config.data_dir).join(active_target);
-        let log_file = fs::OpenOptions::new()
+        let active_metadata_path = Path::new(&config.data_dir).join(active_target);
+        let mut active_metadata_file = fs::OpenOptions::new()
             .read(true)
             .append(true)
-            .open(&log_path)?;
+            .open(&active_metadata_path)?;
 
-        let mut memtable_file = fs::OpenOptions::new().read(true).open(&log_path)?;
+        let active_metadata_header = DB::<Field>::read_metadata_header(&mut active_metadata_file)?;
+        let active_data_path =
+            Path::new(&config.data_dir).join(active_metadata_header.uuid.to_string());
+        let active_data_file = fs::OpenOptions::new()
+            .read(true)
+            .append(true)
+            .open(&active_data_path)?;
 
-        let mut db = DB::<Field> {
+        let db = DB::<Field> {
             config: config.clone(),
-            log_path,
-            log_file,
+            active_metadata_file,
+            active_data_file,
             primary_key_index,
             primary_memtable,
             secondary_memtables,
         };
 
-        info!("Rebuilding memtable indexes...");
-
-        let forward_log_reader = ForwardLogReader::new(&mut memtable_file);
-        for record in forward_log_reader {
-            db.insert_to_memtables(&record);
-        }
+        // info!("Rebuilding memtable indexes...");
+        // TODO FIXME build memtable indexes
 
         info!("Database ready.");
 
@@ -329,10 +332,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
         debug!("Lock acquired, appending to log file");
 
         // Write the record to the log
-        // Each serialized row is suffixed with the field separator character sequence
-        let mut serialized_record = record.serialize();
-        serialized_record.extend(SEQ_RECORD_SEP);
-        self.log_file.write_all(&serialized_record)?;
+        self.log_file.write_all(&record.serialize())?;
 
         // Flush and sync to disk
         if self.config.write_durability == WriteDurability::Flush {
@@ -768,9 +768,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
             .open(temp_path)?;
 
         for entry in map.values() {
-            let mut serialized = entry.serialize();
-            serialized.extend(SEQ_RECORD_SEP);
-            temp_file.write_all(&serialized)?;
+            temp_file.write_all(&entry.serialize())?;
         }
 
         fs::rename(&temp_path, path)?;
@@ -874,4 +872,19 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
 
         Ok(())
     }
+
+    /// Reads the metadata header from the metadata file.
+    /// Leaves the file seek head at the beginning of the records, after the header.
+    fn read_metadata_header(metadata_file: &mut fs::File) -> Result<MetadataHeader, io::Error> {
+        let mut version_buf = vec![0u8; 1];
+        metadata_file.seek(SeekFrom::Start(0))?;
+        metadata_file.read_exact(&mut version_buf)?;
+        let version = version_buf[0];
+
+        let mut uuid_buf = vec![0u8; 16];
+        metadata_file.seek_relative(7)?; // skip over padding
+        metadata_file.read_exact(&mut uuid_buf)?;
+        let uuid = Uuid::from_slice(&uuid_buf).expect("Invalid UUID");
+        Ok(MetadataHeader { version, uuid })
+    }
 }
diff --git a/log_db/src/log_reader_forward.rs b/log_db/src/log_reader_forward.rs
index 03726d0..808d3fa 100644
--- a/log_db/src/log_reader_forward.rs
+++ b/log_db/src/log_reader_forward.rs
@@ -1,81 +1,50 @@
 use super::common::*;
 use std::fs::{self};
-use std::io::{self, BufRead, Read};
+use std::io::{self, Read, Seek};
 
-pub struct ForwardLogReader<'a> {
-    reader: io::BufReader<&'a mut fs::File>,
+pub struct ForwardLogReader {
+    metadata_reader: io::BufReader<fs::File>,
+    data_reader: io::BufReader<fs::File>,
 }
 
-impl<'a> ForwardLogReader<'a> {
-    pub fn new(file: &mut fs::File) -> ForwardLogReader {
-        let reader = io::BufReader::new(file);
-        ForwardLogReader { reader }
+impl<'a> ForwardLogReader {
+    pub fn new(metadata_file: fs::File, data_file: fs::File) -> ForwardLogReader {
+        let mut ret = ForwardLogReader {
+            metadata_reader: io::BufReader::new(metadata_file),
+            data_reader: io::BufReader::new(data_file),
+        };
+
+        ret.metadata_reader
+            .seek(io::SeekFrom::Start(METADATA_FILE_HEADER_SIZE as u64))
+            .expect("Seek failed");
+
+        ret
     }
 
     fn read_record(&mut self) -> Result<Option<Record>, io::Error> {
-        // The buffer that stores the bytes read from the file.
-        let mut read_buf: Vec<u8> = Vec::new();
-        // The buffer that stores all the bytes of the record read so far in reverse order.
-        let mut result_buf: Vec<u8> = Vec::new();
-
-        // Try reading a byte from the file.
-        // If we've reached the end of the file, return None.
-        let mut peek_buf = vec![0];
-        match self.reader.read_exact(&mut peek_buf) {
-            Ok(_) => {
-                // Go back one byte
-                self.reader.seek_relative(-1)?;
-            }
-            Err(ref e) if e.kind() == io::ErrorKind::UnexpectedEof => {
+        let mut metadata_entry_buf = vec![0; 16]; // 2x u64
+        if let Err(e) = self.metadata_reader.read_exact(&mut metadata_entry_buf) {
+            if e.kind() == io::ErrorKind::UnexpectedEof {
                 return Ok(None);
-            }
-            Err(e) => {
+            } else {
                 return Err(e);
             }
         }
 
-        loop {
-            read_buf.clear();
-            self.reader.read_until(ESCAPE_CHARACTER, &mut read_buf)?;
-            self.reader.seek_relative(-1)?;
-            result_buf.extend(&read_buf[..read_buf.len() - 1]);
+        // First u64 is the offset of the record in the data file, second is the length of the record
+        let entry_offset = u64::from_be_bytes(metadata_entry_buf[0..8].try_into().unwrap());
+        let entry_length = u64::from_be_bytes(metadata_entry_buf[8..16].try_into().unwrap());
 
-            // Otherwise, we must have encountered an escape character.
-            match self.read_special_sequence()? {
-                SpecialSequence::RecordSeparator => {
-                    // The record is complete, so we can break out of the loop.
-                    break;
-                }
-                SpecialSequence::LiteralFieldSeparator => {
-                    // The field separator is escaped, so we need to add it to the result buffer.
-                    result_buf.push(FIELD_SEPARATOR);
-                }
-                SpecialSequence::LiteralEscape => {
-                    // The escape character is escaped, so we need to add it to the result buffer.
-                    result_buf.push(ESCAPE_CHARACTER);
-                }
-            }
-        }
+        self.data_reader.seek(io::SeekFrom::Start(entry_offset))?;
+        let mut result_buf = vec![0; entry_length as usize];
+        self.data_reader.read_exact(&mut result_buf)?;
 
         let record = Record::deserialize(&result_buf);
         Ok(Some(record))
     }
-
-    fn read_special_sequence(&mut self) -> Result<SpecialSequence, io::Error> {
-        let mut special_buf: Vec<u8> = vec![0; SEQ_RECORD_SEP.len()];
-        self.reader.read_exact(&mut special_buf)?;
-
-        match validate_special(&special_buf.as_slice()) {
-            Some(special) => Ok(special),
-            None => Err(io::Error::new(
-                io::ErrorKind::InvalidData,
-                "Not a special sequence",
-            )),
-        }
-    }
 }
 
-impl Iterator for ForwardLogReader<'_> {
+impl Iterator for ForwardLogReader {
     type Item = Record;
 
     fn next(&mut self) -> Option<Self::Item> {
@@ -94,28 +63,26 @@ mod tests {
 
     #[test]
     fn test_forward_log_reader_fixture_db1() {
-        let db_path = Path::new(TEST_RESOURCES_DIR).join("test_db1");
-        let mut file = fs::OpenOptions::new()
+        let metadata_path = Path::new(TEST_RESOURCES_DIR).join("test_metadata_1");
+        let data_path = Path::new(TEST_RESOURCES_DIR).join("test_data_1");
+        let metadata_file = fs::OpenOptions::new()
             .read(true)
-            .open(&db_path)
-            .expect("Failed to open file");
-        let mut forward_log_reader = ForwardLogReader::new(&mut file);
+            .open(&metadata_path)
+            .expect("Failed to open metadata file");
+        let data_file = fs::OpenOptions::new()
+            .read(true)
+            .open(&data_path)
+            .expect("Failed to open data file");
+
+        let mut forward_log_reader = ForwardLogReader::new(metadata_file, data_file);
 
-        // There are two records in the log with "schema": Int, Null
+        // There are two records in the log with "schema" with one field: Bytes
 
         let first_record = forward_log_reader
             .next()
             .expect("Failed to read the first record");
         assert!(match first_record.values.as_slice() {
-            [RecordValue::Int(0x1D), RecordValue::Null] => true,
-            _ => false,
-        });
-
-        let last_record = forward_log_reader
-            .next()
-            .expect("Failed to read the last record");
-        assert!(match last_record.values.as_slice() {
-            [RecordValue::Int(10), RecordValue::Null] => true,
+            [RecordValue::Bytes(_)] => true,
             _ => false,
         });
 
diff --git a/log_db/tests/resources/test_data_1 b/log_db/tests/resources/test_data_1
new file mode 100644
index 0000000..d4c6a21
--- /dev/null
+++ b/log_db/tests/resources/test_data_1
@@ -0,0 +1,254 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/log_db/tests/resources/test_metadata_1 b/log_db/tests/resources/test_metadata_1
new file mode 100644
index 0000000..9694599
--- /dev/null
+++ b/log_db/tests/resources/test_metadata_1