Split log_reader into reverse_log_reader and forward_log_reader

1 files changed, 0 insertions, 465 deletions

diff --git a/src/log_reader.rs b/src/log_reader.rs
deleted file mode 100644
index d07326a..0000000
--- a/src/log_reader.rs
+++ /dev/null
@@ -1,465 +0,0 @@
-use super::common::*;
-use std::fs::{self};
-use std::io::{self, BufRead, Read, Seek, SeekFrom};
-use std::path::Path;
-
-/// 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
-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> {
-    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,
-    consumed_record_sep: bool,
-}
-
-const INTERNAL_BUF_SIZE: usize = 4096;
-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; 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![];
-        loop {
-            let mut read_buf = vec![];
-            let read = self.read_until(ESCAPE_CHARACTER, &mut read_buf)?;
-
-            result_buf.extend(&read_buf);
-
-            if self.file.stream_position()? == 0 && self.internal_pos == 0 {
-                // We read until the start of the file, we are done
-                break;
-            }
-
-            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;
-                }
-            }
-        }
-
-        result_buf.reverse();
-        Ok(Some(Record::deserialize(&result_buf)))
-    }
-
-    /// Read exactly `buf.len()` bytes from the file, return an error if the file is exhausted.
-    /// The bytes are returned in start -> end order.
-    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",
-                    ));
-                }
-            }
-            let bytes_to_read = std::cmp::min(buf.len() - read, self.internal_pos);
-            buf[read..read + bytes_to_read].copy_from_slice(
-                &self.internal_buf[self.internal_pos - bytes_to_read..self.internal_pos],
-            );
-            self.internal_pos -= bytes_to_read;
-            read += bytes_to_read;
-        }
-        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))?;
-
-        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);
-                }
-            }
-
-            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;
-            }
-        }
-    }
-
-    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)?;
-
-        match validate_special(&special_buf.as_slice()) {
-            Some(special) => Ok(special),
-            None => Err(io::Error::new(
-                io::ErrorKind::InvalidData,
-                format!("Not a special sequence: {:?}", special_buf),
-            )),
-        }
-    }
-}
-
-#[cfg(test)]
-mod reverse_reader_tests {
-    use super::*;
-    use std::io::Write;
-
-    #[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()
-            .read(true)
-            .open(&db_path)
-            .expect("Failed to open file");
-        let mut reverse_log_reader = ReverseLogReader::new(&mut file).unwrap();
-
-        // There are two records in the log with "schema": Int, Null
-
-        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,
-            _ => false,
-        });
-
-        assert!(reverse_log_reader.next().is_none());
-    }
-
-    #[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()
-            .read(true)
-            .open(&db_path)
-            .expect("Failed to open file");
-        let mut forward_log_reader = ForwardLogReader::new(&mut file);
-
-        // There are two records in the log with "schema": Int, Null
-
-        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,
-            _ => false,
-        });
-
-        assert!(forward_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);
-        assert_eq!(String::from_utf8(buf[..5].to_vec()).unwrap(), "hello");
-    }
-}
-
-impl Iterator for ReverseLogReader<'_> {
-    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),
-        }
-    }
-}
-
-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];
-        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 => {
-                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),
-        }
-    }
-}