use crate::cmp::{Cmp, InternalKeyCmp};
use crate::env::Env;
use crate::error::{err, Result, StatusCode};
use crate::key_types::{parse_internal_key, InternalKey, UserKey};
use crate::log::{LogReader, LogWriter};
use crate::merging_iter::MergingIter;
use crate::options::Options;
use crate::table_cache::TableCache;
use crate::types::{
parse_file_name, share, FileMetaData, FileNum, FileType, LdbIterator, Shared, NUM_LEVELS,
};
use crate::version::{new_version_iter, total_size, FileMetaHandle, Version};
use crate::version_edit::VersionEdit;
use bytes::Bytes;
use std::cmp::Ordering;
use std::collections::HashSet;
use std::io::Write;
use std::path::{Path, PathBuf};
use std::rc::Rc;
pub struct Compaction {
level: usize,
max_file_size: usize,
input_version: Option<Shared<Version>>,
level_ixs: [usize; NUM_LEVELS],
cmp: Rc<Box<dyn Cmp>>,
icmp: InternalKeyCmp,
manual: bool,
inputs: [Vec<FileMetaHandle>; 2],
grandparent_ix: usize,
grandparents: Option<Vec<FileMetaHandle>>,
overlapped_bytes: usize,
seen_key: bool,
edit: VersionEdit,
}
impl Compaction {
pub fn new(opt: &Options, level: usize, input: Option<Shared<Version>>) -> Compaction {
Compaction {
level,
max_file_size: opt.max_file_size,
input_version: input,
level_ixs: Default::default(),
cmp: opt.cmp.clone(),
icmp: InternalKeyCmp(opt.cmp.clone()),
manual: false,
inputs: Default::default(),
grandparent_ix: 0,
grandparents: Default::default(),
overlapped_bytes: 0,
seen_key: false,
edit: VersionEdit::new(),
}
}
fn add_input(&mut self, parent: usize, f: FileMetaHandle) {
assert!(parent <= 1);
self.inputs[parent].push(f)
}
pub fn level(&self) -> usize {
self.level
}
pub fn input(&self, parent: usize, ix: usize) -> FileMetaData {
assert!(parent < 2);
assert!(ix < self.inputs[parent].len());
self.inputs[parent][ix].borrow().clone()
}
pub fn num_inputs(&self, parent: usize) -> usize {
assert!(parent < 2);
self.inputs[parent].len()
}
pub fn edit(&mut self) -> &mut VersionEdit {
&mut self.edit
}
pub fn into_edit(self) -> VersionEdit {
self.edit
}
pub fn add_input_deletions(&mut self) {
for parent in 0..2 {
for f in &self.inputs[parent] {
self.edit.delete_file(self.level + parent, f.borrow().num);
}
}
}
pub fn is_base_level_for(&mut self, k: UserKey<'_>) -> bool {
assert!(self.input_version.is_some());
let inp_version = self.input_version.as_ref().unwrap();
for level in self.level + 2..NUM_LEVELS {
let files = &inp_version.borrow().files[level];
while self.level_ixs[level] < files.len() {
let f = files[self.level_ixs[level]].borrow();
if self.cmp.cmp(k, parse_internal_key(&f.largest).2) <= Ordering::Equal {
if self.cmp.cmp(k, parse_internal_key(&f.smallest).2) >= Ordering::Equal {
return false;
}
break;
}
self.level_ixs[level] += 1;
}
}
true
}
pub fn is_trivial_move(&self) -> bool {
if self.manual {
return false;
}
let inputs_size;
if let Some(gp) = self.grandparents.as_ref() {
inputs_size = total_size(gp.iter());
} else {
inputs_size = 0;
}
self.num_inputs(0) == 1 && self.num_inputs(1) == 0 && inputs_size < 10 * self.max_file_size
}
pub fn should_stop_before(&mut self, k: InternalKey<'_>) -> bool {
if self.grandparents.is_none() {
self.seen_key = true;
return false;
}
let grandparents = self.grandparents.as_ref().unwrap();
while self.grandparent_ix < grandparents.len()
&& self
.icmp
.cmp(k, &grandparents[self.grandparent_ix].borrow().largest)
== Ordering::Greater
{
if self.seen_key {
self.overlapped_bytes += grandparents[self.grandparent_ix].borrow().size;
}
self.grandparent_ix += 1;
}
self.seen_key = true;
if self.overlapped_bytes > 10 * self.max_file_size {
self.overlapped_bytes = 0;
true
} else {
false
}
}
}
pub struct VersionSet {
dbname: PathBuf,
opt: Options,
cmp: InternalKeyCmp,
cache: Shared<TableCache>,
pub next_file_num: u64,
pub manifest_num: u64,
pub last_seq: u64,
pub log_num: u64,
pub prev_log_num: u64,
current: Option<Shared<Version>>,
compaction_ptrs: [Bytes; NUM_LEVELS],
descriptor_log: Option<LogWriter<Box<dyn Write>>>,
}
impl VersionSet {
pub fn new<P: AsRef<Path>>(db: P, opt: Options, cache: Shared<TableCache>) -> VersionSet {
let v = share(Version::new(cache.clone(), opt.cmp.clone()));
VersionSet {
dbname: db.as_ref().to_owned(),
cmp: InternalKeyCmp(opt.cmp.clone()),
opt,
cache,
next_file_num: 2,
manifest_num: 0,
last_seq: 0,
log_num: 0,
prev_log_num: 0,
current: Some(v),
compaction_ptrs: Default::default(),
descriptor_log: None,
}
}
pub fn current_summary(&self) -> String {
self.current.as_ref().unwrap().borrow().level_summary()
}
pub fn live_files(&self) -> HashSet<FileNum> {
let mut files = HashSet::new();
if let Some(ref version) = self.current {
for level in 0..NUM_LEVELS {
for file in &version.borrow().files[level] {
files.insert(file.borrow().num);
}
}
}
files
}
pub fn current(&self) -> Shared<Version> {
assert!(self.current.is_some());
self.current.as_ref().unwrap().clone()
}
pub fn add_version(&mut self, v: Version) {
self.current = Some(share(v));
}
pub fn new_file_number(&mut self) -> FileNum {
self.next_file_num += 1;
self.next_file_num - 1
}
pub fn reuse_file_number(&mut self, n: FileNum) {
if n == self.next_file_num - 1 {
self.next_file_num = n;
}
}
pub fn mark_file_number_used(&mut self, n: FileNum) {
if self.next_file_num <= n {
self.next_file_num = n + 1;
}
}
pub fn needs_compaction(&self) -> bool {
assert!(self.current.is_some());
let v = self.current.as_ref().unwrap();
let v = v.borrow();
v.compaction_score.unwrap_or(0.0) >= 1.0 || v.file_to_compact.is_some()
}
fn approximate_offset(&self, v: &Shared<Version>, key: InternalKey<'_>) -> usize {
let mut offset = 0;
for level in 0..NUM_LEVELS {
for f in &v.borrow().files[level] {
if self.opt.cmp.cmp(&f.borrow().largest, key) <= Ordering::Equal {
offset += f.borrow().size;
} else if self.opt.cmp.cmp(&f.borrow().smallest, key) == Ordering::Greater {
if level > 0 {
break;
}
} else if let Ok(tbl) = self.cache.borrow_mut().get_table(f.borrow().num) {
offset += tbl.approx_offset_of(key);
}
}
}
offset
}
pub fn pick_compaction(&mut self) -> Option<Compaction> {
assert!(self.current.is_some());
let current = self.current();
let current = current.borrow();
let mut c = Compaction::new(&self.opt, 0, self.current.clone());
let level;
if current.compaction_score.unwrap_or(0.0) >= 1.0 {
level = current.compaction_level.unwrap();
assert!(level < NUM_LEVELS - 1);
for f in ¤t.files[level] {
if self.compaction_ptrs[level].is_empty()
|| self
.cmp
.cmp(&f.borrow().largest, &self.compaction_ptrs[level])
== Ordering::Greater
{
c.add_input(0, f.clone());
break;
}
}
if c.num_inputs(0) == 0 {
c.add_input(0, current.files[level][0].clone());
}
} else if let Some(ref ftc) = current.file_to_compact {
level = current.file_to_compact_lvl;
c.add_input(0, ftc.clone());
} else {
return None;
}
c.level = level;
c.input_version.clone_from(&self.current);
if level == 0 {
let (smallest, largest) = get_range(&self.cmp, c.inputs[0].iter());
c.inputs[0] = current.overlapping_inputs(0, &smallest, &largest);
assert!(!c.inputs[0].is_empty());
}
self.setup_other_inputs(&mut c);
Some(c)
}
pub fn compact_range(
&mut self,
level: usize,
from: InternalKey<'_>,
to: InternalKey<'_>,
) -> Option<Compaction> {
assert!(self.current.is_some());
let mut inputs = self
.current
.as_ref()
.unwrap()
.borrow()
.overlapping_inputs(level, from, to);
if inputs.is_empty() {
return None;
}
if level > 0 {
let mut total = 0;
for i in 0..inputs.len() {
total += inputs[i].borrow().size;
if total > self.opt.max_file_size {
inputs.truncate(i + 1);
break;
}
}
}
let mut c = Compaction::new(&self.opt, level, self.current.clone());
c.inputs[0] = inputs;
c.manual = true;
self.setup_other_inputs(&mut c);
Some(c)
}
fn setup_other_inputs(&mut self, compaction: &mut Compaction) {
assert!(self.current.is_some());
let current = self.current.as_ref().unwrap();
let current = current.borrow();
let level = compaction.level;
let (mut smallest, mut largest) = get_range(&self.cmp, compaction.inputs[0].iter());
compaction.inputs[1] = current.overlapping_inputs(level + 1, &smallest, &largest);
let (mut allstart, mut alllimit) = get_range(
&self.cmp,
compaction.inputs[0]
.iter()
.chain(compaction.inputs[1].iter()),
);
if !compaction.inputs[1].is_empty() {
let expanded0 = current.overlapping_inputs(level, &allstart, &alllimit);
let inputs1_size = total_size(compaction.inputs[1].iter());
let expanded0_size = total_size(expanded0.iter());
if expanded0.len() > compaction.num_inputs(0)
&& (inputs1_size + expanded0_size) < 25 * self.opt.max_file_size
{
let (new_start, new_limit) = get_range(&self.cmp, expanded0.iter());
let expanded1 = current.overlapping_inputs(level + 1, &new_start, &new_limit);
if expanded1.len() == compaction.num_inputs(1) {
log!(
self.opt.log,
"Expanding inputs@{} {}+{} ({}+{} bytes) to {}+{} ({}+{} bytes)",
level,
compaction.inputs[0].len(),
compaction.inputs[1].len(),
total_size(compaction.inputs[0].iter()),
total_size(compaction.inputs[1].iter()),
expanded0.len(),
expanded1.len(),
total_size(expanded0.iter()),
total_size(expanded1.iter())
);
smallest = new_start;
largest = new_limit;
compaction.inputs[0] = expanded0;
compaction.inputs[1] = expanded1;
let (newallstart, newalllimit) = get_range(
&self.cmp,
compaction.inputs[0]
.iter()
.chain(compaction.inputs[1].iter()),
);
allstart = newallstart;
alllimit = newalllimit;
}
}
}
if level + 2 < NUM_LEVELS {
let grandparents = self.current.as_ref().unwrap().borrow().overlapping_inputs(
level + 2,
&allstart,
&alllimit,
);
compaction.grandparents = Some(grandparents);
}
log!(
self.opt.log,
"Compacting @{} {:?} .. {:?}",
level,
smallest,
largest
);
compaction.edit().set_compact_pointer(level, &largest);
self.compaction_ptrs[level] = largest;
}
fn write_snapshot(&mut self) -> Result<usize> {
assert!(self.descriptor_log.is_some());
let mut edit = VersionEdit::new();
edit.set_comparator_name(self.opt.cmp.id());
for level in 0..NUM_LEVELS {
if !self.compaction_ptrs[level].is_empty() {
edit.set_compact_pointer(level, &self.compaction_ptrs[level]);
}
}
let current = self.current.as_ref().unwrap().borrow();
for level in 0..NUM_LEVELS {
let fs = ¤t.files[level];
for f in fs {
edit.add_file(level, f.borrow().clone());
}
}
self.descriptor_log
.as_mut()
.unwrap()
.add_record(&edit.encode())
}
pub fn log_and_apply(&mut self, mut edit: VersionEdit) -> Result<()> {
assert!(self.current.is_some());
if edit.log_number.is_none() {
edit.set_log_num(self.log_num);
} else {
assert!(edit.log_number.unwrap() >= self.log_num);
assert!(edit.log_number.unwrap() < self.next_file_num);
}
if edit.prev_log_number.is_none() {
edit.set_prev_log_num(self.prev_log_num);
}
edit.set_next_file(self.next_file_num);
edit.set_last_seq(self.last_seq);
let mut v = Version::new(self.cache.clone(), self.opt.cmp.clone());
{
let mut builder = Builder::new();
builder.apply(&edit, &mut self.compaction_ptrs);
builder.save_to(&self.cmp, self.current.as_ref().unwrap(), &mut v);
}
self.finalize(&mut v);
if self.descriptor_log.is_none() {
let descname = manifest_file_name(&self.dbname, self.manifest_num);
edit.set_next_file(self.next_file_num);
self.descriptor_log = Some(LogWriter::new(
self.opt.env.open_writable_file(Path::new(&descname))?,
));
self.write_snapshot()?;
}
let encoded = edit.encode();
if let Some(ref mut lw) = self.descriptor_log {
lw.add_record(&encoded)?;
lw.flush()?;
}
set_current_file(
self.opt.env.as_ref().as_ref(),
&self.dbname,
self.manifest_num,
)?;
self.add_version(v);
self.log_num = edit.log_number.unwrap();
Ok(())
}
fn finalize(&self, v: &mut Version) {
let mut best_lvl = None;
let mut best_score = None;
for l in 0..NUM_LEVELS - 1 {
let score = if l == 0 {
v.files[l].len() as f64 / 4.0
} else {
let mut max_bytes = 10.0 * f64::from(1 << 20);
for _ in 0..l - 1 {
max_bytes *= 10.0;
}
total_size(v.files[l].iter()) as f64 / max_bytes
};
if let Some(ref mut b) = best_score {
if *b < score {
*b = score;
best_lvl = Some(l);
}
} else {
best_score = Some(score);
best_lvl = Some(l);
}
}
v.compaction_score = best_score;
v.compaction_level = best_lvl;
}
pub fn recover(&mut self) -> Result<bool> {
assert!(self.current.is_some());
let mut current = read_current_file(self.opt.env.as_ref().as_ref(), &self.dbname)?;
let len = current.len();
current.truncate(len - 1);
let current = Path::new(¤t);
let descfilename = self.dbname.join(current);
let mut builder = Builder::new();
{
let mut descfile = self
.opt
.env
.open_sequential_file(Path::new(&descfilename))?;
let mut logreader = LogReader::new(
&mut descfile,
true,
);
let mut log_number = None;
let mut prev_log_number = None;
let mut next_file_number = None;
let mut last_seq = None;
let mut buf = Vec::new();
while let Ok(size) = logreader.read(&mut buf) {
if size == 0 {
break;
}
let edit = VersionEdit::decode_from(&buf)?;
builder.apply(&edit, &mut self.compaction_ptrs);
if let Some(ln) = edit.log_number {
log_number = Some(ln);
}
if let Some(nfn) = edit.next_file_number {
next_file_number = Some(nfn);
}
if let Some(ls) = edit.last_seq {
last_seq = Some(ls);
}
if let Some(pln) = edit.prev_log_number {
prev_log_number = Some(pln);
}
}
if let Some(ln) = log_number {
self.log_num = ln;
self.mark_file_number_used(ln);
} else {
return err(
StatusCode::Corruption,
"no meta-lognumber entry in descriptor",
);
}
if let Some(nfn) = next_file_number {
self.next_file_num = nfn + 1;
} else {
return err(
StatusCode::Corruption,
"no meta-next-file entry in descriptor",
);
}
if let Some(ls) = last_seq {
self.last_seq = ls;
} else {
return err(
StatusCode::Corruption,
"no last-sequence entry in descriptor",
);
}
if let Some(pln) = prev_log_number {
self.prev_log_num = pln;
self.mark_file_number_used(prev_log_number.unwrap());
} else {
self.prev_log_num = 0;
}
}
let mut v = Version::new(self.cache.clone(), self.opt.cmp.clone());
builder.save_to(&self.cmp, self.current.as_ref().unwrap(), &mut v);
self.finalize(&mut v);
self.add_version(v);
self.manifest_num = self.next_file_num - 1;
log!(
self.opt.log,
"Recovered manifest with next_file={} manifest_num={} log_num={} prev_log_num={} \
last_seq={}",
self.next_file_num,
self.manifest_num,
self.log_num,
self.prev_log_num,
self.last_seq
);
Ok(!self.reuse_manifest(&descfilename, current))
}
fn reuse_manifest(
&mut self,
current_manifest_path: &Path,
current_manifest_base: &Path,
) -> bool {
if !self.opt.reuse_manifest {
return false;
}
if let Ok((num, typ)) = parse_file_name(current_manifest_base) {
if typ != FileType::Descriptor {
return false;
}
if let Ok(size) = self.opt.env.size_of(Path::new(current_manifest_path)) {
if size >= self.opt.max_file_size {
return false;
}
assert!(self.descriptor_log.is_none());
let s = self
.opt
.env
.open_appendable_file(Path::new(current_manifest_path));
if let Ok(f) = s {
log!(self.opt.log, "reusing manifest {:?}", current_manifest_path);
self.descriptor_log = Some(LogWriter::new_with_off(f, size));
self.manifest_num = num;
return true;
} else {
log!(self.opt.log, "reuse_manifest: {}", s.err().unwrap());
}
}
}
false
}
pub fn make_input_iterator(&self, c: &Compaction) -> Box<dyn LdbIterator> {
let cap = if c.level == 0 { c.num_inputs(0) + 1 } else { 2 };
let mut iters: Vec<Box<dyn LdbIterator>> = Vec::with_capacity(cap);
for i in 0..2 {
if c.num_inputs(i) == 0 {
continue;
}
if c.level + i == 0 {
for fi in 0..c.num_inputs(i) {
let f = &c.inputs[i][fi];
let s = self.cache.borrow_mut().get_table(f.borrow().num);
if let Ok(tbl) = s {
iters.push(Box::new(tbl.iter()));
} else {
log!(
self.opt.log,
"error opening table {}: {}",
f.borrow().num,
s.err().unwrap()
);
}
}
} else {
iters.push(Box::new(new_version_iter(
c.inputs[i].clone(),
self.cache.clone(),
self.opt.cmp.clone(),
)));
}
}
assert!(iters.len() <= cap);
let cmp: Rc<Box<dyn Cmp>> = Rc::new(Box::new(self.cmp.clone()));
Box::new(MergingIter::new(cmp, iters))
}
}
struct Builder {
deleted: [Vec<FileNum>; NUM_LEVELS],
added: [Vec<FileMetaHandle>; NUM_LEVELS],
}
impl Builder {
fn new() -> Builder {
Builder {
deleted: Default::default(),
added: Default::default(),
}
}
fn apply(&mut self, edit: &VersionEdit, compaction_ptrs: &mut [Bytes; NUM_LEVELS]) {
for c in edit.compaction_ptrs.iter() {
compaction_ptrs[c.level].clone_from(&c.key);
}
for &(level, num) in edit.deleted.iter() {
self.deleted[level].push(num);
}
for &(level, ref f) in edit.new_files.iter() {
let mut f = f.clone();
f.allowed_seeks = f.size / 16384;
if f.allowed_seeks < 100 {
f.allowed_seeks = 100;
}
self.deleted[level] = self.deleted[level]
.iter()
.filter_map(|d| if *d != f.num { Some(*d) } else { None })
.collect();
self.added[level].push(share(f));
}
}
fn maybe_add_file(
&mut self,
cmp: &InternalKeyCmp,
v: &mut Version,
level: usize,
f: FileMetaHandle,
) {
if self.deleted[level].iter().any(|d| *d == f.borrow().num) {
return;
}
{
let files = &v.files[level];
if level > 0 && !files.is_empty() {
assert_eq!(
cmp.cmp(
&files[files.len() - 1].borrow().largest,
&f.borrow().smallest
),
Ordering::Less
);
}
}
v.files[level].push(f);
}
fn save_to(&mut self, cmp: &InternalKeyCmp, base: &Shared<Version>, v: &mut Version) {
for level in 0..NUM_LEVELS {
sort_files_by_smallest(cmp, &mut self.added[level]);
sort_files_by_smallest(cmp, &mut base.borrow_mut().files[level]);
let added = self.added[level].clone();
let basefiles = base.borrow().files[level].clone();
v.files[level].reserve(basefiles.len() + self.added[level].len());
let iadded = added.into_iter();
let ibasefiles = basefiles.into_iter();
let merged = merge_iters(iadded, ibasefiles, |a, b| {
cmp.cmp(&a.borrow().smallest, &b.borrow().smallest)
});
for m in merged {
self.maybe_add_file(cmp, v, level, m);
}
if level == 0 {
continue;
}
for i in 1..v.files[level].len() {
let (prev_end, this_begin) = (
&v.files[level][i - 1].borrow().largest,
&v.files[level][i].borrow().smallest,
);
assert!(cmp.cmp(prev_end, this_begin) < Ordering::Equal);
}
}
}
}
fn manifest_name(file_num: FileNum) -> PathBuf {
Path::new(&format!("MANIFEST-{:06}", file_num)).to_owned()
}
pub fn manifest_file_name<P: AsRef<Path>>(dbname: P, file_num: FileNum) -> PathBuf {
dbname.as_ref().join(manifest_name(file_num))
}
fn temp_file_name<P: AsRef<Path>>(dbname: P, file_num: FileNum) -> PathBuf {
dbname.as_ref().join(format!("{:06}.dbtmp", file_num))
}
fn current_file_name<P: AsRef<Path>>(dbname: P) -> PathBuf {
dbname.as_ref().join("CURRENT").to_owned()
}
pub fn read_current_file(env: &dyn Env, dbname: &Path) -> Result<String> {
let mut current = String::new();
let mut f = env.open_sequential_file(Path::new(¤t_file_name(dbname)))?;
f.read_to_string(&mut current)?;
if current.is_empty() || !current.ends_with('\n') {
return err(
StatusCode::Corruption,
"current file is empty or has no newline",
);
}
Ok(current)
}
pub fn set_current_file<P: AsRef<Path>>(
env: &dyn Env,
dbname: P,
manifest_file_num: FileNum,
) -> Result<()> {
let dbname = dbname.as_ref();
let manifest_base = manifest_name(manifest_file_num);
let tempfile = temp_file_name(dbname, manifest_file_num);
{
let mut f = env.open_writable_file(Path::new(&tempfile))?;
f.write_all(manifest_base.display().to_string().as_bytes())?;
f.write_all(b"\n")?;
}
let currentfile = current_file_name(dbname);
if let Err(e) = env.rename(Path::new(&tempfile), Path::new(¤tfile)) {
let _ = env.delete(Path::new(&tempfile));
return Err(e);
}
Ok(())
}
fn sort_files_by_smallest<C: Cmp>(cmp: &C, files: &mut [FileMetaHandle]) {
files.sort_by(|a, b| cmp.cmp(&a.borrow().smallest, &b.borrow().smallest))
}
fn merge_iters<
Item,
C: Fn(&Item, &Item) -> Ordering,
I: Iterator<Item = Item>,
J: Iterator<Item = Item>,
>(
mut iter_a: I,
mut iter_b: J,
cmp: C,
) -> Vec<Item> {
let mut a = iter_a.next();
let mut b = iter_b.next();
let mut out = vec![];
while a.is_some() && b.is_some() {
let ord = cmp(a.as_ref().unwrap(), b.as_ref().unwrap());
if ord == Ordering::Less {
out.push(a.unwrap());
a = iter_a.next();
} else {
out.push(b.unwrap());
b = iter_b.next();
}
}
if let Some(a_) = a {
out.push(a_);
}
if let Some(b_) = b {
out.push(b_);
}
for a in iter_a {
out.push(a);
}
for b in iter_b {
out.push(b);
}
out
}
fn get_range<'a, C: Cmp, I: Iterator<Item = &'a FileMetaHandle>>(
c: &C,
files: I,
) -> (Bytes, Bytes) {
let mut smallest = None;
let mut largest = None;
for f in files {
if smallest.is_none() {
smallest = Some(f.borrow().smallest.clone());
}
if largest.is_none() {
largest = Some(f.borrow().largest.clone());
}
let f = f.borrow();
if c.cmp(&f.smallest, smallest.as_ref().unwrap()) == Ordering::Less {
smallest = Some(f.smallest.clone());
}
if c.cmp(&f.largest, largest.as_ref().unwrap()) == Ordering::Greater {
largest = Some(f.largest.clone());
}
}
(smallest.unwrap(), largest.unwrap())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::cache::Cache;
use crate::cmp::DefaultCmp;
use crate::key_types::LookupKey;
use crate::test_util::LdbIteratorIter;
use crate::types::FileMetaData;
use crate::version::testutil::make_version;
fn example_files() -> Vec<FileMetaHandle> {
let f1 = FileMetaData {
num: 1,
size: 10,
smallest: b"f".to_vec().into(),
largest: b"g".to_vec().into(),
..Default::default()
};
let f2 = FileMetaData {
num: 2,
size: 20,
smallest: b"e".to_vec().into(),
largest: b"f".to_vec().into(),
..Default::default()
};
let f3 = FileMetaData {
num: 3,
size: 30,
smallest: b"a".to_vec().into(),
largest: b"b".to_vec().into(),
..Default::default()
};
let f4 = FileMetaData {
num: 4,
size: 40,
smallest: b"q".to_vec().into(),
largest: b"z".to_vec().into(),
..Default::default()
};
vec![f1, f2, f3, f4].into_iter().map(share).collect()
}
#[test]
fn test_version_set_merge_iters() {
let v1 = vec![2, 4, 6, 8, 10];
let v2 = vec![1, 3, 5, 7];
assert_eq!(
vec![1, 2, 3, 4, 5, 6, 7, 8, 10],
merge_iters(v1.into_iter(), v2.into_iter(), |a, b| a.cmp(b))
);
}
#[test]
fn test_version_set_total_size() {
assert_eq!(100, total_size(example_files().iter()));
}
#[test]
fn test_version_set_get_range() {
let cmp = DefaultCmp;
let fs = example_files();
assert_eq!(
(b"a".to_vec().into(), b"z".to_vec().into()),
get_range(&cmp, fs.iter())
);
}
#[test]
fn test_version_set_builder() {
let (v, opt) = make_version();
let v = share(v);
let fmd = FileMetaData {
num: 21,
size: 123,
smallest: LookupKey::new(b"klm", 777).internal_key().to_vec().into(),
largest: LookupKey::new(b"kop", 700).internal_key().to_vec().into(),
..Default::default()
};
let mut ve = VersionEdit::new();
ve.add_file(1, fmd);
ve.delete_file(1, 21);
ve.delete_file(0, 2);
ve.set_compact_pointer(2, LookupKey::new(b"xxx", 123).internal_key());
let mut b = Builder::new();
let mut ptrs: [Bytes; NUM_LEVELS] = Default::default();
b.apply(&ve, &mut ptrs);
assert_eq!(
&[120_u8, 120, 120, 1, 123, 0, 0, 0, 0, 0, 0],
ptrs[2].as_ref()
);
assert_eq!(2, b.deleted[0][0]);
assert_eq!(1, b.added[1].len());
let mut v2 = Version::new(
share(TableCache::new(
"db",
opt.clone(),
share(Cache::new(128)),
100,
)),
opt.cmp.clone(),
);
b.save_to(&InternalKeyCmp(opt.cmp.clone()), &v, &mut v2);
assert_eq!(1, v2.files[0].len());
assert_eq!(4, v2.files[1].len());
assert_eq!(21, v2.files[1][3].borrow().num);
}
#[test]
fn test_version_set_log_and_apply() {
let (_, opt) = make_version();
let mut vs = VersionSet::new(
"db",
opt.clone(),
share(TableCache::new(
"db",
opt.clone(),
share(Cache::new(128)),
100,
)),
);
assert_eq!(2, vs.new_file_number());
{
let mut ve = VersionEdit::new();
ve.set_comparator_name("leveldb.BytewiseComparator");
ve.set_log_num(10);
ve.set_next_file(20);
ve.set_last_seq(30);
let manifest = manifest_file_name("db", 19);
let mffile = opt.env.open_writable_file(Path::new(&manifest)).unwrap();
let mut lw = LogWriter::new(mffile);
lw.add_record(&ve.encode()).unwrap();
lw.flush().unwrap();
set_current_file(opt.env.as_ref().as_ref(), "db", 19).unwrap();
}
{
vs.recover().unwrap();
assert_eq!(10, vs.log_num);
assert_eq!(21, vs.next_file_num);
assert_eq!(30, vs.last_seq);
assert_eq!(0, vs.current.as_ref().unwrap().borrow().files[0].len());
assert_eq!(0, vs.current.as_ref().unwrap().borrow().files[1].len());
assert_eq!(35, vs.write_snapshot().unwrap());
}
{
let mut ve = VersionEdit::new();
ve.set_log_num(11);
let fmd = FileMetaData {
num: 21,
size: 123,
smallest: LookupKey::new(b"abc", 777).internal_key().to_vec().into(),
largest: LookupKey::new(b"def", 700).internal_key().to_vec().into(),
..Default::default()
};
ve.add_file(1, fmd);
vs.log_and_apply(ve).unwrap();
assert!(opt.env.exists(&Path::new("db").join("CURRENT")).unwrap());
assert!(opt
.env
.exists(&Path::new("db").join("MANIFEST-000019"))
.unwrap());
assert_eq!(21, vs.new_file_number());
assert_eq!(22, vs.next_file_num);
assert_eq!(30, vs.last_seq);
assert_eq!(11, vs.log_num);
assert_eq!(0, vs.current.as_ref().unwrap().borrow().files[0].len());
assert_eq!(1, vs.current.as_ref().unwrap().borrow().files[1].len());
assert_eq!(63, vs.write_snapshot().unwrap());
}
}
#[test]
fn test_version_set_utils() {
let (v, opt) = make_version();
let mut vs = VersionSet::new(
"db",
opt.clone(),
share(TableCache::new("db", opt, share(Cache::new(128)), 100)),
);
vs.add_version(v);
assert_eq!(9, vs.live_files().len());
assert!(vs.live_files().contains(&3));
let v = vs.current();
let v = v.borrow();
assert_eq!(483, v.num_level_bytes(0));
assert_eq!(651, v.num_level_bytes(1));
assert_eq!(468, v.num_level_bytes(2));
assert_eq!(2, v.num_level_files(0));
assert_eq!(3, v.num_level_files(1));
assert_eq!(2, v.num_level_files(2));
assert_eq!(2, vs.new_file_number());
assert_eq!(3, vs.new_file_number());
}
#[test]
fn test_version_set_pick_compaction() {
let (mut v, opt) = make_version();
let mut vs = VersionSet::new(
"db",
opt.clone(),
share(TableCache::new("db", opt, share(Cache::new(128)), 100)),
);
v.compaction_score = Some(2.0);
v.compaction_level = Some(0);
vs.add_version(v);
{
let c = vs.pick_compaction().unwrap();
assert_eq!(2, c.inputs[0].len());
assert_eq!(1, c.inputs[1].len());
assert_eq!(0, c.level);
assert!(c.input_version.is_some());
}
{
let current = vs.current();
current.borrow_mut().compaction_score = None;
current.borrow_mut().compaction_level = None;
current.borrow_mut().file_to_compact_lvl = 1;
let fmd = current.borrow().files[1][0].clone();
current.borrow_mut().file_to_compact = Some(fmd);
let c = vs.pick_compaction().unwrap();
assert_eq!(3, c.inputs[0].len()); assert_eq!(1, c.inputs[1].len());
assert_eq!(1, c.level);
assert!(c.input_version.is_some());
}
}
fn iterator_properties<It: LdbIterator>(mut it: It, len: usize, cmp: Rc<Box<dyn Cmp>>) {
let mut wr = LdbIteratorIter::wrap(&mut it);
let first = wr.next().unwrap();
let mut count = 1;
wr.fold(first, |(a, _), (b, c)| {
assert_eq!(Ordering::Less, cmp.cmp(&a, &b));
count += 1;
(b, c)
});
assert_eq!(len, count);
}
#[test]
fn test_version_set_compaction() {
let (v, opt) = make_version();
let mut vs = VersionSet::new(
"db",
opt.clone(),
share(TableCache::new("db", opt, share(Cache::new(128)), 100)),
);
time_test!();
vs.add_version(v);
{
let v = vs.current();
assert_eq!(
0,
vs.approximate_offset(&v, LookupKey::new(b"aaa", 9000).internal_key())
);
assert_eq!(
232,
vs.approximate_offset(&v, LookupKey::new(b"bab", 9000).internal_key())
);
assert_eq!(
1134,
vs.approximate_offset(&v, LookupKey::new(b"fab", 9000).internal_key())
);
}
{
time_test!("compaction tests");
let from = LookupKey::new(b"000", 1000);
let to = LookupKey::new(b"ab", 1010);
let c = vs
.compact_range(0, from.internal_key(), to.internal_key())
.unwrap();
assert_eq!(2, c.inputs[0].len());
assert_eq!(1, c.inputs[1].len());
assert_eq!(1, c.grandparents.unwrap().len());
let from = LookupKey::new(b"000", 1000);
let to = LookupKey::new(b"zzz", 1010);
let c = vs
.compact_range(0, from.internal_key(), to.internal_key())
.unwrap();
assert_eq!(2, c.inputs[0].len());
assert_eq!(1, c.inputs[1].len());
assert_eq!(1, c.grandparents.as_ref().unwrap().len());
iterator_properties(
vs.make_input_iterator(&c),
12,
Rc::new(Box::new(vs.cmp.clone())),
);
let from = LookupKey::new(b"dab", 1000);
let to = LookupKey::new(b"eab", 1010);
let c = vs
.compact_range(1, from.internal_key(), to.internal_key())
.unwrap();
assert_eq!(3, c.inputs[0].len());
assert_eq!(1, c.inputs[1].len());
assert_eq!(0, c.grandparents.as_ref().unwrap().len());
iterator_properties(
vs.make_input_iterator(&c),
12,
Rc::new(Box::new(vs.cmp.clone())),
);
let from = LookupKey::new(b"fab", 1000);
let to = LookupKey::new(b"fba", 1010);
let mut c = vs
.compact_range(2, from.internal_key(), to.internal_key())
.unwrap();
c.manual = false;
assert!(c.is_trivial_move());
let from = LookupKey::new(b"000", 1000);
let to = LookupKey::new(b"zzz", 1010);
let mid = LookupKey::new(b"abc", 1010);
let mut c = vs
.compact_range(0, from.internal_key(), to.internal_key())
.unwrap();
assert!(!c.should_stop_before(from.internal_key()));
assert!(!c.should_stop_before(mid.internal_key()));
assert!(!c.should_stop_before(to.internal_key()));
let from = LookupKey::new(b"000", 1000);
let to = LookupKey::new(b"zzz", 1010);
let mut c = vs
.compact_range(0, from.internal_key(), to.internal_key())
.unwrap();
assert!(c.is_base_level_for(b"aaa"));
assert!(!c.is_base_level_for(b"hac"));
let from = LookupKey::new(b"000", 1000);
let to = LookupKey::new(b"zzz", 1010);
let mut c = vs
.compact_range(0, from.internal_key(), to.internal_key())
.unwrap();
for inp in &[(0, 0, 1), (0, 1, 2), (1, 0, 3)] {
let f = &c.inputs[inp.0][inp.1];
assert_eq!(inp.2, f.borrow().num);
}
c.add_input_deletions();
assert_eq!(23, c.edit().encode().len())
}
}
}