// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0. use crate::storage::kv::{Cursor, ScanMode, Snapshot, Statistics}; use crate::storage::mvcc::{default_not_found_error, Result}; use engine::IterOption; use engine_rocks::RocksTablePropertiesCollection; use engine_traits::{TableProperties, TablePropertiesCollection}; use engine_traits::{CF_LOCK, CF_WRITE}; use kvproto::kvrpcpb::IsolationLevel; use raftstore::coprocessor::properties::MvccProperties; use txn_types::{Key, Lock, OldValue, TimeStamp, Value, Write, WriteRef, WriteType}; pub struct MvccReader { snapshot: S, pub statistics: Statistics, // cursors are used for speeding up scans. data_cursor: Option>, lock_cursor: Option>, pub write_cursor: Option>, scan_mode: Option, key_only: bool, fill_cache: bool, isolation_level: IsolationLevel, } impl MvccReader { pub fn new( snapshot: S, scan_mode: Option, fill_cache: bool, isolation_level: IsolationLevel, ) -> Self { Self { snapshot, statistics: Statistics::default(), data_cursor: None, lock_cursor: None, write_cursor: None, scan_mode, isolation_level, key_only: false, fill_cache, } } pub fn get_statistics(&self) -> &Statistics { &self.statistics } pub fn collect_statistics_into(&mut self, stats: &mut Statistics) { stats.add(&self.statistics); self.statistics = Statistics::default(); } pub fn set_key_only(&mut self, key_only: bool) { self.key_only = key_only; } pub fn load_data(&mut self, key: &Key, write: Write) -> Result { assert_eq!(write.write_type, WriteType::Put); if self.key_only { return Ok(vec![]); } if let Some(val) = write.short_value { return Ok(val); } if self.scan_mode.is_some() && self.data_cursor.is_none() { let iter_opt = IterOption::new(None, None, self.fill_cache); self.data_cursor = Some(self.snapshot.iter(iter_opt, self.get_scan_mode(true))?); } let k = key.clone().append_ts(write.start_ts); let val = if let Some(ref mut cursor) = self.data_cursor { cursor .get(&k, &mut self.statistics.data)? .map(|v| v.to_vec()) } else { self.statistics.data.get += 1; self.snapshot.get(&k)? }; match val { Some(val) => { self.statistics.data.processed_keys += 1; Ok(val) } None => Err(default_not_found_error(key.to_raw()?, "get")), } } pub fn load_lock(&mut self, key: &Key) -> Result> { if self.scan_mode.is_some() && self.lock_cursor.is_none() { let iter_opt = IterOption::new(None, None, true); let iter = self .snapshot .iter_cf(CF_LOCK, iter_opt, self.get_scan_mode(true))?; self.lock_cursor = Some(iter); } let res = if let Some(ref mut cursor) = self.lock_cursor { match cursor.get(key, &mut self.statistics.lock)? { Some(v) => Some(Lock::parse(v)?), None => None, } } else { self.statistics.lock.get += 1; match self.snapshot.get_cf(CF_LOCK, key)? { Some(v) => Some(Lock::parse(&v)?), None => None, } }; Ok(res) } fn get_scan_mode(&self, allow_backward: bool) -> ScanMode { match self.scan_mode { Some(ScanMode::Forward) => ScanMode::Forward, Some(ScanMode::Backward) if allow_backward => ScanMode::Backward, _ => ScanMode::Mixed, } } pub fn seek_write(&mut self, key: &Key, ts: TimeStamp) -> Result> { if self.scan_mode.is_some() { if self.write_cursor.is_none() { let iter_opt = IterOption::new(None, None, self.fill_cache); let iter = self .snapshot .iter_cf(CF_WRITE, iter_opt, self.get_scan_mode(false))?; self.write_cursor = Some(iter); } } else { // use prefix bloom filter let iter_opt = IterOption::default() .use_prefix_seek() .set_prefix_same_as_start(true); let iter = self.snapshot.iter_cf(CF_WRITE, iter_opt, ScanMode::Mixed)?; self.write_cursor = Some(iter); } let cursor = self.write_cursor.as_mut().unwrap(); let ok = cursor.near_seek(&key.clone().append_ts(ts), &mut self.statistics.write)?; if !ok { return Ok(None); } let write_key = cursor.key(&mut self.statistics.write); let commit_ts = Key::decode_ts_from(write_key)?; if !Key::is_user_key_eq(write_key, key.as_encoded()) { return Ok(None); } let write = WriteRef::parse(cursor.value(&mut self.statistics.write))?.to_owned(); Ok(Some((commit_ts, write))) } /// Checks if there is a lock which blocks reading the key at the given ts. /// Returns the blocking lock as the `Err` variant. fn check_lock(&mut self, key: &Key, ts: TimeStamp) -> Result<()> { if let Some(lock) = self.load_lock(key)? { if let Err(e) = lock.check_ts_conflict(key, ts, &Default::default()) { self.statistics.lock.processed_keys += 1; return Err(e.into()); } } Ok(()) } pub fn get( &mut self, key: &Key, ts: TimeStamp, skip_lock_check: bool, ) -> Result> { if !skip_lock_check { // Check for locks that signal concurrent writes. match self.isolation_level { IsolationLevel::Si => self.check_lock(key, ts)?, IsolationLevel::Rc => {} } } if let Some(write) = self.get_write(key, ts)? { Ok(Some(self.load_data(key, write)?)) } else { Ok(None) } } pub fn get_write(&mut self, key: &Key, mut ts: TimeStamp) -> Result> { loop { match self.seek_write(key, ts)? { Some((commit_ts, write)) => match write.write_type { WriteType::Put => { return Ok(Some(write)); } WriteType::Delete => { return Ok(None); } WriteType::Lock | WriteType::Rollback => ts = commit_ts.prev(), }, None => return Ok(None), } } } pub fn get_txn_commit_info( &mut self, key: &Key, start_ts: TimeStamp, ) -> Result> { // It's possible a txn with a small `start_ts` has a greater `commit_ts` than a txn with // a greater `start_ts` in pessimistic transaction. // I.e., txn_1.commit_ts > txn_2.commit_ts > txn_2.start_ts > txn_1.start_ts. // // Scan all the versions from `TimeStamp::max()` to `start_ts`. let mut seek_ts = TimeStamp::max(); while let Some((commit_ts, write)) = self.seek_write(key, seek_ts)? { if write.start_ts == start_ts { return Ok(Some((commit_ts, write.write_type))); } if commit_ts <= start_ts { break; } seek_ts = commit_ts.prev(); } Ok(None) } fn create_data_cursor(&mut self) -> Result<()> { if self.data_cursor.is_none() { let iter_opt = IterOption::new(None, None, true); let iter = self.snapshot.iter(iter_opt, self.get_scan_mode(true))?; self.data_cursor = Some(iter); } Ok(()) } fn create_write_cursor(&mut self) -> Result<()> { if self.write_cursor.is_none() { let iter_opt = IterOption::new(None, None, true); let iter = self .snapshot .iter_cf(CF_WRITE, iter_opt, self.get_scan_mode(true))?; self.write_cursor = Some(iter); } Ok(()) } fn create_lock_cursor(&mut self) -> Result<()> { if self.lock_cursor.is_none() { let iter_opt = IterOption::new(None, None, true); let iter = self .snapshot .iter_cf(CF_LOCK, iter_opt, self.get_scan_mode(true))?; self.lock_cursor = Some(iter); } Ok(()) } /// Return the first committed key for which `start_ts` equals to `ts` pub fn seek_ts(&mut self, ts: TimeStamp) -> Result> { assert!(self.scan_mode.is_some()); self.create_write_cursor()?; let cursor = self.write_cursor.as_mut().unwrap(); let mut ok = cursor.seek_to_first(&mut self.statistics.write); while ok { if WriteRef::parse(cursor.value(&mut self.statistics.write))?.start_ts == ts { return Ok(Some( Key::from_encoded(cursor.key(&mut self.statistics.write).to_vec()) .truncate_ts()?, )); } ok = cursor.next(&mut self.statistics.write); } Ok(None) } /// Scan locks that satisfies `filter(lock)` returns true, from the given start key `start`. /// At most `limit` locks will be returned. If `limit` is set to `0`, it means unlimited. /// /// The return type is `(locks, is_remain)`. `is_remain` indicates whether there MAY be /// remaining locks that can be scanned. pub fn scan_locks( &mut self, start: Option<&Key>, filter: F, limit: usize, ) -> Result<(Vec<(Key, Lock)>, bool)> where F: Fn(&Lock) -> bool, { self.create_lock_cursor()?; let cursor = self.lock_cursor.as_mut().unwrap(); let ok = match start { Some(ref x) => cursor.seek(x, &mut self.statistics.lock)?, None => cursor.seek_to_first(&mut self.statistics.lock), }; if !ok { return Ok((vec![], false)); } let mut locks = Vec::with_capacity(limit); while cursor.valid()? { let key = Key::from_encoded_slice(cursor.key(&mut self.statistics.lock)); let lock = Lock::parse(cursor.value(&mut self.statistics.lock))?; if filter(&lock) { locks.push((key, lock)); if limit > 0 && locks.len() == limit { return Ok((locks, true)); } } cursor.next(&mut self.statistics.lock); } self.statistics.lock.processed_keys += locks.len(); // If we reach here, `cursor.valid()` is `false`, so there MUST be no more locks. Ok((locks, false)) } pub fn scan_keys( &mut self, mut start: Option, limit: usize, ) -> Result<(Vec, Option)> { let iter_opt = IterOption::new(None, None, self.fill_cache); let scan_mode = self.get_scan_mode(false); let mut cursor = self.snapshot.iter_cf(CF_WRITE, iter_opt, scan_mode)?; let mut keys = vec![]; loop { let ok = match start { Some(ref x) => cursor.near_seek(x, &mut self.statistics.write)?, None => cursor.seek_to_first(&mut self.statistics.write), }; if !ok { return Ok((keys, None)); } if keys.len() >= limit { self.statistics.write.processed_keys += keys.len(); return Ok((keys, start)); } let key = Key::from_encoded(cursor.key(&mut self.statistics.write).to_vec()).truncate_ts()?; start = Some(key.clone().append_ts(TimeStamp::zero())); keys.push(key); } } // Get all Value of the given key in CF_DEFAULT pub fn scan_values_in_default(&mut self, key: &Key) -> Result> { self.create_data_cursor()?; let cursor = self.data_cursor.as_mut().unwrap(); let mut ok = cursor.seek(key, &mut self.statistics.data)?; if !ok { return Ok(vec![]); } let mut v = vec![]; while ok { let cur_key = cursor.key(&mut self.statistics.data); let ts = Key::decode_ts_from(cur_key)?; if Key::is_user_key_eq(cur_key, key.as_encoded()) { v.push((ts, cursor.value(&mut self.statistics.data).to_vec())); } else { break; } ok = cursor.next(&mut self.statistics.data); } Ok(v) } pub fn need_gc(&self, safe_point: TimeStamp, ratio_threshold: f64) -> bool { let prop = match self.snapshot.get_properties_cf(CF_WRITE) { Ok(v) => v, Err(_) => return true, }; check_need_gc(safe_point, ratio_threshold, prop) } /// Read the old value for key for CDC. /// `prev_write` stands for the previous write record of the key /// it must be read in the caller and be passed in for optimization pub fn get_old_value( &mut self, key: &Key, start_ts: TimeStamp, prev_write_loaded: bool, prev_write: Option, ) -> Result { if prev_write_loaded && prev_write.is_none() { return Ok(OldValue::None); } if let Some(prev_write) = prev_write { match prev_write.write_type { WriteType::Put => { // For Put, there must be an old value either in its // short value or in the default CF. return Ok(OldValue::Value { short_value: prev_write.short_value, start_ts: prev_write.start_ts, }); } WriteType::Delete => { // For Delete, no old value. return Ok(OldValue::None); } // For Rollback and Lock, it's unknown whether there is a more // previous valid write. Call `get_write` to get a valid // previous write. WriteType::Rollback | WriteType::Lock => (), } } Ok(match self.get_write(key, start_ts)? { Some(write) => OldValue::Value { short_value: write.short_value, start_ts: write.start_ts, }, None => OldValue::None, }) } } // Returns true if it needs gc. // This is for optimization purpose, does not mean to be accurate. pub fn check_need_gc( safe_point: TimeStamp, ratio_threshold: f64, write_properties: RocksTablePropertiesCollection, ) -> bool { // Always GC. if ratio_threshold < 1.0 { return true; } let props = match get_mvcc_properties(safe_point, write_properties) { Some(v) => v, None => return true, }; // No data older than safe_point to GC. if props.min_ts > safe_point { return false; } // Note: Since the properties are file-based, it can be false positive. // For example, multiple files can have a different version of the same row. // A lot of MVCC versions to GC. if props.num_versions as f64 > props.num_rows as f64 * ratio_threshold { return true; } // A lot of non-effective MVCC versions to GC. if props.num_versions as f64 > props.num_puts as f64 * ratio_threshold { return true; } false } fn get_mvcc_properties( safe_point: TimeStamp, collection: RocksTablePropertiesCollection, ) -> Option { if collection.is_empty() { return None; } // Aggregate MVCC properties. let mut props = MvccProperties::new(); for (_, v) in collection.iter() { let mvcc = match MvccProperties::decode(&v.user_collected_properties()) { Ok(v) => v, Err(_) => return None, }; // Filter out properties after safe_point. if mvcc.min_ts > safe_point { continue; } props.add(&mvcc); } Some(props) } #[cfg(test)] mod tests { use super::*; use crate::storage::kv::{Modify, WriteData}; use crate::storage::mvcc::{MvccReader, MvccTxn}; use crate::storage::{Engine, TestEngineBuilder}; use engine::rocks::util::CFOptions; use engine::rocks::DB; use engine::rocks::{self, ColumnFamilyOptions, DBOptions}; use engine::IterOption; use engine_rocks::{Compat, RocksEngine}; use engine_traits::{Mutable, WriteBatchExt}; use engine_traits::{ALL_CFS, CF_DEFAULT, CF_LOCK, CF_RAFT, CF_WRITE}; use kvproto::kvrpcpb::{Context, IsolationLevel}; use kvproto::metapb::{Peer, Region}; use raftstore::coprocessor::properties::MvccPropertiesCollectorFactory; use raftstore::store::RegionSnapshot; use std::ops::Bound; use std::sync::Arc; use std::u64; use txn_types::{LockType, Mutation}; struct RegionEngine { db: Arc, region: Region, } impl RegionEngine { fn new(db: &Arc, region: &Region) -> RegionEngine { RegionEngine { db: Arc::clone(&db), region: region.clone(), } } fn put( &mut self, pk: &[u8], start_ts: impl Into, commit_ts: impl Into, ) { let start_ts = start_ts.into(); let m = Mutation::Put((Key::from_raw(pk), vec![])); self.prewrite(m, pk, start_ts); self.commit(pk, start_ts, commit_ts); } fn lock( &mut self, pk: &[u8], start_ts: impl Into, commit_ts: impl Into, ) { let start_ts = start_ts.into(); let m = Mutation::Lock(Key::from_raw(pk)); self.prewrite(m, pk, start_ts); self.commit(pk, start_ts, commit_ts); } fn delete( &mut self, pk: &[u8], start_ts: impl Into, commit_ts: impl Into, ) { let start_ts = start_ts.into(); let m = Mutation::Delete(Key::from_raw(pk)); self.prewrite(m, pk, start_ts); self.commit(pk, start_ts, commit_ts); } fn prewrite(&mut self, m: Mutation, pk: &[u8], start_ts: impl Into) { let snap = RegionSnapshot::::from_raw(Arc::clone(&self.db), self.region.clone()); let mut txn = MvccTxn::new(snap, start_ts.into(), true); txn.prewrite(m, pk, false, 0, 0, TimeStamp::default()) .unwrap(); self.write(txn.into_modifies()); } fn prewrite_pessimistic_lock( &mut self, m: Mutation, pk: &[u8], start_ts: impl Into, ) { let snap = RegionSnapshot::::from_raw(Arc::clone(&self.db), self.region.clone()); let mut txn = MvccTxn::new(snap, start_ts.into(), true); txn.pessimistic_prewrite( m, pk, true, 0, TimeStamp::default(), 0, TimeStamp::default(), ) .unwrap(); self.write(txn.into_modifies()); } fn acquire_pessimistic_lock( &mut self, k: Key, pk: &[u8], start_ts: impl Into, for_update_ts: impl Into, ) { let snap = RegionSnapshot::::from_raw(Arc::clone(&self.db), self.region.clone()); let mut txn = MvccTxn::new(snap, start_ts.into(), true); txn.acquire_pessimistic_lock( k, pk, false, 0, for_update_ts.into(), false, TimeStamp::zero(), ) .unwrap(); self.write(txn.into_modifies()); } fn commit( &mut self, pk: &[u8], start_ts: impl Into, commit_ts: impl Into, ) { let snap = RegionSnapshot::::from_raw(Arc::clone(&self.db), self.region.clone()); let mut txn = MvccTxn::new(snap, start_ts.into(), true); txn.commit(Key::from_raw(pk), commit_ts.into()).unwrap(); self.write(txn.into_modifies()); } fn rollback(&mut self, pk: &[u8], start_ts: impl Into) { let snap = RegionSnapshot::::from_raw(Arc::clone(&self.db), self.region.clone()); let mut txn = MvccTxn::new(snap, start_ts.into(), true); txn.collapse_rollback(false); txn.rollback(Key::from_raw(pk)).unwrap(); self.write(txn.into_modifies()); } fn gc(&mut self, pk: &[u8], safe_point: impl Into + Copy) { loop { let snap = RegionSnapshot::::from_raw( Arc::clone(&self.db), self.region.clone(), ); let mut txn = MvccTxn::new(snap, safe_point.into(), true); txn.gc(Key::from_raw(pk), safe_point.into()).unwrap(); let modifies = txn.into_modifies(); if modifies.is_empty() { return; } self.write(modifies); } } fn write(&mut self, modifies: Vec) { let db = &self.db; let mut wb = db.c().write_batch(); for rev in modifies { match rev { Modify::Put(cf, k, v) => { let k = keys::data_key(k.as_encoded()); wb.put_cf(cf, &k, &v).unwrap(); } Modify::Delete(cf, k) => { let k = keys::data_key(k.as_encoded()); wb.delete_cf(cf, &k).unwrap(); } Modify::DeleteRange(cf, k1, k2, notify_only) => { if !notify_only { let k1 = keys::data_key(k1.as_encoded()); let k2 = keys::data_key(k2.as_encoded()); wb.delete_range_cf(cf, &k1, &k2).unwrap(); } } } } db.c().write(&wb).unwrap(); } fn flush(&mut self) { for cf in ALL_CFS { let cf = rocks::util::get_cf_handle(&self.db, cf).unwrap(); self.db.flush_cf(cf, true).unwrap(); } } fn compact(&mut self) { for cf in ALL_CFS { let cf = rocks::util::get_cf_handle(&self.db, cf).unwrap(); self.db.compact_range_cf(cf, None, None); } } } fn open_db(path: &str, with_properties: bool) -> Arc { let db_opts = DBOptions::new(); let mut cf_opts = ColumnFamilyOptions::new(); cf_opts.set_write_buffer_size(32 * 1024 * 1024); if with_properties { let f = Box::new(MvccPropertiesCollectorFactory::default()); cf_opts.add_table_properties_collector_factory("tikv.test-collector", f); } let cfs_opts = vec![ CFOptions::new(CF_DEFAULT, ColumnFamilyOptions::new()), CFOptions::new(CF_RAFT, ColumnFamilyOptions::new()), CFOptions::new(CF_LOCK, ColumnFamilyOptions::new()), CFOptions::new(CF_WRITE, cf_opts), ]; Arc::new(rocks::util::new_engine_opt(path, db_opts, cfs_opts).unwrap()) } fn make_region(id: u64, start_key: Vec, end_key: Vec) -> Region { let mut peer = Peer::default(); peer.set_id(id); peer.set_store_id(id); let mut region = Region::default(); region.set_id(id); region.set_start_key(start_key); region.set_end_key(end_key); region.mut_peers().push(peer); region } fn check_need_gc( db: Arc, region: Region, safe_point: impl Into, need_gc: bool, ) -> Option { let snap = RegionSnapshot::::from_raw(db, region); let reader = MvccReader::new(snap, None, false, IsolationLevel::Si); let safe_point = safe_point.into(); assert_eq!(reader.need_gc(safe_point, 1.0), need_gc); get_mvcc_properties( safe_point, reader.snapshot.get_properties_cf(CF_WRITE).unwrap(), ) } #[test] fn test_need_gc() { let path = tempfile::Builder::new() .prefix("test_storage_mvcc_reader") .tempdir() .unwrap(); let path = path.path().to_str().unwrap(); let region = make_region(1, vec![0], vec![10]); test_without_properties(path, ®ion); test_with_properties(path, ®ion); } fn test_without_properties(path: &str, region: &Region) { let db = open_db(path, false); let mut engine = RegionEngine::new(&db, ®ion); // Put 2 keys. engine.put(&[1], 1, 1); engine.put(&[4], 2, 2); assert!(check_need_gc(Arc::clone(&db), region.clone(), 10, true).is_none()); engine.flush(); // After this flush, we have a SST file without properties. // Without properties, we always need GC. assert!(check_need_gc(Arc::clone(&db), region.clone(), 10, true).is_none()); } #[test] fn test_ts_filter() { let path = tempfile::Builder::new() .prefix("test_ts_filter") .tempdir() .unwrap(); let path = path.path().to_str().unwrap(); let region = make_region(1, vec![0], vec![13]); let db = open_db(path, true); let mut engine = RegionEngine::new(&db, ®ion); engine.put(&[2], 1, 2); engine.put(&[4], 3, 4); engine.flush(); engine.put(&[6], 5, 6); engine.put(&[8], 7, 8); engine.flush(); engine.put(&[10], 9, 10); engine.put(&[12], 11, 12); engine.flush(); let snap = RegionSnapshot::::from_raw(db, region); let tests = vec![ // set nothing. ( Bound::Unbounded, Bound::Unbounded, vec![2u64, 4, 6, 8, 10, 12], ), // test set both hint_min_ts and hint_max_ts. (Bound::Included(6), Bound::Included(8), vec![6u64, 8]), (Bound::Excluded(5), Bound::Included(8), vec![6u64, 8]), (Bound::Included(6), Bound::Excluded(9), vec![6u64, 8]), (Bound::Excluded(5), Bound::Excluded(9), vec![6u64, 8]), // test set only hint_min_ts. (Bound::Included(10), Bound::Unbounded, vec![10u64, 12]), (Bound::Excluded(9), Bound::Unbounded, vec![10u64, 12]), // test set only hint_max_ts. (Bound::Unbounded, Bound::Included(7), vec![2u64, 4, 6, 8]), (Bound::Unbounded, Bound::Excluded(8), vec![2u64, 4, 6, 8]), ]; for (_, &(min, max, ref res)) in tests.iter().enumerate() { let mut iopt = IterOption::default(); iopt.set_hint_min_ts(min); iopt.set_hint_max_ts(max); let mut iter = snap.iter_cf(CF_WRITE, iopt).unwrap(); for (i, expect_ts) in res.iter().enumerate() { if i == 0 { assert_eq!(iter.seek_to_first().unwrap(), true); } else { assert_eq!(iter.next().unwrap(), true); } let ts = Key::decode_ts_from(iter.key()).unwrap(); assert_eq!(ts.into_inner(), *expect_ts); } assert_eq!(iter.next().unwrap(), false); } } #[test] fn test_ts_filter_lost_delete() { let dir = tempfile::Builder::new() .prefix("test_ts_filter_lost_deletion") .tempdir() .unwrap(); let path = dir.path().to_str().unwrap(); let region = make_region(1, vec![0], vec![]); let db = open_db(&path, true); let mut engine = RegionEngine::new(&db, ®ion); let key1 = &[1]; engine.put(key1, 2, 3); engine.flush(); engine.compact(); // Delete key 1 commit ts@5 and GC@6 // Put key 2 commit ts@7 let key2 = &[2]; engine.put(key2, 6, 7); engine.delete(key1, 4, 5); engine.gc(key1, 6); engine.flush(); // Scan kv with ts filter [1, 6]. let mut iopt = IterOption::default(); iopt.set_hint_min_ts(Bound::Included(1)); iopt.set_hint_max_ts(Bound::Included(6)); let snap = RegionSnapshot::::from_raw(db, region); let mut iter = snap.iter_cf(CF_WRITE, iopt).unwrap(); // Must not omit the latest deletion of key1 to prevent seeing outdated record. assert_eq!(iter.seek_to_first().unwrap(), true); assert_eq!( Key::from_encoded_slice(iter.key()) .to_raw() .unwrap() .as_slice(), key2 ); assert_eq!(iter.next().unwrap(), false); } fn test_with_properties(path: &str, region: &Region) { let db = open_db(path, true); let mut engine = RegionEngine::new(&db, ®ion); // Put 2 keys. engine.put(&[2], 3, 3); engine.put(&[3], 4, 4); engine.flush(); // After this flush, we have a SST file w/ properties, plus the SST // file w/o properties from previous flush. We always need GC as // long as we can't get properties from any SST files. assert!(check_need_gc(Arc::clone(&db), region.clone(), 10, true).is_none()); engine.compact(); // After this compact, the two SST files are compacted into a new // SST file with properties. Now all SST files have properties and // all keys have only one version, so we don't need gc. let props = check_need_gc(Arc::clone(&db), region.clone(), 10, false).unwrap(); assert_eq!(props.min_ts, 1.into()); assert_eq!(props.max_ts, 4.into()); assert_eq!(props.num_rows, 4); assert_eq!(props.num_puts, 4); assert_eq!(props.num_versions, 4); assert_eq!(props.max_row_versions, 1); // Put 2 more keys and delete them. engine.put(&[5], 5, 5); engine.put(&[6], 6, 6); engine.delete(&[5], 7, 7); engine.delete(&[6], 8, 8); engine.flush(); // After this flush, keys 5,6 in the new SST file have more than one // versions, so we need gc. let props = check_need_gc(Arc::clone(&db), region.clone(), 10, true).unwrap(); assert_eq!(props.min_ts, 1.into()); assert_eq!(props.max_ts, 8.into()); assert_eq!(props.num_rows, 6); assert_eq!(props.num_puts, 6); assert_eq!(props.num_versions, 8); assert_eq!(props.max_row_versions, 2); // But if the `safe_point` is older than all versions, we don't need gc too. let props = check_need_gc(Arc::clone(&db), region.clone(), 0, false).unwrap(); assert_eq!(props.min_ts, TimeStamp::max()); assert_eq!(props.max_ts, TimeStamp::zero()); assert_eq!(props.num_rows, 0); assert_eq!(props.num_puts, 0); assert_eq!(props.num_versions, 0); assert_eq!(props.max_row_versions, 0); // We gc the two deleted keys manually. engine.gc(&[5], 10); engine.gc(&[6], 10); engine.compact(); // After this compact, all versions of keys 5,6 are deleted, // no keys have more than one versions, so we don't need gc. let props = check_need_gc(Arc::clone(&db), region.clone(), 10, false).unwrap(); assert_eq!(props.min_ts, 1.into()); assert_eq!(props.max_ts, 4.into()); assert_eq!(props.num_rows, 4); assert_eq!(props.num_puts, 4); assert_eq!(props.num_versions, 4); assert_eq!(props.max_row_versions, 1); // A single lock version need gc. engine.lock(&[7], 9, 9); engine.flush(); let props = check_need_gc(Arc::clone(&db), region.clone(), 10, true).unwrap(); assert_eq!(props.min_ts, 1.into()); assert_eq!(props.max_ts, 9.into()); assert_eq!(props.num_rows, 5); assert_eq!(props.num_puts, 4); assert_eq!(props.num_versions, 5); assert_eq!(props.max_row_versions, 1); } #[test] fn test_get_txn_commit_info() { let path = tempfile::Builder::new() .prefix("_test_storage_mvcc_reader_get_txn_commit_info") .tempdir() .unwrap(); let path = path.path().to_str().unwrap(); let region = make_region(1, vec![], vec![]); let db = open_db(path, true); let mut engine = RegionEngine::new(&db, ®ion); let (k, v) = (b"k", b"v"); let m = Mutation::Put((Key::from_raw(k), v.to_vec())); engine.prewrite(m, k, 1); engine.commit(k, 1, 10); engine.rollback(k, 5); engine.rollback(k, 20); let m = Mutation::Put((Key::from_raw(k), v.to_vec())); engine.prewrite(m, k, 25); engine.commit(k, 25, 30); let m = Mutation::Put((Key::from_raw(k), v.to_vec())); engine.prewrite(m, k, 35); engine.commit(k, 35, 40); let m = Mutation::Put((Key::from_raw(k), v.to_vec())); engine.acquire_pessimistic_lock(Key::from_raw(k), k, 45, 45); engine.prewrite_pessimistic_lock(m, k, 45); engine.commit(k, 45, 50); let snap = RegionSnapshot::::from_raw(db, region); let mut reader = MvccReader::new(snap, None, false, IsolationLevel::Si); // Let's assume `50_45 PUT` means a commit version with start ts is 45 and commit ts // is 50. // Commit versions: [50_45 PUT, 45_40 PUT, 40_35 PUT, 30_25 PUT, 20_20 Rollback, 10_1 PUT, 5_5 Rollback]. let key = Key::from_raw(k); let (commit_ts, write_type) = reader .get_txn_commit_info(&key, 45.into()) .unwrap() .unwrap(); assert_eq!(commit_ts, 50.into()); assert_eq!(write_type, WriteType::Put); let (commit_ts, write_type) = reader .get_txn_commit_info(&key, 35.into()) .unwrap() .unwrap(); assert_eq!(commit_ts, 40.into()); assert_eq!(write_type, WriteType::Put); let (commit_ts, write_type) = reader .get_txn_commit_info(&key, 25.into()) .unwrap() .unwrap(); assert_eq!(commit_ts, 30.into()); assert_eq!(write_type, WriteType::Put); let (commit_ts, write_type) = reader .get_txn_commit_info(&key, 20.into()) .unwrap() .unwrap(); assert_eq!(commit_ts, 20.into()); assert_eq!(write_type, WriteType::Rollback); let (commit_ts, write_type) = reader.get_txn_commit_info(&key, 1.into()).unwrap().unwrap(); assert_eq!(commit_ts, 10.into()); assert_eq!(write_type, WriteType::Put); let (commit_ts, write_type) = reader.get_txn_commit_info(&key, 5.into()).unwrap().unwrap(); assert_eq!(commit_ts, 5.into()); assert_eq!(write_type, WriteType::Rollback); let seek_old = reader.get_statistics().write.seek; assert!(reader .get_txn_commit_info(&key, 30.into()) .unwrap() .is_none()); let seek_new = reader.get_statistics().write.seek; // `get_txn_commit_info(&key, 30)` stopped at `30_25 PUT`. assert_eq!(seek_new - seek_old, 3); } #[test] fn test_get_txn_commit_info_of_pessimistic_txn() { let path = tempfile::Builder::new() .prefix("_test_storage_mvcc_reader_get_txn_commit_info_of_pessimistic_txn") .tempdir() .unwrap(); let path = path.path().to_str().unwrap(); let region = make_region(1, vec![], vec![]); let db = open_db(path, true); let mut engine = RegionEngine::new(&db, ®ion); let (k, v) = (b"k", b"v"); let key = Key::from_raw(k); let m = Mutation::Put((key.clone(), v.to_vec())); // txn: start_ts = 2, commit_ts = 3 engine.acquire_pessimistic_lock(key.clone(), k, 2, 2); engine.prewrite_pessimistic_lock(m.clone(), k, 2); engine.commit(k, 2, 3); // txn: start_ts = 1, commit_ts = 4 engine.acquire_pessimistic_lock(key.clone(), k, 1, 3); engine.prewrite_pessimistic_lock(m, k, 1); engine.commit(k, 1, 4); let snap = RegionSnapshot::::from_raw(db, region); let mut reader = MvccReader::new(snap, None, false, IsolationLevel::Si); let (commit_ts, write_type) = reader.get_txn_commit_info(&key, 2.into()).unwrap().unwrap(); assert_eq!(commit_ts, 3.into()); assert_eq!(write_type, WriteType::Put); let (commit_ts, write_type) = reader.get_txn_commit_info(&key, 1.into()).unwrap().unwrap(); assert_eq!(commit_ts, 4.into()); assert_eq!(write_type, WriteType::Put); } #[test] fn test_seek_write() { let path = tempfile::Builder::new() .prefix("_test_storage_mvcc_reader_seek_write") .tempdir() .unwrap(); let path = path.path().to_str().unwrap(); let region = make_region(1, vec![], vec![]); let db = open_db(path, true); let mut engine = RegionEngine::new(&db, ®ion); let (k, v) = (b"k", b"v"); let m = Mutation::Put((Key::from_raw(k), v.to_vec())); engine.prewrite(m.clone(), k, 1); engine.commit(k, 1, 5); engine.rollback(k, 3); engine.rollback(k, 7); engine.prewrite(m.clone(), k, 15); engine.commit(k, 15, 17); // Timestamp overlap with the previous transaction. engine.acquire_pessimistic_lock(Key::from_raw(k), k, 10, 18); engine.prewrite_pessimistic_lock(Mutation::Lock(Key::from_raw(k)), k, 10); engine.commit(k, 10, 20); engine.prewrite(m, k, 23); engine.commit(k, 23, 25); // Let's assume `2_1 PUT` means a commit version with start ts is 1 and commit ts // is 2. // Commit versions: [25_23 PUT, 20_10 PUT, 17_15 PUT, 7_7 Rollback, 5_1 PUT, 3_3 Rollback]. let snap = RegionSnapshot::::from_raw(Arc::clone(&db), region.clone()); let mut reader = MvccReader::new(snap, None, false, IsolationLevel::Si); let k = Key::from_raw(k); let (commit_ts, write) = reader.seek_write(&k, 30.into()).unwrap().unwrap(); assert_eq!(commit_ts, 25.into()); assert_eq!( write, Write::new(WriteType::Put, 23.into(), Some(v.to_vec())) ); let (commit_ts, write) = reader.seek_write(&k, 25.into()).unwrap().unwrap(); assert_eq!(commit_ts, 25.into()); assert_eq!( write, Write::new(WriteType::Put, 23.into(), Some(v.to_vec())) ); let (commit_ts, write) = reader.seek_write(&k, 20.into()).unwrap().unwrap(); assert_eq!(commit_ts, 20.into()); assert_eq!(write, Write::new(WriteType::Lock, 10.into(), None)); let (commit_ts, write) = reader.seek_write(&k, 19.into()).unwrap().unwrap(); assert_eq!(commit_ts, 17.into()); assert_eq!( write, Write::new(WriteType::Put, 15.into(), Some(v.to_vec())) ); let (commit_ts, write) = reader.seek_write(&k, 3.into()).unwrap().unwrap(); assert_eq!(commit_ts, 3.into()); assert_eq!(write, Write::new_rollback(3.into(), false)); let (commit_ts, write) = reader.seek_write(&k, 16.into()).unwrap().unwrap(); assert_eq!(commit_ts, 7.into()); assert_eq!(write, Write::new_rollback(7.into(), false)); let (commit_ts, write) = reader.seek_write(&k, 6.into()).unwrap().unwrap(); assert_eq!(commit_ts, 5.into()); assert_eq!( write, Write::new(WriteType::Put, 1.into(), Some(v.to_vec())) ); assert!(reader.seek_write(&k, 2.into()).unwrap().is_none()); // Test seek_write should not see the next key. let (k2, v2) = (b"k2", b"v2"); let m2 = Mutation::Put((Key::from_raw(k2), v2.to_vec())); engine.prewrite(m2, k2, 1); engine.commit(k2, 1, 2); let snap = RegionSnapshot::::from_raw(Arc::clone(&db), region); let mut reader = MvccReader::new(snap, None, false, IsolationLevel::Si); let (commit_ts, write) = reader .seek_write(&Key::from_raw(k2), 3.into()) .unwrap() .unwrap(); assert_eq!(commit_ts, 2.into()); assert_eq!( write, Write::new(WriteType::Put, 1.into(), Some(v2.to_vec())) ); assert!(reader.seek_write(&k, 2.into()).unwrap().is_none()); // Test seek_write touches region's end. let region1 = make_region(1, vec![], Key::from_raw(b"k1").into_encoded()); let snap = RegionSnapshot::::from_raw(Arc::clone(&db), region1); let mut reader = MvccReader::new(snap, None, false, IsolationLevel::Si); assert!(reader.seek_write(&k, 2.into()).unwrap().is_none()); } #[test] fn test_get_write() { let path = tempfile::Builder::new() .prefix("_test_storage_mvcc_reader_get_write") .tempdir() .unwrap(); let path = path.path().to_str().unwrap(); let region = make_region(1, vec![], vec![]); let db = open_db(path, true); let mut engine = RegionEngine::new(&db, ®ion); let (k, v) = (b"k", b"v"); let m = Mutation::Put((Key::from_raw(k), v.to_vec())); engine.prewrite(m, k, 1); engine.commit(k, 1, 2); engine.rollback(k, 5); engine.lock(k, 6, 7); engine.delete(k, 8, 9); let m = Mutation::Put((Key::from_raw(k), v.to_vec())); engine.prewrite(m, k, 12); engine.commit(k, 12, 14); let m = Mutation::Lock(Key::from_raw(k)); engine.acquire_pessimistic_lock(Key::from_raw(k), k, 13, 15); engine.prewrite_pessimistic_lock(m, k, 13); engine.commit(k, 13, 15); let m = Mutation::Put((Key::from_raw(k), v.to_vec())); engine.acquire_pessimistic_lock(Key::from_raw(k), k, 18, 18); engine.prewrite_pessimistic_lock(m, k, 18); engine.commit(k, 18, 20); let m = Mutation::Lock(Key::from_raw(k)); engine.acquire_pessimistic_lock(Key::from_raw(k), k, 17, 21); engine.prewrite_pessimistic_lock(m, k, 17); engine.commit(k, 17, 21); let m = Mutation::Put((Key::from_raw(k), v.to_vec())); engine.prewrite(m, k, 24); let snap = RegionSnapshot::::from_raw(db, region); let mut reader = MvccReader::new(snap, None, false, IsolationLevel::Si); // Let's assume `2_1 PUT` means a commit version with start ts is 1 and commit ts // is 2. // Commit versions: [21_17 LOCK, 20_18 PUT, 15_13 LOCK, 14_12 PUT, 9_8 DELETE, 7_6 LOCK, // 5_5 Rollback, 2_1 PUT]. let key = Key::from_raw(k); assert!(reader.get_write(&key, 1.into()).unwrap().is_none()); let write = reader.get_write(&key, 2.into()).unwrap().unwrap(); assert_eq!(write.write_type, WriteType::Put); assert_eq!(write.start_ts, 1.into()); let write = reader.get_write(&key, 5.into()).unwrap().unwrap(); assert_eq!(write.write_type, WriteType::Put); assert_eq!(write.start_ts, 1.into()); let write = reader.get_write(&key, 7.into()).unwrap().unwrap(); assert_eq!(write.write_type, WriteType::Put); assert_eq!(write.start_ts, 1.into()); assert!(reader.get_write(&key, 9.into()).unwrap().is_none()); let write = reader.get_write(&key, 14.into()).unwrap().unwrap(); assert_eq!(write.write_type, WriteType::Put); assert_eq!(write.start_ts, 12.into()); let write = reader.get_write(&key, 16.into()).unwrap().unwrap(); assert_eq!(write.write_type, WriteType::Put); assert_eq!(write.start_ts, 12.into()); let write = reader.get_write(&key, 20.into()).unwrap().unwrap(); assert_eq!(write.write_type, WriteType::Put); assert_eq!(write.start_ts, 18.into()); let write = reader.get_write(&key, 24.into()).unwrap().unwrap(); assert_eq!(write.write_type, WriteType::Put); assert_eq!(write.start_ts, 18.into()); assert!(reader .get_write(&Key::from_raw(b"j"), 100.into()) .unwrap() .is_none()); } #[test] fn test_check_lock() { let path = tempfile::Builder::new() .prefix("_test_storage_mvcc_reader_check_lock") .tempdir() .unwrap(); let path = path.path().to_str().unwrap(); let region = make_region(1, vec![], vec![]); let db = open_db(path, true); let mut engine = RegionEngine::new(&db, ®ion); let (k1, k2, k3, k4, v) = (b"k1", b"k2", b"k3", b"k4", b"v"); engine.prewrite(Mutation::Put((Key::from_raw(k1), v.to_vec())), k1, 5); engine.prewrite(Mutation::Put((Key::from_raw(k2), v.to_vec())), k1, 5); engine.prewrite(Mutation::Lock(Key::from_raw(k3)), k1, 5); let snap = RegionSnapshot::::from_raw(Arc::clone(&db), region.clone()); let mut reader = MvccReader::new(snap, None, false, IsolationLevel::Si); // Ignore the lock if read ts is less than the lock version assert!(reader.check_lock(&Key::from_raw(k1), 4.into()).is_ok()); assert!(reader.check_lock(&Key::from_raw(k2), 4.into()).is_ok()); // Returns the lock if read ts >= lock version assert!(reader.check_lock(&Key::from_raw(k1), 6.into()).is_err()); assert!(reader.check_lock(&Key::from_raw(k2), 6.into()).is_err()); // Read locks don't block any read operation assert!(reader.check_lock(&Key::from_raw(k3), 6.into()).is_ok()); // Ignore the primary lock when reading the latest committed version by setting TimeStamp::max() as ts assert!(reader .check_lock(&Key::from_raw(k1), TimeStamp::max()) .is_ok()); // Should not ignore the secondary lock even though reading the latest version assert!(reader .check_lock(&Key::from_raw(k2), TimeStamp::max()) .is_err()); // Commit the primary lock only engine.commit(k1, 5, 7); let snap = RegionSnapshot::::from_raw(Arc::clone(&db), region.clone()); let mut reader = MvccReader::new(snap, None, false, IsolationLevel::Si); // Then reading the primary key should succeed assert!(reader.check_lock(&Key::from_raw(k1), 6.into()).is_ok()); // Reading secondary keys should still fail assert!(reader.check_lock(&Key::from_raw(k2), 6.into()).is_err()); assert!(reader .check_lock(&Key::from_raw(k2), TimeStamp::max()) .is_err()); // Pessimistic locks engine.acquire_pessimistic_lock(Key::from_raw(k4), k4, 9, 9); let snap = RegionSnapshot::::from_raw(db, region); let mut reader = MvccReader::new(snap, None, false, IsolationLevel::Si); // Pessimistic locks don't block any read operation assert!(reader.check_lock(&Key::from_raw(k4), 10.into()).is_ok()); } #[test] fn test_scan_locks() { let path = tempfile::Builder::new() .prefix("_test_storage_mvcc_reader_scan_locks") .tempdir() .unwrap(); let path = path.path().to_str().unwrap(); let region = make_region(1, vec![], vec![]); let db = open_db(path, true); let mut engine = RegionEngine::new(&db, ®ion); // Put some locks to the db. engine.prewrite( Mutation::Put((Key::from_raw(b"k1"), b"v1".to_vec())), b"k1", 5, ); engine.prewrite( Mutation::Put((Key::from_raw(b"k2"), b"v2".to_vec())), b"k1", 10, ); engine.prewrite(Mutation::Delete(Key::from_raw(b"k3")), b"k1", 10); engine.prewrite(Mutation::Lock(Key::from_raw(b"k3\x00")), b"k1", 10); engine.prewrite(Mutation::Delete(Key::from_raw(b"k4")), b"k1", 12); engine.acquire_pessimistic_lock(Key::from_raw(b"k5"), b"k1", 10, 12); engine.acquire_pessimistic_lock(Key::from_raw(b"k6"), b"k1", 12, 12); // All locks whose ts <= 10. let visible_locks: Vec<_> = vec![ // key, lock_type, short_value, ts, for_update_ts ( b"k1".to_vec(), LockType::Put, Some(b"v1".to_vec()), 5.into(), TimeStamp::zero(), ), ( b"k2".to_vec(), LockType::Put, Some(b"v2".to_vec()), 10.into(), TimeStamp::zero(), ), ( b"k3".to_vec(), LockType::Delete, None, 10.into(), TimeStamp::zero(), ), ( b"k3\x00".to_vec(), LockType::Lock, None, 10.into(), TimeStamp::zero(), ), ( b"k5".to_vec(), LockType::Pessimistic, None, 10.into(), 12.into(), ), ] .into_iter() .map(|(k, lock_type, short_value, ts, for_update_ts)| { ( Key::from_raw(&k), Lock::new( lock_type, b"k1".to_vec(), ts, 0, short_value, for_update_ts, 0, TimeStamp::zero(), ), ) }) .collect(); // Creates a reader and scan locks, let check_scan_lock = |start_key: Option, limit, expect_res: &[_], expect_is_remain| { let snap = RegionSnapshot::::from_raw(Arc::clone(&db), region.clone()); let mut reader = MvccReader::new(snap, None, false, IsolationLevel::Si); let res = reader .scan_locks(start_key.as_ref(), |l| l.ts <= 10.into(), limit) .unwrap(); assert_eq!(res.0, expect_res); assert_eq!(res.1, expect_is_remain); }; check_scan_lock(None, 6, &visible_locks, false); check_scan_lock(None, 5, &visible_locks, true); check_scan_lock(None, 4, &visible_locks[0..4], true); check_scan_lock(Some(Key::from_raw(b"k2")), 3, &visible_locks[1..4], true); check_scan_lock( Some(Key::from_raw(b"k3\x00")), 1, &visible_locks[3..4], true, ); check_scan_lock( Some(Key::from_raw(b"k3\x00")), 10, &visible_locks[3..], false, ); // limit = 0 means unlimited. check_scan_lock(None, 0, &visible_locks, false); } #[test] fn test_get() { let path = tempfile::Builder::new() .prefix("_test_storage_mvcc_reader_get_write") .tempdir() .unwrap(); let path = path.path().to_str().unwrap(); let region = make_region(1, vec![], vec![]); let db = open_db(path, true); let mut engine = RegionEngine::new(&db, ®ion); let (k, short_value, long_value) = ( b"k", b"v", "v".repeat(txn_types::SHORT_VALUE_MAX_LEN + 1).into_bytes(), ); let m = Mutation::Put((Key::from_raw(k), short_value.to_vec())); engine.prewrite(m, k, 1); engine.commit(k, 1, 2); engine.rollback(k, 5); engine.lock(k, 6, 7); engine.delete(k, 8, 9); let m = Mutation::Put((Key::from_raw(k), long_value.to_vec())); engine.prewrite(m, k, 10); let snap = RegionSnapshot::::from_raw(db.clone(), region.clone()); let mut reader = MvccReader::new(snap, None, false, IsolationLevel::Si); let key = Key::from_raw(k); for &skip_lock_check in &[false, true] { assert_eq!( reader.get(&key, 2.into(), skip_lock_check).unwrap(), Some(short_value.to_vec()) ); assert_eq!( reader.get(&key, 5.into(), skip_lock_check).unwrap(), Some(short_value.to_vec()) ); assert_eq!( reader.get(&key, 7.into(), skip_lock_check).unwrap(), Some(short_value.to_vec()) ); assert_eq!(reader.get(&key, 9.into(), skip_lock_check).unwrap(), None); } assert!(reader.get(&key, 11.into(), false).is_err()); assert_eq!(reader.get(&key, 9.into(), true).unwrap(), None); // Commit the long value engine.commit(k, 10, 11); let snap = RegionSnapshot::::from_raw(db, region); let mut reader = MvccReader::new(snap, None, false, IsolationLevel::Si); for &skip_lock_check in &[false, true] { assert_eq!( reader.get(&key, 11.into(), skip_lock_check).unwrap(), Some(long_value.to_vec()) ); } } #[test] fn test_get_old_value() { struct Case { expected: OldValue, // (write_record, put_ts) // all data to write to the engine // current write_cursor will be on the last record in `written` // which also means prev_write is `Write` in the record written: Vec<(Write, TimeStamp)>, } let cases = vec![ // prev_write is Rollback, and there exists a more previous valid write Case { expected: OldValue::Value { short_value: None, start_ts: TimeStamp::new(4), }, written: vec![ ( Write::new(WriteType::Put, TimeStamp::new(4), None), TimeStamp::new(6), ), ( Write::new(WriteType::Rollback, TimeStamp::new(5), None), TimeStamp::new(7), ), ], }, Case { expected: OldValue::Value { short_value: Some(b"v".to_vec()), start_ts: TimeStamp::new(4), }, written: vec![ ( Write::new(WriteType::Put, TimeStamp::new(4), Some(b"v".to_vec())), TimeStamp::new(6), ), ( Write::new(WriteType::Rollback, TimeStamp::new(5), None), TimeStamp::new(7), ), ], }, // prev_write is Rollback, and there isn't a more previous valid write Case { expected: OldValue::None, written: vec![( Write::new(WriteType::Rollback, TimeStamp::new(5), None), TimeStamp::new(6), )], }, // prev_write is Lock, and there exists a more previous valid write Case { expected: OldValue::Value { short_value: None, start_ts: TimeStamp::new(3), }, written: vec![ ( Write::new(WriteType::Put, TimeStamp::new(3), None), TimeStamp::new(6), ), ( Write::new(WriteType::Lock, TimeStamp::new(5), None), TimeStamp::new(7), ), ], }, // prev_write is Lock, and there isn't a more previous valid write Case { expected: OldValue::None, written: vec![( Write::new(WriteType::Lock, TimeStamp::new(5), None), TimeStamp::new(6), )], }, // prev_write is Delete, and there exists a more previous valid write Case { expected: OldValue::None, written: vec![ ( Write::new(WriteType::Put, TimeStamp::new(3), None), TimeStamp::new(6), ), ( Write::new(WriteType::Delete, TimeStamp::new(7), None), TimeStamp::new(8), ), ], }, ]; for (i, case) in cases.into_iter().enumerate() { let engine = TestEngineBuilder::new().build().unwrap(); let snapshot = engine.snapshot(&Context::default()).unwrap(); let mut txn = MvccTxn::new(snapshot, TimeStamp::new(10), true); for (write_record, put_ts) in case.written.iter() { txn.put_write( Key::from_raw(b"a"), *put_ts, write_record.as_ref().to_bytes(), ); } engine .write( &Context::default(), WriteData::from_modifies(txn.into_modifies()), ) .unwrap(); let snapshot = engine.snapshot(Default::default()).unwrap(); let mut reader = MvccReader::new(snapshot, None, true, IsolationLevel::Si); if !case.written.is_empty() { let prev_write = reader .seek_write(&Key::from_raw(b"a"), case.written.last().unwrap().1) .unwrap() .map(|w| w.1); let prev_write_loaded = true; let result = reader .get_old_value( &Key::from_raw(b"a"), TimeStamp::new(25), prev_write_loaded, prev_write, ) .unwrap(); assert_eq!(result, case.expected, "case #{}", i); } } // Must return Oldvalue::None when prev_write_loaded is true and prev_write is None. let engine = TestEngineBuilder::new().build().unwrap(); let snapshot = engine.snapshot(Default::default()).unwrap(); let mut reader = MvccReader::new(snapshot, None, true, IsolationLevel::Si); let prev_write_loaded = true; let prev_write = None; let result = reader .get_old_value( &Key::from_raw(b"a"), TimeStamp::new(25), prev_write_loaded, prev_write, ) .unwrap(); assert_eq!(result, OldValue::None); } }