use block::{Block, BlockIter};
use blockhandle::BlockHandle;
use cache;
use error::Result;
use filter_block::FilterBlockReader;
use options::Options;
use table_block;
use table_builder::{self, Footer};
use types::{current_key_val, RandomAccess, SSIterator};
use std::cmp::Ordering;
use std::fs;
use std::path;
use std::rc::Rc;
use integer_encoding::FixedIntWriter;
fn read_footer(f: &dyn RandomAccess, size: usize) -> Result<Footer> {
let mut buf = vec![0; table_builder::FULL_FOOTER_LENGTH];
f.read_at(size - table_builder::FULL_FOOTER_LENGTH, &mut buf)?;
Ok(Footer::decode(&buf))
}
#[derive(Clone)]
pub struct Table {
file: Rc<Box<dyn RandomAccess>>,
file_size: usize,
cache_id: cache::CacheID,
opt: Options,
footer: Footer,
indexblock: Block,
filters: Option<FilterBlockReader>,
}
impl Table {
pub fn new_from_file(opt: Options, path: &path::Path) -> Result<Table> {
let f = fs::OpenOptions::new().read(true).open(path)?;
let size = f.metadata()?.len() as usize;
Table::new(opt, Box::new(f), size)
}
pub fn new(opt: Options, file: Box<dyn RandomAccess>, size: usize) -> Result<Table> {
let footer = read_footer(file.as_ref(), size)?;
let indexblock = table_block::read_table_block(opt.clone(), file.as_ref(), &footer.index)?;
let metaindexblock =
table_block::read_table_block(opt.clone(), file.as_ref(), &footer.meta_index)?;
let filter_block_reader = Table::read_filter_block(&metaindexblock, file.as_ref(), &opt)?;
let cache_id = opt.block_cache.borrow_mut().new_cache_id();
Ok(Table {
file: Rc::new(file),
file_size: size,
cache_id: cache_id,
opt: opt,
footer: footer,
filters: filter_block_reader,
indexblock: indexblock,
})
}
fn read_filter_block(
metaix: &Block,
file: &dyn RandomAccess,
options: &Options,
) -> Result<Option<FilterBlockReader>> {
let filter_name = format!("filter.{}", options.filter_policy.name())
.as_bytes()
.to_vec();
let mut metaindexiter = metaix.iter();
metaindexiter.seek(&filter_name);
if let Some((_key, val)) = current_key_val(&metaindexiter) {
let filter_block_location = BlockHandle::decode(&val).0;
if filter_block_location.size() > 0 {
return Ok(Some(table_block::read_filter_block(
file,
&filter_block_location,
options.filter_policy.clone(),
)?));
}
}
Ok(None)
}
fn block_cache_handle(&self, block_off: usize) -> cache::CacheKey {
let mut dst = [0; 2 * 8];
(&mut dst[..8])
.write_fixedint(self.cache_id)
.expect("error writing to vec");
(&mut dst[8..])
.write_fixedint(block_off as u64)
.expect("error writing to vec");
dst
}
fn read_block(&self, location: &BlockHandle) -> Result<Block> {
let cachekey = self.block_cache_handle(location.offset());
if let Some(block) = self.opt.block_cache.borrow_mut().get(&cachekey) {
return Ok(block.clone());
}
let b =
table_block::read_table_block(self.opt.clone(), self.file.as_ref().as_ref(), location)?;
self.opt
.block_cache
.borrow_mut()
.insert(&cachekey, b.clone());
Ok(b)
}
pub fn approx_offset_of(&self, key: &[u8]) -> usize {
let mut iter = self.indexblock.iter();
iter.seek(key);
if let Some((_, val)) = current_key_val(&iter) {
let location = BlockHandle::decode(&val).0;
return location.offset();
}
return self.footer.meta_index.offset();
}
pub fn iter(&self) -> TableIterator {
let iter = TableIterator {
current_block: None,
current_block_off: 0,
index_block: self.indexblock.iter(),
table: self.clone(),
};
iter
}
pub fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
let mut index_iter = self.indexblock.iter();
index_iter.seek(key);
let handle;
if let Some((last_in_block, h)) = current_key_val(&index_iter) {
if self.opt.cmp.cmp(key, &last_in_block) == Ordering::Less {
handle = BlockHandle::decode(&h).0;
} else {
return Ok(None);
}
} else {
return Ok(None);
}
if let Some(ref filters) = self.filters {
if !filters.key_may_match(handle.offset(), key) {
return Ok(None);
}
}
let tb = self.read_block(&handle)?;
let mut iter = tb.iter();
iter.seek(key);
if let Some((k, v)) = current_key_val(&iter) {
if self.opt.cmp.cmp(&k, key) == Ordering::Equal {
return Ok(Some(v));
}
}
Ok(None)
}
}
pub struct TableIterator {
table: Table,
current_block: Option<BlockIter>,
current_block_off: usize,
index_block: BlockIter,
}
impl TableIterator {
fn skip_to_next_entry(&mut self) -> Result<bool> {
if let Some((_key, val)) = self.index_block.next() {
self.load_block(&val).map(|_| true)
} else {
Ok(false)
}
}
fn load_block(&mut self, handle: &[u8]) -> Result<()> {
let (new_block_handle, _) = BlockHandle::decode(handle);
let block = self.table.read_block(&new_block_handle)?;
self.current_block = Some(block.iter());
self.current_block_off = new_block_handle.offset();
Ok(())
}
}
impl SSIterator for TableIterator {
fn advance(&mut self) -> bool {
if self.current_block.is_none() {
match self.skip_to_next_entry() {
Ok(true) => return self.advance(),
Ok(false) => {
self.reset();
return false;
}
Err(_) => return self.advance(),
}
}
if let Some(ref mut cb) = self.current_block {
if cb.advance() {
return true;
}
}
self.current_block = None;
match self.skip_to_next_entry() {
Ok(true) => self.advance(),
Ok(false) => {
self.reset();
false
}
Err(_) => self.advance(),
}
}
fn seek(&mut self, to: &[u8]) {
self.index_block.seek(to);
if let Some((past_block, handle)) = current_key_val(&self.index_block) {
if self.table.opt.cmp.cmp(to, &past_block) <= Ordering::Equal {
if let Ok(()) = self.load_block(&handle) {
self.current_block.as_mut().unwrap().seek(to);
return;
}
}
}
self.reset();
}
fn prev(&mut self) -> bool {
if let Some(ref mut cb) = self.current_block {
if cb.prev() {
return true;
}
}
if self.index_block.prev() {
if let Some((_, handle)) = current_key_val(&self.index_block) {
if self.load_block(&handle).is_ok() {
self.current_block.as_mut().unwrap().seek_to_last();
self.current_block.as_ref().unwrap().valid()
} else {
self.reset();
false
}
} else {
false
}
} else {
false
}
}
fn reset(&mut self) {
self.index_block.reset();
self.current_block = None;
}
fn valid(&self) -> bool {
self.current_block.is_some() && (self.current_block.as_ref().unwrap().valid())
}
fn current(&self, key: &mut Vec<u8>, val: &mut Vec<u8>) -> bool {
if let Some(ref cb) = self.current_block {
cb.current(key, val)
} else {
false
}
}
}
#[cfg(test)]
mod tests {
use options::CompressionType;
use table_builder::TableBuilder;
use test_util::{test_iterator_properties, SSIteratorIter};
use types::{current_key_val, SSIterator};
use super::*;
fn build_data() -> Vec<(&'static str, &'static str)> {
vec![
("abc", "def"),
("abd", "dee"),
("bcd", "asa"),
("bsr", "a00"),
("xyz", "xxx"),
("xzz", "yyy"),
("zzz", "111"),
]
}
fn build_table(data: Vec<(&'static str, &'static str)>) -> (Vec<u8>, usize) {
let mut d = Vec::with_capacity(512);
let mut opt = Options::default();
opt.block_restart_interval = 2;
opt.block_size = 32;
opt.compression_type = CompressionType::CompressionSnappy;
{
let mut b = TableBuilder::new(opt, &mut d);
for &(k, v) in data.iter() {
b.add(k.as_bytes(), v.as_bytes()).unwrap();
}
b.finish().unwrap();
}
let size = d.len();
(d, size)
}
fn wrap_buffer(src: Vec<u8>) -> Box<dyn RandomAccess> {
Box::new(src)
}
#[test]
fn test_table_approximate_offset() {
let (src, size) = build_table(build_data());
let mut opt = Options::default();
opt.block_size = 32;
let table = Table::new(opt, wrap_buffer(src), size).unwrap();
let mut iter = table.iter();
let expected_offsets = vec![0, 0, 0, 44, 44, 44, 89];
let mut i = 0;
for (k, _) in SSIteratorIter::wrap(&mut iter) {
assert_eq!(expected_offsets[i], table.approx_offset_of(&k));
i += 1;
}
assert_eq!(137, table.approx_offset_of("{aa".as_bytes()));
}
#[test]
fn test_table_block_cache_use() {
let (src, size) = build_table(build_data());
let mut opt = Options::default();
opt.block_size = 32;
let table = Table::new(opt.clone(), wrap_buffer(src), size).unwrap();
let mut iter = table.iter();
assert_eq!(opt.block_cache.borrow().count(), 0);
iter.next();
assert_eq!(opt.block_cache.borrow().count(), 1);
iter.next();
iter.next();
iter.next();
iter.next();
assert_eq!(opt.block_cache.borrow().count(), 2);
}
#[test]
fn test_table_iterator_fwd_bwd() {
let (src, size) = build_table(build_data());
let data = build_data();
let table = Table::new(Options::default(), wrap_buffer(src), size).unwrap();
let mut iter = table.iter();
let mut i = 0;
while let Some((k, v)) = iter.next() {
assert_eq!(
(data[i].0.as_bytes(), data[i].1.as_bytes()),
(k.as_ref(), v.as_ref())
);
i += 1;
}
assert_eq!(i, data.len());
assert!(!iter.valid());
while let Some((key, _)) = iter.next() {
if key.as_slice() == b"zzz" {
break;
}
}
assert!(iter.valid());
let mut j = 0;
while iter.prev() {
if let Some((k, v)) = current_key_val(&iter) {
j += 1;
assert_eq!(
(
data[data.len() - 1 - j].0.as_bytes(),
data[data.len() - 1 - j].1.as_bytes()
),
(k.as_ref(), v.as_ref())
);
} else {
break;
}
}
assert_eq!(j, 6);
}
#[test]
fn test_table_iterator_filter() {
let (src, size) = build_table(build_data());
let table = Table::new(Options::default(), wrap_buffer(src), size).unwrap();
assert!(table.filters.is_some());
let filter_reader = table.filters.clone().unwrap();
let mut iter = table.iter();
loop {
if let Some((k, _)) = iter.next() {
assert!(filter_reader.key_may_match(iter.current_block_off, &k));
assert!(!filter_reader.key_may_match(iter.current_block_off, b"somerandomkey"));
} else {
break;
}
}
}
#[test]
fn test_table_iterator_state_behavior() {
let (src, size) = build_table(build_data());
let table = Table::new(Options::default(), wrap_buffer(src), size).unwrap();
let mut iter = table.iter();
assert!(!iter.valid());
assert!(current_key_val(&iter).is_none());
assert!(!iter.prev());
assert!(iter.advance());
let first = current_key_val(&iter);
assert!(iter.valid());
assert!(current_key_val(&iter).is_some());
assert!(iter.advance());
assert!(iter.prev());
assert!(iter.valid());
iter.reset();
assert!(!iter.valid());
assert!(current_key_val(&iter).is_none());
assert_eq!(first, iter.next());
}
#[test]
fn test_table_iterator_behavior_standard() {
let mut data = build_data();
data.truncate(4);
let (src, size) = build_table(data);
let table = Table::new(Options::default(), wrap_buffer(src), size).unwrap();
test_iterator_properties(table.iter());
}
#[test]
fn test_table_iterator_values() {
let (src, size) = build_table(build_data());
let data = build_data();
let table = Table::new(Options::default(), wrap_buffer(src), size).unwrap();
let mut iter = table.iter();
let mut i = 0;
iter.next();
iter.next();
loop {
iter.prev();
if let Some((k, v)) = current_key_val(&iter) {
assert_eq!(
(data[i].0.as_bytes(), data[i].1.as_bytes()),
(k.as_ref(), v.as_ref())
);
} else {
break;
}
i += 1;
if iter.next().is_none() || iter.next().is_none() {
break;
}
}
assert_eq!(i, 6);
}
#[test]
fn test_table_iterator_seek() {
let (src, size) = build_table(build_data());
let table = Table::new(Options::default(), wrap_buffer(src), size).unwrap();
let mut iter = table.iter();
iter.seek(b"bcd");
assert!(iter.valid());
assert_eq!(
current_key_val(&iter),
Some((b"bcd".to_vec(), b"asa".to_vec()))
);
iter.seek(b"abc");
assert!(iter.valid());
assert_eq!(
current_key_val(&iter),
Some((b"abc".to_vec(), b"def".to_vec()))
);
iter.seek("{{{".as_bytes());
assert!(!iter.valid());
iter.seek(b"bbb");
assert!(iter.valid());
}
#[test]
fn test_table_get() {
let (src, size) = build_table(build_data());
let table = Table::new(Options::default(), wrap_buffer(src), size).unwrap();
let table2 = table.clone();
let mut _iter = table.iter();
for (k, v) in SSIteratorIter::wrap(&mut _iter) {
let r = table2.get(&k);
assert_eq!(Ok(Some(v)), r);
}
assert_eq!(table.opt.block_cache.borrow().count(), 3);
assert!(table.get(b"aaa").unwrap().is_none());
assert!(table.get(b"aaaa").unwrap().is_none());
assert!(table.get(b"aa").unwrap().is_none());
assert!(table.get(b"abcd").unwrap().is_none());
assert!(table.get(b"abb").unwrap().is_none());
assert!(table.get(b"xyy").unwrap().is_none());
assert!(table.get(b"zzy").unwrap().is_none());
assert!(table.get(b"zz1").unwrap().is_none());
assert!(table.get("zz{".as_bytes()).unwrap().is_none());
}
#[test]
fn test_table_reader_checksum() {
let (mut src, size) = build_table(build_data());
src[10] += 1;
let table = Table::new(Options::default(), wrap_buffer(src), size).unwrap();
assert!(table.filters.is_some());
assert_eq!(table.filters.as_ref().unwrap().num(), 1);
{
let mut _iter = table.iter();
let iter = SSIteratorIter::wrap(&mut _iter);
assert_eq!(iter.count(), 4);
}
{
let mut _iter = table.iter();
let iter = SSIteratorIter::wrap(&mut _iter);
for (k, _) in iter {
if k == build_data()[5].0.as_bytes() {
return;
}
}
panic!("Should have hit 5th record in table!");
}
}
}