From 6532e4f4a94ada35c958cf53e1f98aaf175188c6 Mon Sep 17 00:00:00 2001 From: Jan Tuomi Date: Sat, 5 Oct 2024 10:00:45 +0200 Subject: Split log_reader into reverse_log_reader and forward_log_reader --- src/common.rs | 16 ++ src/forward_log_reader.rs | 124 +++++++++++++ src/lib.rs | 7 +- src/log_reader.rs | 465 ---------------------------------------------- src/reverse_log_reader.rs | 336 +++++++++++++++++++++++++++++++++ 5 files changed, 480 insertions(+), 468 deletions(-) create mode 100644 src/forward_log_reader.rs delete mode 100644 src/log_reader.rs create mode 100644 src/reverse_log_reader.rs diff --git a/src/common.rs b/src/common.rs index 9ea43aa..ea179ab 100644 --- a/src/common.rs +++ b/src/common.rs @@ -39,6 +39,22 @@ pub const SEQ_LIT_FIELD_SEP: &[u8] = &[ 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 { + 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, diff --git a/src/forward_log_reader.rs b/src/forward_log_reader.rs new file mode 100644 index 0000000..10788a7 --- /dev/null +++ b/src/forward_log_reader.rs @@ -0,0 +1,124 @@ +use super::common::*; +use std::fs::{self}; +use std::io::{self, BufRead, Read}; + +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, io::Error> { + // The buffer that stores the bytes read from the file. + let mut read_buf: Vec = Vec::new(); + // The buffer that stores all the bytes of the record read so far in reverse order. + let mut result_buf: Vec = 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 { + let mut special_buf: Vec = 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 { + match self.read_record() { + Ok(Some(record)) => Some(record), + Ok(None) => None, + Err(err) => panic!("Error reading record: {:?}", err), + } + } +} + +#[cfg(test)] +mod reverse_reader_tests { + use super::*; + use std::path::Path; + + #[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()); + } +} diff --git a/src/lib.rs b/src/lib.rs index 7146948..0ba2cad 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -2,16 +2,17 @@ extern crate log; mod common; -mod log_reader; +mod forward_log_reader; mod primary_memtable; +mod reverse_log_reader; mod secondary_memtable; pub use common::*; +pub use forward_log_reader::ForwardLogReader; use fs2::lock_contended_error; use fs2::FileExt; -pub use log_reader::ForwardLogReader; -pub use log_reader::ReverseLogReader; use primary_memtable::PrimaryMemtable; +pub use reverse_log_reader::ReverseLogReader; use secondary_memtable::SecondaryMemtable; use std::fmt::Debug; use std::fs::{self}; 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 { - 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, - /// 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 { - 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 { - 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, 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 = 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 { - 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 { - 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) -> Result { - // 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 { - let mut special_buf: Vec = 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 { - 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, io::Error> { - // The buffer that stores the bytes read from the file. - let mut read_buf: Vec = Vec::new(); - // The buffer that stores all the bytes of the record read so far in reverse order. - let mut result_buf: Vec = 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 { - let mut special_buf: Vec = 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 { - match self.read_record() { - Ok(Some(record)) => Some(record), - Ok(None) => None, - Err(err) => panic!("Error reading record: {:?}", err), - } - } -} 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, + /// 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 { + 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 { + 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, 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 = 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 { + 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 { + 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) -> Result { + // 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 { + let mut special_buf: Vec = 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 { + match self.read_record() { + Ok(Some(record)) => Some(record), + Ok(None) => None, + Err(err) => panic!("Error reading record: {:?}", err), + } + } +} -- cgit v1.3