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
|
use super::common::*;
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> {
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());
debug!(
"Read offset {} and length {} from metadata file at position {}",
entry_offset, entry_length, i
);
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"
);
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.as_slice() {
[RecordValue::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)
}
}
}
}
|
