Rewrite everything to use metadata and data files, TODO memtables

5 files changed, 266 insertions, 478 deletions

diff --git a/log_db/src/common.rs b/log_db/src/common.rs
index 448226d..ce1a30f 100644
--- a/log_db/src/common.rs
+++ b/log_db/src/common.rs
@@ -17,6 +17,7 @@ use std::os::windows::fs::MetadataExt;
 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 INIT_LOCK_FILENAME: &str = "init_lock";
 pub const DEFAULT_READ_BUF_SIZE: usize = 1024 * 1024; // 1 MB
 pub const TEST_RESOURCES_DIR: &str = "tests/resources";
 
@@ -216,32 +217,26 @@ impl RecordValue {
             }
             RecordValue::Int(i) => {
                 let mut bytes = vec![1]; // Tag for Int
-                let data_bytes = escape_bytes(&i.to_be_bytes());
-                bytes.extend(&data_bytes);
+                bytes.extend(&i.to_be_bytes());
                 bytes
             }
             RecordValue::Float(f) => {
                 let mut bytes = vec![2]; // Tag for Float
-                let data_bytes = escape_bytes(&f.to_be_bytes());
-                bytes.extend(&data_bytes);
+                bytes.extend(&f.to_be_bytes());
                 bytes
             }
             RecordValue::String(s) => {
                 let mut bytes = vec![3]; // Tag for String
                 let length = s.len() as u64;
-                let length_bytes = escape_bytes(&length.to_be_bytes());
-                bytes.extend(&length_bytes);
-                let data_bytes = escape_bytes(s.as_bytes());
-                bytes.extend(&data_bytes);
+                bytes.extend(&length.to_be_bytes());
+                bytes.extend(s.as_bytes());
                 bytes
             }
             RecordValue::Bytes(b) => {
                 let mut bytes = vec![4]; // Tag for Bytes
                 let length = b.len() as u64;
-                let length_bytes = escape_bytes(&length.to_be_bytes());
-                bytes.extend(&length_bytes);
-                let data_bytes = escape_bytes(b);
-                bytes.extend(&data_bytes);
+                bytes.extend(&length.to_be_bytes());
+                bytes.extend(b);
                 bytes
             }
         }
@@ -319,22 +314,6 @@ impl Record {
     }
 }
 
