1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
use super::common::*;
use super::record::*;
use std::fs::{self};
use std::io::{self, Read, Seek};
pub struct ReverseLogReader {
/// The contents of the metadata file are read into this buffer in one go.
metadata_buf: Vec<u8>,
/// The data log file that is read based on offset + length information from the metadata file.
data_reader: io::BufReader<fs::File>,
/// The current position in metadata_buf.
metadata_pos: usize,
}
impl ReverseLogReader {
pub fn new(
mut metadata_file: fs::File,
data_file: fs::File,
) -> Result<ReverseLogReader, io::Error> {
let mut metadata_buf = vec![];
metadata_file.seek(io::SeekFrom::Start(0))?;
metadata_file.read_to_end(&mut metadata_buf)?;
let len = metadata_buf.len();
let ret = ReverseLogReader {
metadata_buf,
data_reader: io::BufReader::new(data_file),
metadata_pos: len,
};
Ok(ret)
}
fn read_record(&mut self) -> Result<Option<Record>, io::Error> {
loop {
assert!(
(self.metadata_pos - METADATA_FILE_HEADER_SIZE) % 16 == 0,
"metadata_pos is not aligned"
);
// Return None if we have read the entire metadata file and
// reached the end of the header
if self.metadata_pos == METADATA_FILE_HEADER_SIZE {
return Ok(None);
}
self.metadata_pos -= 16;
let i = self.metadata_pos; // shorter alias
// 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(self.metadata_buf[i..i + 8].try_into().unwrap());
let entry_length =
u64::from_be_bytes(self.metadata_buf[i + 8..i + 16].try_into().unwrap());
if entry_offset == 0 && entry_length == 0 {
// This is an unused entry in the metadata file, skip
continue;
}
self.data_reader.seek(io::SeekFrom::Start(entry_offset))?;
let mut result_buf = vec![0; entry_length as usize];
self.data_reader.read_exact(&mut result_buf)?;
let record = Record::deserialize(&result_buf);
assert!(
(self.metadata_pos - METADATA_FILE_HEADER_SIZE) % 16 == 0,
"metadata_pos is not aligned"
);
return Ok(Some(record));
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::path::Path;
#[test]
fn test_reverse_log_reader_fixture_db1() {
let _ = env_logger::builder().is_test(true).try_init();
let metadata_path = Path::new(TEST_RESOURCES_DIR).join("test_metadata_1");
let metadata_file = fs::OpenOptions::new()
.read(true)
.open(&metadata_path)
.expect("Failed to open file");
let data_file = fs::OpenOptions::new()
.read(true)
.open(Path::new(TEST_RESOURCES_DIR).join("test_data_1"))
.expect("Failed to open file");
let mut reverse_log_reader = ReverseLogReader::new(metadata_file, data_file).unwrap();
// There are two records in the log with "schema": Int
let last_record = reverse_log_reader
.next()
.expect("Failed to read the last record");
assert!(match &last_record.values[..] {
[Value::Bytes(bytes)] => bytes.len() == 256,
_ => false,
});
assert!(reverse_log_reader.next().is_none());
}
}
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)
}
}
}
}
|
