Implement forward log reader, rebuild memtable at init

1 files changed, 104 insertions, 19 deletions

diff --git a/src/log_reader.rs b/src/log_reader.rs
index 0ee0a81..e32ef53 100644
--- a/src/log_reader.rs
+++ b/src/log_reader.rs
@@ -1,16 +1,30 @@
 use super::common::*;
 use rev_buf_reader::RevBufReader;
 use std::fs::{self};
-use std::io::{self, BufRead, Read, Seek};
+use std::io::{self, BufRead, Read, Seek, SeekFrom};
 
-pub struct LogReader<'a> {
+/// There are three special characters that need to be handled:
+/// Here: SC = escape char, FS = field separator.
+/// - FS FS SC  -> actual record separator
+/// - SC FS SC  -> literal FS
+/// - SC SC SC  -> literal SC
+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,
+    }
+}
+
+pub struct ReverseLogReader<'a> {
     rev_reader: RevBufReader<&'a mut fs::File>,
 }
 
-impl<'a> LogReader<'a> {
-    pub fn new(file: &mut fs::File) -> Result<LogReader, io::Error> {
+impl<'a> ReverseLogReader<'a> {
+    pub fn new(file: &mut fs::File) -> Result<ReverseLogReader, io::Error> {
         let rev_reader = RevBufReader::new(file);
-        Ok(LogReader { rev_reader })
+        Ok(ReverseLogReader { rev_reader })
     }
 
     fn read_record(&mut self) -> Result<Option<Record>, io::Error> {
@@ -48,7 +62,7 @@ impl<'a> LogReader<'a> {
                 SpecialSequence::RecordSeparator => {
                     // The record is complete, so we can break out of the loop.
                     // Move the cursor back to the beginning of the special sequence.
-                    self.rev_reader.seek_relative(3)?;
+                    self.rev_reader.seek_relative(SEQ_RECORD_SEP.len() as i64)?;
                     break;
                 }
                 SpecialSequence::LiteralFieldSeparator => {
@@ -68,7 +82,7 @@ impl<'a> LogReader<'a> {
     }
 
     fn read_special_sequence(&mut self) -> Result<SpecialSequence, io::Error> {
-        let mut special_buf: Vec<u8> = vec![0; 3];
+        let mut special_buf: Vec<u8> = vec![0; SEQ_RECORD_SEP.len()];
         self.rev_reader.read_exact(&mut special_buf)?;
 
         match validate_special(&special_buf.as_slice()) {
@@ -81,7 +95,7 @@ impl<'a> LogReader<'a> {
     }
 }
 
-impl Iterator for LogReader<'_> {
+impl Iterator for ReverseLogReader<'_> {
     type Item = Record;
 
     fn next(&mut self) -> Option<Self::Item> {
@@ -93,16 +107,87 @@ impl Iterator for LogReader<'_> {
     }
 }
 
-/// There are three special characters that need to be handled:
-/// Here: SC = escape char, FS = field separator.
-/// - FS FS SC  -> actual record separator
-/// - SC FS SC  -> literal FS
-/// - SC SC SC  -> literal SC
-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,
+pub struct ForwardLogReader<'a> {
+    reader: io::BufReader<&'a mut fs::File>,
+}
+
+impl<'a> ForwardLogReader<'a> {
+    pub fn new(file: &mut fs::File) -> ForwardLogReader {
+        let reader = io::BufReader::new(file);
+        ForwardLogReader { reader }
+    }
+
+    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, 1];
+        match self.reader.read_exact(&mut peek_buf) {
+            Ok(_) => {
+                // Go back one byte (not sure why you need to seek by -2 here?)
+                self.reader.seek_relative(-2)?;
+            }
+            Err(ref e) if e.kind() == io::ErrorKind::UnexpectedEof => {
+                return Ok(None);
+            }
+            Err(e) => {
+                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]);
+
+            // 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);
+                }
+            }
+        }
+
+        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<'_> {
+    type Item = Record;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.read_record() {
+            Ok(Some(record)) => Some(record),
+            Ok(None) => None,
+            Err(err) => panic!("Error reading record: {:?}", err),
+        }
     }
 }