Split log_reader into reverse_log_reader and forward_log_reader

1 files changed, 336 insertions, 0 deletions

diff --git a/src/reverse_log_reader.rs b/src/reverse_log_reader.rs
new file mode 100644
index 0000000..778d1b4
--- /dev/null
+++ b/src/reverse_log_reader.rs
@@ -0,0 +1,336 @@
+use super::common::*;
+use std::fs::{self};
+use std::io::{self, Read, Seek, SeekFrom};
+
+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;
+    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()
+            .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_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),
+        }
+    }
+}