aboutsummaryrefslogtreecommitdiffstats
path: root/log_db/src/log_reader_forward.rs
diff options
context:
space:
mode:
Diffstat (limited to 'log_db/src/log_reader_forward.rs')
-rw-r--r--log_db/src/log_reader_forward.rs134
1 files changed, 0 insertions, 134 deletions
diff --git a/log_db/src/log_reader_forward.rs b/log_db/src/log_reader_forward.rs
deleted file mode 100644
index f3fc16c..0000000
--- a/log_db/src/log_reader_forward.rs
+++ /dev/null
@@ -1,134 +0,0 @@
-use super::*;
-
-pub struct ForwardLogReader {
- metadata_reader: io::BufReader<fs::File>,
- data_reader: io::BufReader<fs::File>,
-}
-
-pub struct ForwardLogReaderItem {
- pub row: Row,
- pub index: u64,
-}
-
-impl ForwardLogReader {
- pub fn new(metadata_file: fs::File, data_file: fs::File) -> ForwardLogReader {
- let mut ret = ForwardLogReader {
- metadata_reader: io::BufReader::new(metadata_file),
- data_reader: io::BufReader::new(data_file),
- };
-
- ret.metadata_reader
- .seek(io::SeekFrom::Start(METADATA_FILE_HEADER_SIZE as u64))
- .expect("Seek failed");
-
- ret
- }
-
- pub fn new_with_index(
- metadata_file: fs::File,
- data_file: fs::File,
- index: u64,
- ) -> ForwardLogReader {
- let mut ret = ForwardLogReader {
- metadata_reader: io::BufReader::new(metadata_file),
- data_reader: io::BufReader::new(data_file),
- };
-
- ret.metadata_reader
- .seek(io::SeekFrom::Start(
- METADATA_FILE_HEADER_SIZE as u64 + METADATA_ROW_LENGTH as u64 * index,
- ))
- .expect("Seek failed");
-
- ret
- }
-
- fn read_record(&mut self) -> Result<Option<ForwardLogReaderItem>, io::Error> {
- loop {
- let pos = self.metadata_reader.stream_position()?;
- let index = (pos - METADATA_FILE_HEADER_SIZE as u64) / METADATA_ROW_LENGTH as u64;
-
- let mut metadata_entry_buf = vec![0; 16]; // 2x u64
- if let Err(e) = self.metadata_reader.read_exact(&mut metadata_entry_buf) {
- if e.kind() == io::ErrorKind::UnexpectedEof {
- return Ok(None);
- } else {
- return Err(e);
- }
- }
-
- // 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());
-
- if entry_offset == 0 && entry_length == 0 {
- // This is an unused entry in the metadata file, skip
- continue;
- }
-
- // Use .seek_relative instead of .seek to avoid dropping the BufReader internal buffer when
- // the seek distance is small
- let seek_distance = entry_offset as i64 - self.data_reader.stream_position()? as i64;
- self.data_reader.seek_relative(seek_distance)?;
-
- let mut result_buf = vec![0; entry_length as usize];
- self.data_reader.read_exact(&mut result_buf)?;
-
- let row = Row::deserialize(&result_buf);
- return Ok(Some(ForwardLogReaderItem { row, index }));
- }
- }
-}
-
-impl Iterator for ForwardLogReader {
- type Item = ForwardLogReaderItem;
-
- fn next(&mut self) -> Option<Self::Item> {
- self.read_record().unwrap_or_else(|err| {
- panic!("Error reading record: {:?}", err);
- })
- }
-}
-
-#[cfg(test)]
-mod tests {
- use ctor::ctor;
- use env_logger;
-
- use super::*;
-
- #[ctor]
- fn init_logger() {
- let _ = env_logger::builder().is_test(true).try_init();
- }
-
- const TEST_RESOURCES_DIR: &str = "tests/resources";
-
- #[test]
- fn test_forward_log_reader_fixture_db1() {
- 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()
- .read(true)
- .open(&metadata_path)
- .expect("Failed to open metadata file");
- let data_file = fs::OpenOptions::new()
- .read(true)
- .open(&data_path)
- .expect("Failed to open data file");
-
- let mut forward_log_reader = ForwardLogReader::new(metadata_file, data_file);
-
- // There are two records in the log with "schema" with one field: Bytes
-
- let ForwardLogReaderItem { row, index: _ } = forward_log_reader
- .next()
- .expect("Failed to read the first record");
- assert!(match &row.values[..] {
- [Value::Bytes(bytes)] => bytes.len() == 256,
- _ => false,
- });
-
- assert!(forward_log_reader.next().is_none());
- }
-}