-pub fn escape_bytes(buf: &[u8]) -> Vec<u8> {
-    let mut result = Vec::new();
-    for byte in buf {
-        match byte {
-            &FIELD_SEPARATOR => {
-                result.extend(SEQ_LIT_FIELD_SEP);
-            }
-            &ESCAPE_CHARACTER => {
-                result.extend(SEQ_LIT_ESCAPE);
-            }
-            _ => result.push(*byte),
-        }
-    }
-    result
-}
-
 /// A path to a log segment file along with its type
 pub enum SegmentPath {
     /// A symbolic link to the active log file
diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs
index b0de03f..320ec6c 100644
--- a/log_db/src/lib.rs
+++ b/log_db/src/lib.rs
@@ -14,7 +14,6 @@ pub use log_reader_forward::ForwardLogReader;
 pub use log_reader_reverse::ReverseLogReader;
 use memtable_primary::PrimaryMemtable;
 use memtable_secondary::SecondaryMemtable;
-use std::collections::BTreeMap;
 use std::fmt::Debug;
 use std::fs::{self};
 use std::io::Seek;
@@ -23,9 +22,6 @@ use std::io::{self, Read, Write};
 use std::os::unix::fs::MetadataExt;
 use std::path::{Path, PathBuf};
 use std::thread;
-use tempfile;
-use tempfile::tempfile_in;
-use tempfile::NamedTempFile;
 use uuid::Uuid;
 
 pub struct ConfigBuilder<Field: Eq + Clone + Debug> {
@@ -162,26 +158,43 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
         // After creation, the directory should always be in a complete state
         // without missing files.
         // A tempdir-move strategy is used to achieve one-phase commit.
-        if !fs::exists(&config.data_dir)?
-            || !fs::exists(&Path::new(&config.data_dir).join(ACTIVE_SYMLINK_FILENAME))?
-        {
-            let tmpdir = tempfile::tempdir()?;
-            let tmpdir_path = tmpdir.into_path();
 
-            let (segment_uuid, _) = DB::<Field>::create_segment_data_file(&tmpdir_path)?;
+        // Ensure the data directory exists
+        let data_dir_path = Path::new(&config.data_dir);
+        match fs::create_dir(&data_dir_path) {
+            Ok(_) => {}
+            Err(e) => {
+                if e.kind() != io::ErrorKind::AlreadyExists {
+                    return Err(e);
+                }
+            }
+        }
+
+        // Create an initialize lock file to prevent multiple concurrent initializations
+        let init_lock_file = fs::OpenOptions::new()
+            .create(true)
+            .write(true)
+            .open(&data_dir_path.join(INIT_LOCK_FILENAME))?;
+
+        init_lock_file.lock_exclusive()?;
+
+        // We have acquired the lock, check if the data directory is in a complete state
+        // If not, initialize it, otherwise skip.
+        if !fs::exists(data_dir_path.join(ACTIVE_SYMLINK_FILENAME))? {
+            let (segment_uuid, _) = DB::<Field>::create_segment_data_file(data_dir_path)?;
             let (segment_num, _) =
-                DB::<Field>::create_segment_metadata_file(&tmpdir_path, &segment_uuid)?;
-            DB::<Field>::set_active_segment(&tmpdir_path, segment_num)?;
+                DB::<Field>::create_segment_metadata_file(data_dir_path, &segment_uuid)?;
+            DB::<Field>::set_active_segment(data_dir_path, segment_num)?;
 
             // Create the exclusive lock request file
             fs::OpenOptions::new()
                 .create(true)
                 .write(true)
-                .open(&tmpdir_path.join(EXCL_LOCK_REQUEST_FILENAME))?;
-
-            fs::rename(tmpdir_path, &config.data_dir)?;
+                .open(data_dir_path.join(EXCL_LOCK_REQUEST_FILENAME))?;
         }
 
+        init_lock_file.unlock()?;
+
         // Calculate the index of the primary value in a record
         let primary_key_index = config
             .fields
@@ -238,7 +251,6 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
         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)?;
 
@@ -322,9 +334,9 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
         debug!("Opening file in append mode and acquiring exclusive lock...");
 
         // Acquire an exclusive lock for writing
-        self.request_exclusive_lock()?;
+        self.request_exclusive_lock_on_active()?;
 
-        if self.ensure_correct_file_is_open()? {
+        if self.ensure_active_file_is_open()? {
             // The log file has been rotated, so we must try again
             return self.upsert(record);
         }
@@ -332,18 +344,37 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
         debug!("Lock acquired, appending to log file");
 
         // Write the record to the log
-        self.log_file.write_all(&record.serialize())?;
+        let serialized = &record.serialize();
+        let record_offset = self.active_data_file.metadata()?.len();
+        let record_length = serialized.len() as u64;
+        self.active_data_file.write_all(serialized)?;
 
-        // Flush and sync to disk
+        // Flush and sync data to disk
         if self.config.write_durability == WriteDurability::Flush {
-            self.log_file.flush()?;
+            self.active_data_file.flush()?;
         }
         if self.config.write_durability == WriteDurability::FlushSync {
-            self.log_file.flush()?;
-            self.log_file.sync_all()?;
+            self.active_data_file.flush()?;
+            self.active_data_file.sync_all()?;
         }
 
-        self.log_file.unlock()?;
+        // Write the record metadata to the metadata file
+        let mut metadata_buf = vec![];
+        metadata_buf.extend(&record_offset.to_be_bytes());
+        metadata_buf.extend(&record_length.to_be_bytes());
+        self.active_metadata_file.write_all(&metadata_buf)?;
+
+        // Flush and sync metadata to disk
+        if self.config.write_durability == WriteDurability::Flush {
+            self.active_metadata_file.flush()?;
+        }
+        if self.config.write_durability == WriteDurability::FlushSync {
+            self.active_metadata_file.flush()?;
+            self.active_metadata_file.sync_all()?;
+        }
+
+        self.active_data_file.unlock()?;
+        self.active_metadata_file.unlock()?;
 
         debug!("Record appended to log file, lock released");
 
@@ -382,65 +413,48 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
             "Matching records based on value at primary key index ({})",
             &self.primary_key_index
         );
-        debug!("Opening file in read mode and acquiring shared lock...");
 
-        // Open the file and acquire a shared lock for reading
-        let mut file = fs::OpenOptions::new().read(true).open(&self.log_path)?;
+        let data_dir_path = Path::new(&self.config.data_dir);
+        let greatest = DB::<Field>::greatest_segment_number(&data_dir_path)?;
+        debug!("Searching segments {} through 1", greatest);
 
-        self.request_shared_lock(&mut file)?;
+        let mut found_record: Option<Record> = None;
+        for segment_num in (1..=greatest).rev() {
+            let segment_path = data_dir_path.join(format!("metadata.{}", segment_num));
 
-        if !is_file_same_as_path(&file, &self.log_path)? {
-            // The log file has been rotated, so we must try again
-            debug!("Lock acquired, but the log file has been rotated. Retrying get...");
-            file.unlock()?;
-            drop(file);
-            return self.get(query_key_original);
-        }
+            debug!(
+                "Opening segment {} in read mode and acquiring shared lock...",
+                segment_num
+            );
 
-        debug!("Lock acquired, searching log files for record");
+            let mut metadata_file = fs::OpenOptions::new().read(true).open(&segment_path)?;
 
-        let greatest = DB::<Field>::greatest_segment_number(&Path::new(&self.config.data_dir))?;
-        let segment_numbers = (0..=greatest); // FIXME all of this needs fixing
-        let mut result: Option<Record> = None;
-        for n in segment_numbers {
-            if n == 0 {
-                debug!("Searching the active log file...");
-                result = ReverseLogReader::new(&mut file)?.find(|record| {
-                    let record_key = record.values[self.primary_key_index]
-                        .as_indexable()
-                        .expect("A non-indexable value was stored at key index");
-                    record_key == query_key
-                });
+            self.request_shared_lock(&mut metadata_file)?;
 
-                debug!("Active log file searched, releasing shared lock...");
-                file.unlock()?;
-            } else {
-                debug!("Locking and searching rotated log segment file {}...", n);
-                let path = Path::new(&self.config.data_dir)
-                    .join(ACTIVE_SYMLINK_FILENAME)
-                    .with_extension(n.to_string());
+            let metadata_header = DB::<Field>::read_metadata_header(&mut metadata_file)?;
+            let data_path = data_dir_path.join(metadata_header.uuid.to_string());
+            let data_file = fs::OpenOptions::new().read(true).open(&data_path)?;
 
-                let mut segm_file = fs::OpenOptions::new().read(true).open(&path)?;
-                self.request_shared_lock(&mut segm_file)?;
-                result = ReverseLogReader::new(&mut segm_file)?.find(|record| {
-                    let record_key = record.values[self.primary_key_index]
-                        .as_indexable()
-                        .expect("A non-indexable value was stored at key index");
-                    record_key == query_key
-                });
+            // We should not "request_shared_lock()" here because we do not want
+            // to give way to writers at this point. That would possibly lead to a deadlock.
+            data_file.lock_shared()?;
 
-                debug!("Segment file searched, releasing shared lock...");
-                segm_file.unlock()?;
-            };
+            let mut reader = ReverseLogReader::new(metadata_file, data_file)?;
 
-            if result.is_some() {
+            if let Some(found) = reader.find(|record| {
+                let record_key = record.values[self.primary_key_index]
+                    .as_indexable()
+                    .expect("Primary key must be indexable");
+                record_key == query_key
+            }) {
+                found_record = Some(found);
                 break;
             }
         }
 
         debug!("Record search complete");
 
-        let result_value = match &result {
+        let result_value = match &found_record {
             Some(record) => record,
             None => {
                 debug!("No record found for key {:?}", query_key);
@@ -453,7 +467,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
         debug!("Updating memtables");
         self.insert_to_memtables(&result_value);
 
-        Ok(result)
+        Ok(found_record)
     }
 
     /// Get a collection of records based on a field value.
@@ -512,39 +526,52 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
                 "Key not found in schema after initialize",
             ))?;
 
-        debug!("Matching key index {}", key_index);
-        debug!("Acquiring shared lock...");
+        let data_dir_path = Path::new(&self.config.data_dir);
+        let greatest = DB::<Field>::greatest_segment_number(&data_dir_path)?;
 
-        // Acquire a shared lock for reading
-        self.log_file.lock_shared()?;
+        let mut found_records = vec![];
+        for segment_num in (1..=greatest).rev() {
+            let segment_path = data_dir_path.join(format!("metadata.{}", segment_num));
 
-        if self.ensure_correct_file_is_open()? {
-            // The log file has been rotated, so we must try again
-            return self.find_all(field, query_key_original);
-        }
+            debug!(
+                "Opening segment {} in read mode and acquiring shared lock...",
+                segment_num
+            );
+
+            let mut metadata_file = fs::OpenOptions::new().read(true).open(&segment_path)?;
 
-        debug!("Lock acquired, searching log file for record");
+            self.request_shared_lock(&mut metadata_file)?;
 
-        let result = ReverseLogReader::new(&mut self.log_file)?
-            .filter(|record| {
+            let metadata_header = DB::<Field>::read_metadata_header(&mut metadata_file)?;
+            let data_path = &data_dir_path.join(metadata_header.uuid.to_string());
+            let data_file = fs::OpenOptions::new().read(true).open(&data_path)?;
+
+            // We should not "request_shared_lock()" here because we do not want
+            // to give way to writers at this point. That would possibly lead to a deadlock.
+            data_file.lock_shared()?;
+
+            let reader = ReverseLogReader::new(metadata_file, data_file)?;
+
+            for record in reader {
                 let record_key = record.values[key_index]
                     .as_indexable()
-                    .expect("A non-indexable value was stored at key index");
-                record_key == query_key
-            })
-            .collect::<Vec<Record>>();
+                    .expect("Secondary key must be indexable");
+                if record_key == query_key {
+                    found_records.push(record);
+                }
+            }
+        }
 
-        self.log_file.unlock()?;
-        debug!("Record search complete, lock released");
+        debug!("Record search complete");
 
         debug!(
             "Number of matching records found in log file: {}",
-            result.len()
+            found_records.len()
         );
 
         if let Some(memtable_index) = found_memtable_index {
             debug!("Inserting result set into secondary index");
-            let primary_values: Vec<IndexableValue> = result
+            let primary_values: Vec<IndexableValue> = found_records
                 .iter()
                 .map(|r| {
                     r.values[self.primary_key_index]
@@ -555,7 +582,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
             self.secondary_memtables[memtable_index].set_all(&query_key, &primary_values);
         }
 
-        Ok(result)
+        Ok(found_records)
     }
 
     fn get_secondary_memtable_index_by_field(&self, field: &Field) -> Option<usize> {
@@ -565,26 +592,34 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
             .position(|schema_field| schema_field == field)
     }
 
-    /// Ensures that the `self.log_file` handle is still pointing to the correct file.
-    /// If the file has been rotated, the handle will be closed and reopened.
+    /// Ensures that the `self.metadata_file` and `self.data_file` handles are still pointing to the correct files.
+    /// If the segment has been rotated, the handle will be closed and reopened.
     /// Returns `true` if the file has been rotated and the handle has been reopened.
-    fn ensure_correct_file_is_open(&mut self) -> Result<bool, io::Error> {
-        if !is_file_same_as_path(&self.log_file, &self.log_path)? {
-            // The log file has been rotated, so we must try again
-            debug!(
-                "Lock acquired, but the log file has been rotated. Reopening file and retrying..."
-            );
-            self.log_file.unlock()?;
+    fn ensure_active_file_is_open(&mut self) -> Result<bool, io::Error> {
+        let data_dir_path = Path::new(&self.config.data_dir);
+        let active_target = fs::read_link(data_dir_path.join("active"))?;
+        let active_metadata_path = data_dir_path.join(active_target);
 
-            self.log_file = fs::OpenOptions::new()
-                .create(true)
+        let correct = is_file_same_as_path(&self.active_metadata_file, &active_metadata_path)?;
+        if !correct {
+            debug!("Metadata file has been rotated. Reopening...");
+            let mut metadata_file = fs::OpenOptions::new()
                 .read(true)
-                .append(true)
-                .open(&self.log_path)?;
+                .write(true)
+                .open(&active_metadata_path)?;
+
+            self.request_shared_lock(&mut metadata_file)?;
+
+            let metadata_header =
+                DB::<Field>::read_metadata_header(&mut self.active_metadata_file)?;
+            let data_file_path = data_dir_path.join(metadata_header.uuid.to_string());
 
-            Ok(true)
+            self.active_metadata_file = metadata_file;
+            self.active_data_file = fs::OpenOptions::new().append(true).open(&data_file_path)?;
+
+            return Ok(true);
         } else {
-            Ok(false)
+            return Ok(false);
         }
     }
 
@@ -619,7 +654,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
         }
     }
 
-    fn request_exclusive_lock(&mut self) -> Result<(), io::Error> {
+    fn request_exclusive_lock_on_active(&mut self) -> Result<(), io::Error> {
         // Create a lock on the exclusive lock request file to signal to readers that they should wait
         let lock_request_path = Path::new(&self.config.data_dir).join(EXCL_LOCK_REQUEST_FILENAME);
         let lock_request_file = fs::OpenOptions::new()
@@ -641,8 +676,9 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
         //     ));
         // }
 
-        // Acquire an exclusive lock on the log file
-        self.log_file.lock_exclusive()?;
+        // Acquire an exclusive lock on the segment files
+        self.active_metadata_file.lock_exclusive()?;
+        self.active_data_file.lock_exclusive()?;
 
         // Unlock the request file
         lock_request_file.unlock()?;
@@ -705,38 +741,32 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
     /// You may call this function in a separate thread or process to avoid blocking the main thread.
     /// However, the database will be exclusively locked, so all writes will be blocked during the tasks.
     pub fn do_maintenance_tasks(&mut self) -> Result<(), io::Error> {
-        let active_log_path = Path::new(&self.config.data_dir).join(ACTIVE_SYMLINK_FILENAME);
+        let data_dir = self.config.data_dir.to_owned();
+        let data_dir_path = Path::new(&data_dir);
+        let active_log_path = data_dir_path.join(ACTIVE_SYMLINK_FILENAME);
         let active_log_md = fs::metadata(&active_log_path)?;
 
         if active_log_md.size() >= self.config.segment_size as u64 {
             // Rotate the active log file
 
             debug!("Starting rotation, requesting exclusive lock...");
-            self.request_exclusive_lock()?;
+            self.request_exclusive_lock_on_active()?;
 
             debug!("Exclusive lock acquired, rotating active log file...");
-            let next_segment_number =
-                DB::<Field>::greatest_segment_number(&Path::new(&self.config.data_dir))? + 1;
-            let next_segment_path =
-                &active_log_path.with_extension(next_segment_number.to_string());
-
-            debug!("Renaming active log file to {:?}", &next_segment_path);
-            fs::rename(&active_log_path, &next_segment_path)?;
 
-            // Create a new active log file
-            self.log_file = fs::OpenOptions::new()
-                .create(true)
-                .write(true)
-                .append(true)
-                .open(&active_log_path)?;
+            // Create a new active log segment
+            let (data_file_uuid, _) = DB::<Field>::create_segment_data_file(data_dir_path)?;
+            let (new_segment_num, new_segment_path) =
+                DB::<Field>::create_segment_metadata_file(data_dir_path, &data_file_uuid)?;
+            DB::<Field>::set_active_segment(data_dir_path, new_segment_num)?;
 
             // The new active log file is not locked by this client so it cannot be touched.
-            debug!("Active log file rotated");
+            debug!("Active log file rotated, new segment: {}", new_segment_num);
 
             // Compact the rotated segment without a lock.
             // Since the rotated segment and the compacted segment based on it will be
             // a) read-only, and b) identical in effective content, there is no need to lock it.
-            self.compact_segment(&next_segment_path)?;
+            self.compact_segment(&new_segment_path)?;
 
             debug!("Segment compacted");
         }
@@ -745,34 +775,36 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
     }
 
     fn compact_segment(&self, path: &Path) -> Result<(), io::Error> {
-        debug!("Opening segment file {:?} for compaction", path);
-        let mut segment_file = fs::OpenOptions::new().read(true).open(path)?;
+        warn!("TODO compact_segment");
+        // debug!("Opening segment file {:?} for compaction", path);
+        // let mut segment_file = fs::OpenOptions::new().read(true).open(path)?;
 
-        debug!("Reading segment data into a BTreeMap");
-        let mut map = BTreeMap::new();
-        let forward_log_reader = ForwardLogReader::new(&mut segment_file);
-        for entry in forward_log_reader {
-            let primary_key = entry.values[self.primary_key_index]
-                .as_indexable()
-                .expect("Primary key was not indexable");
-            map.insert(primary_key, entry);
-        }
+        // debug!("Reading segment data into a BTreeMap");
+        // let mut map = BTreeMap::new();
+        // let forward_log_reader = ForwardLogReader::new(&mut segment_file);
+        // for entry in forward_log_reader {
+        //     let primary_key = entry.values[self.primary_key_index]
+        //         .as_indexable()
+        //         .expect("Primary key was not indexable");
+        //     map.insert(primary_key, entry);
+        // }
 
-        debug!("Opening temporary file for writing compacted data");
-        let temp_file = tempfile::NamedTempFile::new()?;
-        let temp_path = temp_file.as_ref();
+        // debug!("Opening temporary file for writing compacted data");
+        // let temp_file = tempfile::NamedTempFile::new()?;
+        // let temp_path = temp_file.as_ref();
 
-        let mut temp_file = fs::OpenOptions::new()
-            .create(true)
-            .append(true)
-            .open(temp_path)?;
+        // let mut temp_file = fs::OpenOptions::new()
+        //     .create(true)
+        //     .append(true)
+        //     .open(temp_path)?;
 
-        for entry in map.values() {
-            temp_file.write_all(&entry.serialize())?;
-        }
+        // for entry in map.values() {
+        //     temp_file.write_all(&entry.serialize())?;
+        // }
 
-        fs::rename(&temp_path, path)?;
+        // fs::rename(&temp_path, path)?;
 
+        // Ok(())
         Ok(())
     }
 
diff --git a/log_db/src/log_reader_forward.rs b/log_db/src/log_reader_forward.rs
index 808d3fa..1b34dfc 100644
--- a/log_db/src/log_reader_forward.rs
+++ b/log_db/src/log_reader_forward.rs
@@ -31,13 +31,26 @@ impl<'a> ForwardLogReader {
             }
         }
 
+        debug!("Read 16 bytes from metadata file: {:?}", metadata_entry_buf);
+
         // 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());
 
-        self.data_reader.seek(io::SeekFrom::Start(entry_offset))?;
+        debug!(
+            "Read offset {} and length {} from metadata file",
+            entry_offset, entry_length
+        );
+
+        // Use .seek_relative instead of .seek to avoid dropping the BufReader internal buffer when
+        // the seek distance is small
+        let seek_distance = entry_offset - self.data_reader.stream_position()?;
+        self.data_reader.seek_relative(seek_distance as i64)?;
+
         let mut result_buf = vec![0; entry_length as usize];
+        debug!("Reading {} bytes from data file", entry_length);
         self.data_reader.read_exact(&mut result_buf)?;
+        debug!("Read {} bytes from data file", result_buf.len());
 
         let record = Record::deserialize(&result_buf);
         Ok(Some(record))
@@ -63,6 +76,7 @@ mod tests {
 
     #[test]
     fn test_forward_log_reader_fixture_db1() {
+        let _ = env_logger::builder().is_test(true).try_init();
         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()
@@ -82,7 +96,7 @@ mod tests {
             .next()
             .expect("Failed to read the first record");
         assert!(match first_record.values.as_slice() {
-            [RecordValue::Bytes(_)] => true,
+            [RecordValue::Bytes(bytes)] => bytes.len() == 256,
             _ => false,
         });
 
diff --git a/log_db/src/log_reader_reverse.rs b/log_db/src/log_reader_reverse.rs
index f042406..ce17e59 100644
--- a/log_db/src/log_reader_reverse.rs
+++ b/log_db/src/log_reader_reverse.rs
@@ -1,345 +1,109 @@
 use super::common::*;
 use std::fs::{self};
-use std::io::{self, Read, Seek, SeekFrom};
+use std::io::{self, Read, Seek};
 
-pub struct ReverseLogReader<'a> {
-    /// The file to read from end to beginning.
-    file: &'a mut fs::File,
-    /// The internal buffer used to read from the file.
-    /// It is populated with the last INTERNAL_BUF_SIZE bytes read from the file
-    /// and is used to read records in reverse order
-    internal_buf: Vec<u8>,
-    /// The current position in the internal buffer. It is decremented as bytes are read
-    /// from the buffer. When a read is requested and the internal position is 0, the buffer
-    /// is populated with the next (= closer to the start of the file) INTERNAL_BUF_SIZE bytes from the file.
-    /// Note: This is the index of the next byte to be read from the internal buffer + 1
-    internal_pos: usize,
-    /// A flag indicating whether the record separator at the cursor position has been consumed.
-    /// Useful to avoid consuming the separator once when reading until an escape character, and
-    /// a second time when reading a new record and validating it ends in a separator.
-    consumed_record_sep: bool,
-}
-
-// This value is based on the reverse_read_file_with_various_buffer_sizes benchmark.
-// Greater values yield little to no performance improvement.
-const DEFAULT_INTERNAL_BUF_SIZE: usize = 32768;
-impl<'a> ReverseLogReader<'a> {
-    pub fn new(file: &mut fs::File) -> Result<ReverseLogReader, io::Error> {
-        file.seek(SeekFrom::End(0))?;
-        Ok(ReverseLogReader {
-            file,
-            internal_buf: vec![0; DEFAULT_INTERNAL_BUF_SIZE],
-            internal_pos: 0,
-            consumed_record_sep: false,
-        })
-    }
-
-    pub fn new_with_size(
-        file: &mut fs::File,
-        internal_buf_size: usize,
-    ) -> Result<ReverseLogReader, io::Error> {
-        file.seek(SeekFrom::End(0))?;
-        Ok(ReverseLogReader {
-            file,
-            internal_buf: vec![0; internal_buf_size],
-            internal_pos: 0,
-            consumed_record_sep: false,
-        })
-    }
-
-    pub fn read_record(&mut self) -> Result<Option<Record>, io::Error> {
-        if self.file.stream_position()? == 0 && self.internal_pos == 0 {
-            return Ok(None);
-        }
-
-        if !self.consumed_record_sep {
-            // Check that record ends with a record separator
-            match self.read_special_sequence()? {
-                SpecialSequence::RecordSeparator => {}
-                _ => {
-                    return Err(io::Error::new(
-                        io::ErrorKind::InvalidData,
-                        "Record does not end in record separator",
-                    ));
-                }
-            }
-        }
-        self.consumed_record_sep = false;
-
-        let mut result_buf: Vec<u8> = vec![];
-        let mut read_buf = Vec::with_capacity(self.internal_buf.len());
-        loop {
-            read_buf.clear();
-            let read = self.read_until(ESCAPE_CHARACTER, &mut read_buf)?;
+pub struct ReverseLogReader {
+    /// The contents of the metadata file are read into this buffer in one go.
+    metadata_buf: Vec<u8>,
 
-            result_buf.extend(&read_buf[..read]);
+    /// The data log file that is read based on offset + length information from the metadata file.
+    data_reader: io::BufReader<fs::File>,
 
-            if self.file.stream_position()? == 0 && self.internal_pos == 0 {
-                // We read until the start of the file, we are done
-                break;
-            }
+    /// The current position in metadata_buf.
+    metadata_pos: usize,
+}
 
-            match self.read_special_sequence()? {
-                SpecialSequence::LiteralEscape => {
-                    result_buf.push(ESCAPE_CHARACTER);
-                }
-                SpecialSequence::LiteralFieldSeparator => {
-                    result_buf.push(FIELD_SEPARATOR);
-                }
-                SpecialSequence::RecordSeparator => {
-                    self.consumed_record_sep = true;
-                    break;
-                }
-            }
-        }
+impl ReverseLogReader {
+    pub fn new(
+        mut metadata_file: fs::File,
+        data_file: fs::File,
+    ) -> Result<ReverseLogReader, io::Error> {
+        let mut metadata_buf = vec![];
+        metadata_file.seek(io::SeekFrom::Start(0))?;
+        metadata_file.read_to_end(&mut metadata_buf)?;
+        let len = metadata_buf.len();
 
-        result_buf.reverse();
-        Ok(Some(Record::deserialize(&result_buf)))
+        let ret = ReverseLogReader {
+            metadata_buf,
+            data_reader: io::BufReader::new(data_file),
+            metadata_pos: len,
+        };
+        Ok(ret)
     }
 
-    /// Read exactly `buf.len()` bytes from the file, return an error if the file is exhausted.
-    /// The bytes are returned in start -> end order.
-    /// If an error is returned, the contents of `buf` are in an undefined state.
-    fn read_exact(&mut self, buf: &mut [u8]) -> Result<usize, io::Error> {
-        let mut read = 0;
-        while read < buf.len() {
-            if self.internal_pos == 0 {
-                let populated_n = self.populate_internal_buf()?;
-                if populated_n == 0 {
-                    return Err(io::Error::new(
-                        io::ErrorKind::UnexpectedEof,
-                        "Unexpected end of file",
-                    ));
-                }
-            }
+    fn read_record(&mut self) -> Result<Option<Record>, io::Error> {
+        assert!(
+            (self.metadata_pos - METADATA_FILE_HEADER_SIZE) % 16 == 0,
+            "metadata_pos is not aligned"
+        );
 
-            let end = buf.len() - read;
-            let n = std::cmp::min(self.internal_pos, end);
-            buf[end - n..end]
-                .copy_from_slice(&self.internal_buf[self.internal_pos - n..self.internal_pos]);
-            read += n;
-            self.internal_pos -= n;
+        // Return None if we have read the entire metadata file and
+        // reached the end of the header
+        if self.metadata_pos == METADATA_FILE_HEADER_SIZE {
+            return Ok(None);
         }
 
-        Ok(read)
-    }
-
-    fn populate_internal_buf(&mut self) -> Result<usize, io::Error> {
-        let current_seek_pos = self.file.stream_position()? as usize;
-
-        // Seek back by the size of the internal buffer or to the beginning of the file
-        let seek_length = if current_seek_pos > self.internal_buf.len() {
-            self.internal_pos = self.internal_buf.len();
-            self.internal_buf.len()
-        } else {
-            self.internal_buf = vec![0; current_seek_pos as usize];
-            self.internal_pos = current_seek_pos as usize;
-            current_seek_pos
-        };
-
-        self.file.seek_relative(-(seek_length as i64))?;
-        self.file.read_exact(&mut self.internal_buf)?;
-        self.file.seek_relative(-(seek_length as i64))?;
+        self.metadata_pos -= 16;
+        let i = self.metadata_pos; // shorter alias
 
-        Ok(seek_length as usize)
-    }
-
-    /// Iterate over the internal buffer with internal_pos as the index.
-    /// If the byte is found or the file has been exhausted, we return the number of bytes read.
-    /// The `buf` parameter is used to store the bytes read from the internal buffer, excluding the found byte,
-    /// in reverse order.
-    fn read_until(&mut self, byte: u8, buf: &mut Vec<u8>) -> Result<usize, io::Error> {
-        // TODO: optimize this
-        let mut read = 0;
-        loop {
-            // If we reach internal_pos == 0, we need to populate the internal buffer.
-            if self.internal_pos == 0 {
-                let populated_n = self.populate_internal_buf()?;
-                if populated_n == 0 {
-                    return Ok(read);
-                }
-            }
+        // 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(self.metadata_buf[i..i + 8].try_into().unwrap());
+        let entry_length = u64::from_be_bytes(self.metadata_buf[i + 8..i + 16].try_into().unwrap());
 
-            while self.internal_pos > 0 {
-                let index = self.internal_pos - 1;
-                if self.internal_buf[index] == byte {
-                    return Ok(read);
-                }
-                buf.push(self.internal_buf[index]);
-                read += 1;
-                self.internal_pos -= 1;
-            }
-        }
-    }
+        debug!(
+            "Read offset {} and length {} from metadata file at position {}",
+            entry_offset, entry_length, i
+        );
 
-    fn read_special_sequence(&mut self) -> Result<SpecialSequence, io::Error> {
-        let mut special_buf: Vec<u8> = vec![0; SEQ_RECORD_SEP.len()];
-        self.read_exact(&mut special_buf)?;
+        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)?;
 
-        match validate_special(&special_buf.as_slice()) {
-            Some(special) => Ok(special),
-            None => {
-                let pos = self.file.stream_position().unwrap() + self.internal_pos as u64;
+        let record = Record::deserialize(&result_buf);
 
-                Err(io::Error::new(
-                    io::ErrorKind::InvalidData,
-                    format!(
-                        "Not a special sequence: {:?} at pos: {:x}",
-                        special_buf, pos,
-                    ),
-                ))
-            }
-        }
+        assert!(
+            (self.metadata_pos - METADATA_FILE_HEADER_SIZE) % 16 == 0,
+            "metadata_pos is not aligned"
+        );
+        Ok(Some(record))
     }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
-    use std::io::Write;
     use std::path::Path;
 
     #[test]
-    fn test_read_until_found() {
-        let mut file = tempfile::tempfile().unwrap();
-        file.write_all(b"hello,world").unwrap();
-        let mut reader = ReverseLogReader::new(&mut file).unwrap();
-        let mut buf = vec![];
-        assert_eq!(reader.read_until(b',', &mut buf).unwrap(), 5);
-        assert_eq!(buf, b"dlrow");
-    }
-
-    #[test]
-    fn test_read_until_not_found() {
-        let mut file = tempfile::tempfile().unwrap();
-        file.write_all(b"hello,world").unwrap();
-        let mut reader = ReverseLogReader::new(&mut file).unwrap();
-        let mut buf = vec![];
-        let read = reader.read_until(b'!', &mut buf).unwrap();
-        assert_eq!(buf, b"dlrow,olleh");
-        assert_eq!(read, 11);
-    }
-
-    #[test]
-    fn test_read_special_sequence() {
-        let mut file = tempfile::tempfile().unwrap();
-        let mut buf = vec![];
-        buf.extend(SEQ_RECORD_SEP);
-        buf.extend(SEQ_LIT_ESCAPE);
-        buf.extend(SEQ_LIT_FIELD_SEP);
-        // Note: written in start -> end order, read end -> start
-        file.write_all(&buf).unwrap();
-
-        let mut reader = ReverseLogReader::new(&mut file).unwrap();
-        assert_eq!(
-            reader.read_special_sequence().unwrap(),
-            SpecialSequence::LiteralFieldSeparator
-        );
-        assert_eq!(
-            reader.read_special_sequence().unwrap(),
-            SpecialSequence::LiteralEscape
-        );
-        assert_eq!(
-            reader.read_special_sequence().unwrap(),
-            SpecialSequence::RecordSeparator
-        );
-    }
-
-    #[test]
-    fn test_populate_internal_buf() {
-        let mut file = tempfile::tempfile().unwrap();
-        file.write_all(b"hello,world").unwrap();
-        let mut reader = ReverseLogReader::new_with_size(&mut file, 3).unwrap();
-
-        reader.populate_internal_buf().unwrap();
-        assert_eq!(
-            String::from_utf8(reader.internal_buf.clone()).unwrap(),
-            "rld".to_string()
-        );
-
-        reader.populate_internal_buf().unwrap();
-        assert_eq!(
-            String::from_utf8(reader.internal_buf.clone()).unwrap(),
-            ",wo".to_string()
-        );
-
-        reader.populate_internal_buf().unwrap();
-        assert_eq!(
-            String::from_utf8(reader.internal_buf.clone()).unwrap(),
-            "llo".to_string()
-        );
-
-        reader.populate_internal_buf().unwrap();
-        assert_eq!(
-            String::from_utf8(reader.internal_buf.clone()).unwrap(),
-            "he".to_string()
-        );
-    }
-
-    #[test]
     fn test_reverse_log_reader_fixture_db1() {
-        let db_path = Path::new(TEST_RESOURCES_DIR).join("test_db1");
-        let mut file = fs::OpenOptions::new()
+        let _ = env_logger::builder().is_test(true).try_init();
+        let metadata_path = Path::new(TEST_RESOURCES_DIR).join("test_metadata_1");
+        let metadata_file = fs::OpenOptions::new()
             .read(true)
-            .open(&db_path)
+            .open(&metadata_path)
             .expect("Failed to open file");
-        let mut reverse_log_reader = ReverseLogReader::new(&mut file).unwrap();
+        let data_file = fs::OpenOptions::new()
+            .read(true)
+            .open(Path::new(TEST_RESOURCES_DIR).join("test_data_1"))
+            .expect("Failed to open file");
+
+        let mut reverse_log_reader = ReverseLogReader::new(metadata_file, data_file).unwrap();
 
-        // There are two records in the log with "schema": Int, Null
+        // There are two records in the log with "schema": Int
 
         let last_record = reverse_log_reader
             .next()
             .expect("Failed to read the last record");
         assert!(match last_record.values.as_slice() {
-            [RecordValue::Int(10), RecordValue::Null] => true,
-            _ => false,
-        });
-
-        let first_record = reverse_log_reader
-            .next()
-            .expect("Failed to read the first record");
-        assert!(match first_record.values.as_slice() {
-            // Note: the int value is equal to the escape byte
-            [RecordValue::Int(0x1D), RecordValue::Null] => true,
+            [RecordValue::Bytes(bytes)] => bytes.len() == 256,
             _ => false,
         });
 
         assert!(reverse_log_reader.next().is_none());
     }
-
-    #[test]
-    fn test_read_exact() {
-        let mut file = tempfile::tempfile().unwrap();
-        file.write_all(b"hello,world").unwrap();
-        let mut reader = ReverseLogReader::new(&mut file).unwrap();
-        let mut buf = vec![0; 3];
-
-        let read = reader.read_exact(&mut buf).unwrap();
-        assert_eq!(buf, b"rld");
-        assert_eq!(read, 3);
-
-        let read = reader.read_exact(&mut buf).unwrap();
-        assert_eq!(buf, b",wo");
-        assert_eq!(read, 3);
-
-        let read = reader.read_exact(&mut buf).unwrap();
-        assert_eq!(buf, b"llo");
-        assert_eq!(read, 3);
-
-        assert!(reader.read_exact(&mut buf).unwrap_err().kind() == io::ErrorKind::UnexpectedEof);
-    }
-
-    #[test]
-    fn test_read_exact_insufficient_bytes() {
-        let mut file = tempfile::tempfile().unwrap();
-        file.write_all(b"hello").unwrap();
-        let mut reader = ReverseLogReader::new(&mut file).unwrap();
-        let mut buf = vec![0; 10];
-        assert!(reader.read_exact(&mut buf).unwrap_err().kind() == io::ErrorKind::UnexpectedEof);
-    }
 }
 
-impl Iterator for ReverseLogReader<'_> {
+impl Iterator for ReverseLogReader {
     type Item = Record;
 
     fn next(&mut self) -> Option<Self::Item> {
@@ -347,7 +111,7 @@ impl Iterator for ReverseLogReader<'_> {
             Ok(Some(record)) => Some(record),
             Ok(None) => None,
             Err(err) => {
-                panic!("Error reading record: {:?}", err,)
+                panic!("Error reading record: {:?}", err)
             }
         }
     }
diff --git a/log_db/src/memtable_secondary.rs b/log_db/src/memtable_secondary.rs
index b4ebb8f..3f9863b 100644
--- a/log_db/src/memtable_secondary.rs
+++ b/log_db/src/memtable_secondary.rs
@@ -1,7 +1,6 @@
 use super::*;
 use std::collections::BTreeMap;
 use std::collections::HashSet;
-use std::fmt::Debug;
 
 pub struct SecondaryMemtable {
     /// Map of records indexed by key. The value is the set of primary key values of records