// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0. // Modifications Copyright 2022 PingCAP, Inc. // #[PerformanceCriticalPath] use std::{marker::PhantomData, num::NonZeroU64, ops::Bound, sync::Arc}; use engine_traits::{CF_DEFAULT, CF_LOCK, CF_WRITE}; use kvproto::kvrpcpb::Context; use raftstore::store::TxnExt; use tikv_kv::{SEEK_BOUND, SnapshotExt}; use tikv_util::Either; use txn_types::{ Key, Lock, LockOrSharedLocks, OldValue, TimeStamp, Value, Write, WriteRef, WriteType, }; use crate::storage::{ config::ExtraCfScanMode, kv::{ Cursor, CursorBuilder, Error as KvError, ScanMode, Snapshot as EngineSnapshot, Statistics, }, mvcc::{ ErrorInner, Result, default_not_found_error, reader::{ OverlappedWrite, TxnCommitRecord, cloud_reader::{CloudReader, InMemoryLockContext}, }, }, }; /// Read from an MVCC snapshot, i.e., a logical view of the database at a /// specific timestamp (the start_ts). /// /// This represents the view of the database from a single transaction. /// /// Confusingly, there are two meanings of the word 'snapshot' here. In the name /// of the struct, 'snapshot' means an mvcc snapshot. In the type parameter /// bound (of `S`), 'snapshot' means a view of the underlying storage engine at /// a given point in time. This latter snapshot will include values for keys at /// multiple timestamps. pub struct SnapshotReader { pub start_ts: TimeStamp, pub cloud_reader: CloudReader, /// In-memory pessimistic locks are stored on the raft peer extension. /// /// CloudReader owns only the kvengine snapshot, so SnapshotReader keeps the /// peer extension metadata needed to check the in-memory lock table before /// reading persisted LOCK_CF records. txn_ext: Option>, term: u64, version: u64, _phantom: PhantomData, } impl SnapshotReader { pub fn new(start_ts: TimeStamp, snapshot: S, fill_cache: bool) -> Self { let txn_ext = snapshot.ext().get_txn_ext().cloned(); let term = snapshot.ext().get_term().map_or(0, NonZeroU64::get); let snap = snapshot.get_kvengine_snap().unwrap(); let cloud_reader = CloudReader::new(snap.clone(), fill_cache); SnapshotReader { start_ts, cloud_reader, txn_ext, term, version: 0, _phantom: PhantomData, } } pub fn new_with_ctx(start_ts: TimeStamp, snapshot: S, ctx: &Context) -> Self { Self::new_with_ctx_and_scan_mode(start_ts, snapshot, ctx, ExtraCfScanMode::default()) } /// Creates a reader with an explicitly owned EXTRA_CF scan mode. /// /// Storage commands that perform EXTRA_CF conflict checks must use this /// constructor so concurrent Storage instances cannot affect each other. pub fn new_with_ctx_and_scan_mode( start_ts: TimeStamp, snapshot: S, ctx: &Context, extra_cf_scan_mode: ExtraCfScanMode, ) -> Self { let txn_ext = snapshot.ext().get_txn_ext().cloned(); let term = snapshot.ext().get_term().map_or(0, NonZeroU64::get); let snap = snapshot.get_kvengine_snap().unwrap(); let cloud_reader = CloudReader::new_with_scan_mode( snap.clone(), !ctx.get_not_fill_cache(), extra_cf_scan_mode, ); SnapshotReader { start_ts, cloud_reader, txn_ext, term, version: ctx.get_region_epoch().get_version(), _phantom: PhantomData, } } #[maybe_async::both] #[inline(always)] pub async fn get_txn_commit_record(&mut self, key: &Key) -> Result { self.cloud_reader .get_txn_commit_record(key, self.start_ts) .await } #[inline(always)] pub fn load_lock(&mut self, key: &Key) -> Result> { if let Some(pessimistic_lock) = self.load_in_memory_pessimistic_lock(key)? { return Ok(Some(Either::Left(pessimistic_lock))); } self.cloud_reader.load_lock(key) } /// Loads an in-memory pessimistic lock for the given key. Returns a /// StaleCommand / EpochNotMatch error if the snapshot is from a different /// leader term or region epoch than the current lock table. fn load_in_memory_pessimistic_lock(&self, key: &Key) -> Result> { self.txn_ext .as_ref() .and_then(|txn_ext| { let locks = txn_ext.pessimistic_locks.read(); if let Err(e) = locks.check_term_version_status(self.term, self.version) { return Some(Err(KvError::from(e).into())); } locks.get(key).map(|(lock, _deleted)| { // For write commands that are executed in serial, it should be impossible // to read a deleted lock. // For read commands in the scheduler, it should read the lock marked deleted // because the lock is not actually deleted from the underlying storage. Ok(lock.to_lock()) }) }) .transpose() } #[maybe_async::both] #[inline(always)] pub async fn key_exist(&mut self, key: &Key, ts: TimeStamp) -> Result { Ok(self .cloud_reader .get_write(key, ts, Some(self.start_ts)) .await? .is_some()) } #[maybe_async::both] #[inline(always)] pub async fn get(&mut self, key: &Key, ts: TimeStamp) -> Result> { self.cloud_reader.get(key, ts, Some(self.start_ts)).await } #[maybe_async::both] #[inline(always)] pub async fn get_write(&mut self, key: &Key, ts: TimeStamp) -> Result> { self.cloud_reader .get_write(key, ts, Some(self.start_ts)) .await } #[maybe_async::both] #[inline(always)] pub async fn get_write_with_commit_ts( &mut self, key: &Key, ts: TimeStamp, ) -> Result> { match self.cloud_reader.seek_write(key, ts).await? { Some((commit_ts, write)) => Ok(match write.write_type { WriteType::Put => Some((write, commit_ts)), WriteType::Delete => None, _ => { return Err(ErrorInner::Other(box_err!( "unexpected write type in WRITE CF: key={:?}, write={:?}", key, write )) .into()); } }), None => Ok(None), } } #[maybe_async::both] #[inline(always)] pub async fn seek_write( &mut self, key: &Key, ts: TimeStamp, ) -> Result> { self.cloud_reader.seek_write(key, ts).await } #[inline(always)] pub fn load_data(&mut self, _key: &Key, write: Write) -> Result { Ok(write.short_value.unwrap()) } #[maybe_async::both] #[inline(always)] pub async fn get_old_value( &mut self, key: &Key, ts: TimeStamp, _prev_write_loaded: bool, prev_write: Option, ) -> Result { self.cloud_reader.get_old_value(key, ts, prev_write).await } #[inline(always)] pub fn take_statistics(&mut self) -> Statistics { self.cloud_reader.take_statistics() } pub fn scan_values_in_default(&mut self, _key: &Key) -> Result> { Ok(vec![]) } // TODO: support async. pub fn seek_ts(&mut self, ts: TimeStamp) -> Result> { self.cloud_reader.seek_ts(ts) } /// Scans locks from both storage and in-memory pessimistic locks. /// /// This method scans locks from two sources: /// 1. Storage locks (persisted in LOCK_CF) /// 2. In-memory pessimistic locks (for pipelined pessimistic locking) /// /// The results are merged and sorted by key, with in-memory locks taking /// precedence over storage locks for the same key. pub fn scan_locks( &mut self, start: Option<&Key>, end: Option<&Key>, filter: F, limit: usize, ) -> Result<(Vec<(Key, LockOrSharedLocks)>, bool)> where F: Fn(&Lock) -> bool, { let in_memory_ctx = self.txn_ext.as_ref().map(|txn_ext| InMemoryLockContext { txn_ext: txn_ext.clone(), term: self.term, version: self.version, }); self.cloud_reader .scan_locks(in_memory_ctx, start, end, filter, limit) } } pub struct MvccReader { snapshot: S, pub statistics: Statistics, // cursors are used for speeding up scans. data_cursor: Option>, lock_cursor: Option>, write_cursor: Option>, lower_bound: Option, upper_bound: Option, hint_min_ts: Option>, /// None means following operations are performed on a single user key, /// i.e., different versions of the same key. It can use prefix seek to /// speed up reads from the write-cf. scan_mode: Option, // Records the current key for prefix seek. Will Reset the write cursor when switching to // another key. current_key: Option, fill_cache: bool, // The term and the epoch version when the snapshot is created. They will be zero // if the two properties are not available. term: u64, version: u64, } impl MvccReader { pub fn new(snapshot: S, scan_mode: Option, fill_cache: bool) -> Self { Self { snapshot, statistics: Statistics::default(), data_cursor: None, lock_cursor: None, write_cursor: None, lower_bound: None, upper_bound: None, hint_min_ts: None, scan_mode, current_key: None, fill_cache, term: 0, version: 0, } } pub fn new_with_ctx(snapshot: S, scan_mode: Option, ctx: &Context) -> Self { Self { snapshot, statistics: Statistics::default(), data_cursor: None, lock_cursor: None, write_cursor: None, lower_bound: None, upper_bound: None, hint_min_ts: None, scan_mode, current_key: None, fill_cache: !ctx.get_not_fill_cache(), term: ctx.get_term(), version: ctx.get_region_epoch().get_version(), } } /// get the value of a user key with the given `start_ts`. pub fn get_value(&mut self, key: &Key, start_ts: TimeStamp) -> Result> { if self.scan_mode.is_some() { self.create_data_cursor()?; } let k = key.clone().append_ts(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)? }; if let Some(value) = val.as_ref() { self.statistics.data.processed_keys += 1; self.statistics.processed_size += key.len() + value.len(); self.statistics.total_size += key.len() + value.len(); } else { self.statistics.total_size += key.len(); } Ok(val) } /// load the value associated with `key` and pointed by `write` pub fn load_data(&mut self, key: &Key, write: Write) -> Result { assert_eq!(write.write_type, WriteType::Put); if let Some(val) = write.short_value { return Ok(val); } let start_ts = write.start_ts; match self.get_value(key, start_ts)? { Some(val) => Ok(val), None => Err(default_not_found_error( key.clone().append_ts(start_ts).into_encoded(), "get", )), } } pub fn load_lock(&mut self, key: &Key) -> Result> { if let Some(pessimistic_lock) = self.load_in_memory_pessimistic_lock(key)? { return Ok(Some(Either::Left(pessimistic_lock))); } if self.scan_mode.is_some() { self.create_lock_cursor()?; } let res = if let Some(ref mut cursor) = self.lock_cursor { match cursor.get(key, &mut self.statistics.lock)? { Some(v) => Some(txn_types::parse_lock(v)?), None => None, } } else { self.statistics.lock.get += 1; match self.snapshot.get_cf(CF_LOCK, key)? { Some(v) => Some(txn_types::parse_lock(&v)?), None => None, } }; Ok(res) } // Kept for forward compatibility: production callers go through SnapshotReader, // but a direct MvccReader caller carrying term/version must still get // StaleCommand / EpochNotMatch on a mismatched lock table. fn load_in_memory_pessimistic_lock(&self, key: &Key) -> Result> { self.snapshot .ext() .get_txn_ext() .and_then(|txn_ext| { let locks = txn_ext.pessimistic_locks.read(); if let Err(e) = locks.check_term_version_status(self.term, self.version) { return Some(Err(KvError::from(e).into())); } locks.get(key).map(|(lock, _)| { // For write commands that are executed in serial, it should be impossible // to read a deleted lock. // For read commands in the scheduler, it should read the lock marked deleted // because the lock is not actually deleted from the underlying storage. Ok(lock.to_lock()) }) }) .transpose() } 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, } } /// Return: /// (commit_ts, write_record) of the write record for `key` committed /// before or equal to`ts` Post Condition: /// leave the write_cursor at the first record which key is less or equal /// to the `ts` encoded version of `key` pub fn seek_write(&mut self, key: &Key, ts: TimeStamp) -> Result> { // Get the cursor for write record // // When it switches to another key in prefix seek mode, creates a new cursor for // it because the current position of the cursor is seldom around `key`. if self.scan_mode.is_none() && (self.current_key.as_ref() != Some(key)) { self.current_key = Some(key.clone()); self.write_cursor.take(); } self.create_write_cursor()?; let cursor = self.write_cursor.as_mut().unwrap(); // find a `ts` encoded key which is less than the `ts` encoded version of the // `key` let found = cursor.near_seek(&key.clone().append_ts(ts), &mut self.statistics.write)?; if !found { return Ok(None); } let write_key = cursor.key(&mut self.statistics.write); let commit_ts = Key::decode_ts_from(write_key)?; // check whether the found written_key's "real key" part equals the `key` we // want to find if !Key::is_user_key_eq(write_key, key.as_encoded()) { return Ok(None); } // parse out the write record let write = WriteRef::parse(cursor.value(&mut self.statistics.write))?.to_owned(); Ok(Some((commit_ts, write))) } /// Gets the value of the specified key's latest version before specified /// `ts`. /// /// It tries to ensure the write record's `gc_fence`'s ts, if any, greater /// than specified `gc_fence_limit`. Pass `None` to `gc_fence_limit` to /// skip the check. The caller must guarantee that there's no other `PUT` or /// `DELETE` versions whose `commit_ts` is between the found version and /// the provided `gc_fence_limit` (`gc_fence_limit` is inclusive). /// /// For transactional reads, the `gc_fence_limit` must be provided to ensure /// the result is correct. Generally, it should be the read_ts of the /// current transaction, which might be different from the `ts` passed to /// this function. /// /// Note that this function does not check for locks on `key`. #[allow(dead_code)] fn get( &mut self, key: &Key, ts: TimeStamp, gc_fence_limit: Option, ) -> Result> { Ok(match self.get_write(key, ts, gc_fence_limit)? { Some(write) => Some(self.load_data(key, write)?), None => None, }) } /// Gets the write record of the specified key's latest version before /// specified `ts`. It tries to ensure the write record's `gc_fence`'s /// ts, if any, greater than specified `gc_fence_limit`. Pass `None` to /// `gc_fence_limit` to skip the check. The caller must guarantee that /// there's no other `PUT` or `DELETE` versions whose `commit_ts` is between /// the found version and the provided `gc_fence_limit` (`gc_fence_limit` is /// inclusive). /// For transactional reads, the `gc_fence_limit` must be provided to ensure /// the result is correct. Generally, it should be the read_ts of the /// current transaction, which might be different from the `ts` passed to /// this function. pub fn get_write( &mut self, key: &Key, ts: TimeStamp, gc_fence_limit: Option, ) -> Result> { Ok(self .get_write_with_commit_ts(key, ts, gc_fence_limit)? .map(|(w, _)| w)) } /// Gets the write record of the specified key's latest version before /// specified `ts`, and additionally the write record's `commit_ts`, if any. /// /// See also [`MvccReader::get_write`]. pub fn get_write_with_commit_ts( &mut self, key: &Key, mut ts: TimeStamp, gc_fence_limit: Option, ) -> Result> { let mut seek_res = self.seek_write(key, ts)?; loop { match seek_res { Some((commit_ts, write)) => { if let Some(limit) = gc_fence_limit { if !write.as_ref().check_gc_fence_as_latest_version(limit) { return Ok(None); } } match write.write_type { WriteType::Put => { return Ok(Some((write, commit_ts))); } WriteType::Delete => { return Ok(None); } WriteType::Lock | WriteType::Rollback => { if write.versions_to_last_change > 0 && write.last_change_ts.is_zero() { return Ok(None); } if write.versions_to_last_change < SEEK_BOUND { ts = commit_ts.prev(); } else { let commit_ts = write.last_change_ts; let key_with_ts = key.clone().append_ts(commit_ts); let Some(value) = self.snapshot.get_cf(CF_WRITE, &key_with_ts)? else { return Ok(None); }; self.statistics.write.get += 1; let write = WriteRef::parse(&value)?.to_owned(); assert!( write.write_type == WriteType::Put || write.write_type == WriteType::Delete, "Write record pointed by last_change_ts {} should be Put or Delete, but got {:?}", commit_ts, write.write_type, ); seek_res = Some((commit_ts, write)); continue; } } } } None => return Ok(None), } seek_res = self.seek_write(key, ts)?; } } #[allow(dead_code)] fn get_txn_commit_record(&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(); let mut gc_fence = TimeStamp::from(0); while let Some((commit_ts, write)) = self.seek_write(key, seek_ts)? { if write.start_ts == start_ts { return Ok(TxnCommitRecord::SingleRecord { commit_ts, write }); } if commit_ts == start_ts { if write.has_overlapped_rollback { return Ok(TxnCommitRecord::OverlappedRollback { commit_ts }); } return Ok(TxnCommitRecord::None { overlapped_write: Some(OverlappedWrite { write, gc_fence }), }); } if write.write_type == WriteType::Put || write.write_type == WriteType::Delete { gc_fence = commit_ts; } if commit_ts < start_ts { break; } seek_ts = commit_ts.prev(); } Ok(TxnCommitRecord::None { overlapped_write: None, }) } fn create_data_cursor(&mut self) -> Result<()> { if self.data_cursor.is_none() { let cursor = CursorBuilder::new(&self.snapshot, CF_DEFAULT) .fill_cache(self.fill_cache) .scan_mode(self.get_scan_mode(true)) .range(self.lower_bound.clone(), self.upper_bound.clone()) .build()?; self.data_cursor = Some(cursor); } Ok(()) } fn create_write_cursor(&mut self) -> Result<()> { if self.write_cursor.is_none() { let cursor = CursorBuilder::new(&self.snapshot, CF_WRITE) .fill_cache(self.fill_cache) // Only use prefix seek in non-scan mode. .prefix_seek(self.scan_mode.is_none()) .scan_mode(self.get_scan_mode(true)) .range(self.lower_bound.clone(), self.upper_bound.clone()) // `hint_min_ts` filters data by the `commit_ts`. .hint_min_ts(self.hint_min_ts) .build()?; self.write_cursor = Some(cursor); } Ok(()) } fn create_lock_cursor(&mut self) -> Result<()> { if self.lock_cursor.is_none() { let cursor = CursorBuilder::new(&self.snapshot, CF_LOCK) .fill_cache(self.fill_cache) .scan_mode(self.get_scan_mode(true)) .range(self.lower_bound.clone(), self.upper_bound.clone()) .build()?; self.lock_cursor = Some(cursor); } 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 in the key range /// [start, end). At most `limit` locks will be returned. If `limit` is /// set to `0`, it means unlimited. /// /// The return type is `(locks, has_remain)`. `has_remain` indicates whether /// there MAY be remaining locks that can be scanned. pub fn scan_locks( &mut self, start: Option<&Key>, end: Option<&Key>, filter: F, limit: usize, ) -> Result<(Vec<(Key, LockOrSharedLocks)>, bool)> where F: Fn(&Lock) -> bool, { self.create_lock_cursor()?; let cursor = self.lock_cursor.as_mut().unwrap(); let ok = match start { Some(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); let mut lock_count = 0usize; let mut has_remain = false; while cursor.valid()? { let key = Key::from_encoded_slice(cursor.key(&mut self.statistics.lock)); if let Some(end) = end { if key >= *end { has_remain = false; break; } } let mut lock_or_shared_locks = txn_types::parse_lock(cursor.value(&mut self.statistics.lock))?; match &mut lock_or_shared_locks { Either::Left(lock) => { if filter(lock) { lock_count += 1; locks.push((key, lock_or_shared_locks)); } } Either::Right(shared_locks) => { shared_locks.filter_shared_locks(|lock| filter(lock))?; if !shared_locks.is_empty() { if limit > 0 { let remaining = limit.saturating_sub(lock_count); if remaining == 0 { has_remain = true; break; } if shared_locks.len() > remaining { shared_locks.truncate_to(remaining); has_remain = true; } } lock_count += shared_locks.len(); locks.push((key, lock_or_shared_locks)); } } } if limit > 0 && lock_count >= limit { has_remain = true; break; } cursor.next(&mut self.statistics.lock); } self.statistics.lock.processed_keys += lock_count; Ok((locks, has_remain)) } /// Scan the writes to get all the latest user keys. The return type is: /// * `(Vec, has_remain)`. /// - `key` is the encoded user key without `commit_ts`. /// - `has_remain` indicates whether there MAY be remaining user keys that /// can be scanned. /// /// This function is mainly used by /// `txn::commands::FlashbackToVersionReadPhase` /// and `txn::commands::FlashbackToVersion` to achieve the MVCC /// overwriting. pub fn scan_latest_user_keys( &mut self, start: Option<&Key>, end: Option<&Key>, filter: F, limit: usize, ) -> Result<(Vec, bool)> where F: Fn(&Key /* user key */, TimeStamp /* latest `commit_ts` */) -> bool, { self.create_write_cursor()?; let cursor = self.write_cursor.as_mut().unwrap(); let ok = match start { Some(x) => cursor.seek(x, &mut self.statistics.write)?, None => cursor.seek_to_first(&mut self.statistics.write), }; if !ok { return Ok((vec![], false)); } let mut cur_user_key = None; let mut keys = Vec::with_capacity(limit); let mut has_remain = false; while cursor.valid()? { let key = Key::from_encoded_slice(cursor.key(&mut self.statistics.write)); if let Some(end) = end { if key >= *end { has_remain = false; break; } } let commit_ts = key.decode_ts()?; let user_key = key.truncate_ts()?; // To make sure we only check each unique user key once and the filter returns // true. let is_same_user_key = cur_user_key.as_ref() == Some(&user_key); if !is_same_user_key { cur_user_key = Some(user_key.clone()); } if is_same_user_key || !filter(&user_key, commit_ts) { cursor.next(&mut self.statistics.write); continue; } keys.push(user_key.clone()); if limit > 0 && keys.len() == limit { has_remain = true; break; } // Seek once to skip all the writes of the same user key. cursor.near_seek( &user_key.append_ts(TimeStamp::zero()), &mut self.statistics.write, )?; } self.statistics.write.processed_keys += keys.len(); resource_metering::record_read_keys(keys.len() as u32); Ok((keys, has_remain)) } pub fn scan_keys( &mut self, mut start: Option, limit: usize, ) -> Result<(Vec, Option)> { let mut cursor = CursorBuilder::new(&self.snapshot, CF_WRITE) .fill_cache(self.fill_cache) .scan_mode(self.get_scan_mode(false)) .build()?; 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 { self.statistics.write.processed_keys += keys.len(); return Ok((keys, None)); } if keys.len() >= limit { self.statistics.write.processed_keys += keys.len(); resource_metering::record_read_keys(keys.len() as u32); 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) } /// 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 #[allow(dead_code)] 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 { if !prev_write .as_ref() .check_gc_fence_as_latest_version(start_ts) { return Ok(OldValue::None); } 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(match prev_write.short_value { Some(value) => OldValue::Value { value }, None => OldValue::ValueTimeStamp { 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(start_ts))? { Some(write) => match write.short_value { Some(value) => OldValue::Value { value }, None => OldValue::ValueTimeStamp { start_ts: write.start_ts, }, }, None => OldValue::None, }) } pub fn set_range(&mut self, lower: Option, upper: Option) { self.lower_bound = lower; self.upper_bound = upper; } pub fn set_hint_min_ts(&mut self, ts_bound: Option>) { self.hint_min_ts = ts_bound; } } #[cfg(test)] pub mod tests { use concurrency_manager::ConcurrencyManager; use engine_traits::{CF_DEFAULT, CF_WRITE}; use kvproto::{ kvrpcpb::{AssertionLevel, Context, PrewriteRequestPessimisticAction::*}, metapb::{Peer, Region}, }; use tikv_kv::{BTreeEngine, BTreeEngineSnapshot}; use txn_types::{LockType, Mutation}; use super::*; use crate::storage::{ Engine, TestEngineBuilder, kv::{Modify, WriteData}, mvcc::{MvccReader, MvccTxn, tests::write}, txn::{ CommitKind, TransactionKind, TransactionProperties, acquire_pessimistic_lock, cleanup, commit, prewrite, }, }; pub struct RegionEngine { db: BTreeEngine, } impl RegionEngine { pub fn new(db: &BTreeEngine, _region: &Region) -> RegionEngine { RegionEngine { db: db.clone() } } pub fn snapshot(&self) -> BTreeEngineSnapshot { let mut db = self.db.clone(); db.snapshot(Default::default()).unwrap() } pub fn lock( &mut self, pk: &[u8], start_ts: impl Into, commit_ts: impl Into, ) { let start_ts = start_ts.into(); let m = Mutation::make_lock(Key::from_raw(pk)); self.prewrite(m, pk, start_ts); self.commit(pk, start_ts, commit_ts); } pub fn delete( &mut self, pk: &[u8], start_ts: impl Into, commit_ts: impl Into, ) { let start_ts = start_ts.into(); let m = Mutation::make_delete(Key::from_raw(pk)); self.prewrite(m, pk, start_ts); self.commit(pk, start_ts, commit_ts); } pub fn txn_props( start_ts: TimeStamp, primary: &[u8], pessimistic: bool, ) -> TransactionProperties<'_> { let kind = if pessimistic { TransactionKind::Pessimistic(TimeStamp::default()) } else { TransactionKind::Optimistic(false) }; TransactionProperties { start_ts, kind, commit_kind: CommitKind::TwoPc, primary, txn_size: 0, lock_ttl: 0, min_commit_ts: TimeStamp::default(), need_old_value: false, is_retry_request: false, assertion_level: AssertionLevel::Off, txn_source: 0, } } pub fn prewrite(&mut self, m: Mutation, pk: &[u8], start_ts: impl Into) { let snap = self.snapshot(); let start_ts = start_ts.into(); let cm = ConcurrencyManager::new(start_ts); let mut txn = MvccTxn::new(start_ts, cm); let mut reader = SnapshotReader::new(start_ts, snap, true); prewrite( &mut txn, &mut reader, &Self::txn_props(start_ts, pk, false), m, &None, SkipPessimisticCheck, ) .unwrap(); self.write(txn.into_modifies()); } pub fn prewrite_pessimistic_lock( &mut self, m: Mutation, pk: &[u8], start_ts: impl Into, ) { let snap = self.snapshot(); let start_ts = start_ts.into(); let cm = ConcurrencyManager::new(start_ts); let mut txn = MvccTxn::new(start_ts, cm); let mut reader = SnapshotReader::new(start_ts, snap, true); prewrite( &mut txn, &mut reader, &Self::txn_props(start_ts, pk, true), m, &None, DoPessimisticCheck, ) .unwrap(); self.write(txn.into_modifies()); } pub fn acquire_pessimistic_lock( &mut self, k: Key, pk: &[u8], start_ts: impl Into, for_update_ts: impl Into, ) { let snap = self.snapshot(); let for_update_ts = for_update_ts.into(); let cm = ConcurrencyManager::new(for_update_ts); let start_ts = start_ts.into(); let mut txn = MvccTxn::new(start_ts, cm); let mut reader = SnapshotReader::new(start_ts, snap, true); acquire_pessimistic_lock( &mut txn, &mut reader, k, pk, false, 0, for_update_ts, false, false, TimeStamp::zero(), true, false, false, false, ) .unwrap(); self.write(txn.into_modifies()); } pub fn commit( &mut self, pk: &[u8], start_ts: impl Into, commit_ts: impl Into, ) { let snap = self.snapshot(); let start_ts = start_ts.into(); let cm = ConcurrencyManager::new(start_ts); let mut txn = MvccTxn::new(start_ts, cm); let mut reader = SnapshotReader::new(start_ts, snap, true); commit(&mut txn, &mut reader, Key::from_raw(pk), commit_ts.into()).unwrap(); self.write(txn.into_modifies()); } pub fn rollback(&mut self, pk: &[u8], start_ts: impl Into) { let snap = self.snapshot(); let start_ts = start_ts.into(); let cm = ConcurrencyManager::new(start_ts); let mut txn = MvccTxn::new(start_ts, cm); let mut reader = SnapshotReader::new(start_ts, snap, true); cleanup( &mut txn, &mut reader, Key::from_raw(pk), TimeStamp::zero(), true, ) .unwrap(); self.write(txn.into_modifies()); } pub fn write(&mut self, modifies: Vec) { if modifies.is_empty() { return; } self.db .write(&Context::default(), WriteData::from_modifies(modifies)) .unwrap(); } } pub fn open_db(_path: &str, _with_properties: bool) -> BTreeEngine { TestEngineBuilder::new().build().unwrap() } struct RegionSnapshot; impl RegionSnapshot { fn from_raw(mut db: BTreeEngine, _region: Region) -> BTreeEngineSnapshot { db.snapshot(Default::default()).unwrap() } } pub 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 } #[test] fn test_get_txn_commit_record() { let path = tempfile::Builder::new() .prefix("_test_storage_mvcc_reader_get_txn_commit_record") .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::make_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::make_put(Key::from_raw(k), v.to_vec()); engine.prewrite(m, k, 25); engine.commit(k, 25, 30); let m = Mutation::make_put(Key::from_raw(k), v.to_vec()); engine.prewrite(m, k, 35); engine.commit(k, 35, 40); let m = Mutation::make_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); // 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, 40_35 PUT, 30_25 PUT, 20_20 Rollback, // 10_1 PUT, 5_5 Rollback]. let key = Key::from_raw(k); let overlapped_write = reader .get_txn_commit_record(&key, 55.into()) .unwrap() .unwrap_none(); assert!(overlapped_write.is_none()); let (commit_ts, write_type) = reader .get_txn_commit_record(&key, 45.into()) .unwrap() .unwrap_single_record(); assert_eq!(commit_ts, 50.into()); assert_eq!(write_type, WriteType::Put); let (commit_ts, write_type) = reader .get_txn_commit_record(&key, 35.into()) .unwrap() .unwrap_single_record(); assert_eq!(commit_ts, 40.into()); assert_eq!(write_type, WriteType::Put); let (commit_ts, write_type) = reader .get_txn_commit_record(&key, 25.into()) .unwrap() .unwrap_single_record(); assert_eq!(commit_ts, 30.into()); assert_eq!(write_type, WriteType::Put); let (commit_ts, write_type) = reader .get_txn_commit_record(&key, 20.into()) .unwrap() .unwrap_single_record(); assert_eq!(commit_ts, 20.into()); assert_eq!(write_type, WriteType::Rollback); let (commit_ts, write_type) = reader .get_txn_commit_record(&key, 1.into()) .unwrap() .unwrap_single_record(); assert_eq!(commit_ts, 10.into()); assert_eq!(write_type, WriteType::Put); let (commit_ts, write_type) = reader .get_txn_commit_record(&key, 5.into()) .unwrap() .unwrap_single_record(); assert_eq!(commit_ts, 5.into()); assert_eq!(write_type, WriteType::Rollback); let seek_old = reader.statistics.write.seek; let next_old = reader.statistics.write.next; assert!( !reader .get_txn_commit_record(&key, 30.into()) .unwrap() .exist() ); let seek_new = reader.statistics.write.seek; let next_new = reader.statistics.write.next; // `get_txn_commit_record(&key, 30)` stopped at `30_25 PUT`. assert_eq!(seek_new - seek_old, 1); assert_eq!(next_new - next_old, 2); } #[test] fn test_get_txn_commit_record_of_pessimistic_txn() { let path = tempfile::Builder::new() .prefix("_test_storage_mvcc_reader_get_txn_commit_record_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::make_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); let (commit_ts, write_type) = reader .get_txn_commit_record(&key, 2.into()) .unwrap() .unwrap_single_record(); assert_eq!(commit_ts, 3.into()); assert_eq!(write_type, WriteType::Put); let (commit_ts, write_type) = reader .get_txn_commit_record(&key, 1.into()) .unwrap() .unwrap_single_record(); 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::make_put(Key::from_raw(k), v.to_vec()); engine.prewrite(m.clone(), k, 1); engine.commit(k, 1, 5); engine.write(vec![ Modify::Put( CF_WRITE, Key::from_raw(k).append_ts(TimeStamp::new(3)), vec![b'R', 3], ), Modify::Put( CF_WRITE, Key::from_raw(k).append_ts(TimeStamp::new(7)), vec![b'R', 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::make_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(db.clone(), region.clone()); let mut reader = MvccReader::new(snap, None, false); 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())) ); assert_eq!(reader.statistics.write.seek, 1); assert_eq!(reader.statistics.write.next, 0); 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())) ); assert_eq!(reader.statistics.write.seek, 1); assert_eq!(reader.statistics.write.next, 0); let (commit_ts, write) = reader.seek_write(&k, 20.into()).unwrap().unwrap(); assert_eq!(commit_ts, 20.into()); assert_eq!(write.write_type, WriteType::Lock); assert_eq!(write.start_ts, 10.into()); assert_eq!(reader.statistics.write.seek, 1); assert_eq!(reader.statistics.write.next, 1); 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())) ); assert_eq!(reader.statistics.write.seek, 1); assert_eq!(reader.statistics.write.next, 2); 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)); assert_eq!(reader.statistics.write.seek, 1); assert_eq!(reader.statistics.write.next, 5); 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)); assert_eq!(reader.statistics.write.seek, 1); assert_eq!(reader.statistics.write.next, 6); assert_eq!(reader.statistics.write.prev, 3); 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_eq!(reader.statistics.write.seek, 1); assert_eq!(reader.statistics.write.next, 7); assert_eq!(reader.statistics.write.prev, 3); assert!(reader.seek_write(&k, 2.into()).unwrap().is_none()); assert_eq!(reader.statistics.write.seek, 1); assert_eq!(reader.statistics.write.next, 9); assert_eq!(reader.statistics.write.prev, 3); // Test seek_write should not see the next key. let (k2, v2) = (b"k2", b"v2"); let m2 = Mutation::make_put(Key::from_raw(k2), v2.to_vec()); engine.prewrite(m2, k2, 1); engine.commit(k2, 1, 2); let snap = RegionSnapshot::from_raw(db.clone(), region); let mut reader = MvccReader::new(snap, None, false); 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_eq!(reader.statistics.write.seek, 1); assert_eq!(reader.statistics.write.next, 0); // Should seek for another key. assert!(reader.seek_write(&k, 2.into()).unwrap().is_none()); assert_eq!(reader.statistics.write.seek, 2); assert_eq!(reader.statistics.write.next, 0); } #[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::make_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::make_put(Key::from_raw(k), v.to_vec()); engine.prewrite(m, k, 12); engine.commit(k, 12, 14); let m = Mutation::make_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::make_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::make_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::make_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); // 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(), None).unwrap().is_none()); let write = reader.get_write(&key, 2.into(), None).unwrap().unwrap(); assert_eq!(write.write_type, WriteType::Put); assert_eq!(write.start_ts, 1.into()); let write = reader.get_write(&key, 5.into(), None).unwrap().unwrap(); assert_eq!(write.write_type, WriteType::Put); assert_eq!(write.start_ts, 1.into()); let write = reader.get_write(&key, 7.into(), None).unwrap().unwrap(); assert_eq!(write.write_type, WriteType::Put); assert_eq!(write.start_ts, 1.into()); assert!(reader.get_write(&key, 9.into(), None).unwrap().is_none()); let write = reader.get_write(&key, 14.into(), None).unwrap().unwrap(); assert_eq!(write.write_type, WriteType::Put); assert_eq!(write.start_ts, 12.into()); let write = reader.get_write(&key, 16.into(), None).unwrap().unwrap(); assert_eq!(write.write_type, WriteType::Put); assert_eq!(write.start_ts, 12.into()); let write = reader.get_write(&key, 20.into(), None).unwrap().unwrap(); assert_eq!(write.write_type, WriteType::Put); assert_eq!(write.start_ts, 18.into()); let write = reader.get_write(&key, 24.into(), None).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(), None) .unwrap() .is_none() ); } #[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::make_put(Key::from_raw(b"k1"), b"v1".to_vec()), b"k1", 5, ); engine.prewrite( Mutation::make_put(Key::from_raw(b"k2"), b"v2".to_vec()), b"k1", 10, ); engine.prewrite(Mutation::make_delete(Key::from_raw(b"k3")), b"k1", 10); engine.prewrite(Mutation::make_lock(Key::from_raw(b"k3\x00")), b"k1", 10); engine.prewrite(Mutation::make_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(), ) .set_last_change( TimeStamp::zero(), (lock_type == LockType::Pessimistic) as u64, ), ) }) .collect(); // Creates a reader and scan locks, let check_scan_lock = |start_key: Option, end_key: Option, limit, expect_res: &[_], expect_is_remain| { let snap = RegionSnapshot::from_raw(db.clone(), region.clone()); let mut reader = MvccReader::new(snap, None, false); let res = reader .scan_locks( start_key.as_ref(), end_key.as_ref(), |l| l.ts <= 10.into(), limit, ) .unwrap(); let locks: Vec<_> = res .0 .into_iter() .map(|(key, lock)| { ( key, lock.left().expect("expected non-shared locks in this test"), ) }) .collect(); assert_eq!(locks, expect_res); assert_eq!(res.1, expect_is_remain); }; check_scan_lock(None, None, 6, &visible_locks, false); check_scan_lock(None, None, 5, &visible_locks, true); check_scan_lock(None, None, 4, &visible_locks[0..4], true); check_scan_lock( Some(Key::from_raw(b"k2")), None, 3, &visible_locks[1..4], true, ); check_scan_lock( Some(Key::from_raw(b"k3\x00")), None, 1, &visible_locks[3..4], true, ); check_scan_lock( Some(Key::from_raw(b"k3\x00")), None, 10, &visible_locks[3..], false, ); // limit = 0 means unlimited. check_scan_lock(None, None, 0, &visible_locks, false); // Test scanning with limited end_key check_scan_lock( None, Some(Key::from_raw(b"k3")), 0, &visible_locks[..2], false, ); check_scan_lock( None, Some(Key::from_raw(b"k3\x00")), 0, &visible_locks[..3], false, ); check_scan_lock( None, Some(Key::from_raw(b"k3\x00")), 3, &visible_locks[..3], true, ); check_scan_lock( None, Some(Key::from_raw(b"k3\x00")), 2, &visible_locks[..2], true, ); } #[test] fn test_scan_latest_user_keys() { let path = tempfile::Builder::new() .prefix("_test_storage_mvcc_reader_scan_latest_user_keys") .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); // Prewrite and rollback k0. engine.prewrite( Mutation::make_put(Key::from_raw(b"k0"), b"v0@1".to_vec()), b"k0", 1, ); engine.rollback(b"k0", 1); // Prewrite and commit k0 with a large ts. engine.prewrite( Mutation::make_put(Key::from_raw(b"k0"), b"v0@999".to_vec()), b"k0", 999, ); engine.commit(b"k0", 999, 1000); // Prewrite and commit k1. engine.prewrite( Mutation::make_put(Key::from_raw(b"k1"), b"v1@1".to_vec()), b"k1", 1, ); engine.commit(b"k1", 1, 2); engine.prewrite( Mutation::make_put(Key::from_raw(b"k1"), b"v1@3".to_vec()), b"k1", 3, ); engine.commit(b"k1", 3, 4); // Uncommitted prewrite k1. engine.prewrite( Mutation::make_put(Key::from_raw(b"k1"), b"v1@5".to_vec()), b"k1", 5, ); // Prewrite and commit k2. engine.prewrite( Mutation::make_put(Key::from_raw(b"k2"), b"v2@1".to_vec()), b"k2", 1, ); engine.commit(b"k2", 1, 2); engine.prewrite( Mutation::make_put(Key::from_raw(b"k2"), b"v2@3".to_vec()), b"k2", 3, ); engine.commit(b"k2", 3, 4); // Prewrite and commit k3. engine.prewrite( Mutation::make_put(Key::from_raw(b"k3"), b"v3@5".to_vec()), b"k3", 5, ); engine.commit(b"k3", 5, 6); // Prewrite and rollback k3. engine.prewrite( Mutation::make_put(Key::from_raw(b"k3"), b"v3@7".to_vec()), b"k3", 7, ); engine.rollback(b"k3", 7); // Prewrite and commit k3. engine.prewrite( Mutation::make_put(Key::from_raw(b"k3"), b"v3@8".to_vec()), b"k3", 8, ); engine.commit(b"k3", 8, 9); // Prewrite and rollback k4. engine.prewrite( Mutation::make_put(Key::from_raw(b"k4"), b"v4@1".to_vec()), b"k4", 10, ); engine.rollback(b"k4", 10); // Current MVCC keys in `CF_WRITE` should be: // PUT k0 -> v0@999 // ROLLBACK k0 -> v0@1 // PUT k1 -> v1@3 // PUT k1 -> v1@1 // PUT k2 -> v2@3 // PUT k2 -> v2@1 // PUT k3 -> v3@8 // ROLLBACK k3 -> v3@7 // PUT k3 -> v3@5 // ROLLBACK k4 -> v4@1 struct Case { start_key: Option, end_key: Option, limit: usize, expect_res: Vec, expect_is_remain: bool, } let cases = [ // Test the limit. Case { start_key: None, end_key: None, limit: 1, expect_res: vec![Key::from_raw(b"k0")], expect_is_remain: true, }, Case { start_key: None, end_key: None, limit: 6, expect_res: vec![ Key::from_raw(b"k0"), Key::from_raw(b"k1"), Key::from_raw(b"k2"), Key::from_raw(b"k3"), Key::from_raw(b"k4"), ], expect_is_remain: false, }, // Test the start/end key. Case { start_key: Some(Key::from_raw(b"k2")), end_key: None, limit: 4, expect_res: vec![ Key::from_raw(b"k2"), Key::from_raw(b"k3"), Key::from_raw(b"k4"), ], expect_is_remain: false, }, Case { start_key: None, end_key: Some(Key::from_raw(b"k3")), limit: 4, expect_res: vec![ Key::from_raw(b"k0"), Key::from_raw(b"k1"), Key::from_raw(b"k2"), ], expect_is_remain: false, }, ]; for (idx, case) in cases.iter().enumerate() { let snap = RegionSnapshot::from_raw(db.clone(), region.clone()); let mut reader = MvccReader::new(snap, Some(ScanMode::Forward), false); let res = reader .scan_latest_user_keys( case.start_key.as_ref(), case.end_key.as_ref(), |_, _| true, case.limit, ) .unwrap(); assert_eq!(res.0, case.expect_res, "case #{}", idx); assert_eq!(res.1, case.expect_is_remain, "case #{}", idx); } } #[test] fn test_load_data() { let path = tempfile::Builder::new() .prefix("_test_storage_mvcc_reader_load_data") .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(), ); struct Case { expected: Result, // modifies to put into the engine modifies: Vec, // these are used to construct the mvcc reader scan_mode: Option, key: Key, write: Write, } let cases = vec![ Case { // write has short_value expected: Ok(short_value.to_vec()), modifies: vec![Modify::Put( CF_DEFAULT, Key::from_raw(k).append_ts(TimeStamp::new(1)), vec![], )], scan_mode: None, key: Key::from_raw(k), write: Write::new( WriteType::Put, TimeStamp::new(1), Some(short_value.to_vec()), ), }, Case { // write has no short_value, the reader has a cursor, got something expected: Ok(long_value.to_vec()), modifies: vec![Modify::Put( CF_DEFAULT, Key::from_raw(k).append_ts(TimeStamp::new(2)), long_value.to_vec(), )], scan_mode: Some(ScanMode::Forward), key: Key::from_raw(k), write: Write::new(WriteType::Put, TimeStamp::new(2), None), }, Case { // write has no short_value, the reader has a cursor, got nothing expected: Err(default_not_found_error( Key::from_raw(k).append_ts(TimeStamp::new(3)).into_encoded(), "get", )), modifies: vec![Modify::Put( CF_WRITE, Key::from_raw(k).append_ts(TimeStamp::new(1)), Write::new(WriteType::Put, TimeStamp::new(1), None) .as_ref() .to_bytes(), )], scan_mode: Some(ScanMode::Forward), key: Key::from_raw(k), write: Write::new(WriteType::Put, TimeStamp::new(3), None), }, Case { // write has no short_value, the reader has no cursor, got something expected: Ok(long_value.to_vec()), modifies: vec![Modify::Put( CF_DEFAULT, Key::from_raw(k).append_ts(TimeStamp::new(4)), long_value.to_vec(), )], scan_mode: None, key: Key::from_raw(k), write: Write::new(WriteType::Put, TimeStamp::new(4), None), }, Case { // write has no short_value, the reader has no cursor, got nothing expected: Err(default_not_found_error( Key::from_raw(k).append_ts(TimeStamp::new(5)).into_encoded(), "get", )), modifies: vec![], scan_mode: None, key: Key::from_raw(k), write: Write::new(WriteType::Put, TimeStamp::new(5), None), }, ]; for case in cases { engine.write(case.modifies); let snap = RegionSnapshot::from_raw(db.clone(), region.clone()); let mut reader = MvccReader::new(snap, case.scan_mode, false); let result = reader.load_data(&case.key, case.write.clone()); assert_eq!(format!("{:?}", result), format!("{:?}", case.expected)); if let Ok(expected) = case.expected { if expected == long_value.to_vec() { let result = reader .get_value(&case.key, case.write.start_ts) .unwrap() .unwrap(); assert_eq!(format!("{:?}", result), format!("{:?}", expected)); } } } } #[test] fn test_get() { let path = tempfile::Builder::new() .prefix("_test_storage_mvcc_reader_get") .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, long_value) = ( b"k", "v".repeat(txn_types::SHORT_VALUE_MAX_LEN + 1).into_bytes(), ); struct Case { expected: Result>, // modifies to put into the engine modifies: Vec, // arguments to do the function call key: Key, ts: TimeStamp, gc_fence_limit: Option, } let cases = vec![ Case { // no write for `key` at `ts` exists expected: Ok(None), modifies: vec![Modify::Delete( CF_DEFAULT, Key::from_raw(k).append_ts(TimeStamp::new(1)), )], key: Key::from_raw(k), ts: TimeStamp::new(1), gc_fence_limit: None, }, Case { // some write for `key` at `ts` exists, load data return Err // todo: "some write for `key` at `ts` exists" should be checked by `test_get_write` // "load data return Err" is checked by test_load_data expected: Err(default_not_found_error( Key::from_raw(k).append_ts(TimeStamp::new(2)).into_encoded(), "get", )), modifies: vec![Modify::Put( CF_WRITE, Key::from_raw(k).append_ts(TimeStamp::new(2)), Write::new(WriteType::Put, TimeStamp::new(2), None) .as_ref() .to_bytes(), )], key: Key::from_raw(k), ts: TimeStamp::new(3), gc_fence_limit: None, }, Case { // some write for `key` at `ts` exists, load data success // todo: "some write for `key` at `ts` exists" should be checked by `test_get_write` // "load data success" is checked by test_load_data expected: Ok(Some(long_value.to_vec())), modifies: vec![ Modify::Put( CF_WRITE, Key::from_raw(k).append_ts(TimeStamp::new(4)), Write::new(WriteType::Put, TimeStamp::new(4), None) .as_ref() .to_bytes(), ), Modify::Put( CF_DEFAULT, Key::from_raw(k).append_ts(TimeStamp::new(4)), long_value, ), ], key: Key::from_raw(k), ts: TimeStamp::new(5), gc_fence_limit: None, }, ]; for case in cases { engine.write(case.modifies); let snap = RegionSnapshot::from_raw(db.clone(), region.clone()); let mut reader = MvccReader::new(snap, None, false); let result = reader.get(&case.key, case.ts, case.gc_fence_limit); assert_eq!(format!("{:?}", result), format!("{:?}", case.expected)); } } #[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 None Case { expected: OldValue::None, written: vec![], }, // prev_write is Rollback, and there exists a more previous valid write Case { expected: OldValue::ValueTimeStamp { 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 { value: b"v".to_vec(), }, 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::ValueTimeStamp { 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 not Rollback or Lock, check_gc_fence_as_latest_version is true Case { expected: OldValue::ValueTimeStamp { start_ts: TimeStamp::new(7), }, written: vec![( Write::new(WriteType::Put, TimeStamp::new(7), None) .set_overlapped_rollback(true, Some(27.into())), TimeStamp::new(5), )], }, // prev_write is not Rollback or Lock, check_gc_fence_as_latest_version is false Case { expected: OldValue::None, written: vec![( Write::new(WriteType::Put, TimeStamp::new(4), None) .set_overlapped_rollback(true, Some(3.into())), TimeStamp::new(5), )], }, // prev_write is Delete, check_gc_fence_as_latest_version is true 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), ), ], }, // prev_write is Delete, check_gc_fence_as_latest_version is false 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) .set_overlapped_rollback(true, Some(6.into())), TimeStamp::new(8), ), ], }, ]; for (i, case) in cases.into_iter().enumerate() { let mut engine = TestEngineBuilder::new().build().unwrap(); let cm = ConcurrencyManager::new(42.into()); let mut txn = MvccTxn::new(TimeStamp::new(10), cm.clone()); 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(), ); } write(&engine, &Context::default(), txn.into_modifies()); let snapshot = engine.snapshot(Default::default()).unwrap(); let mut reader = MvccReader::new(snapshot, None, true); 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 mut engine = TestEngineBuilder::new().build().unwrap(); let snapshot = engine.snapshot(Default::default()).unwrap(); let mut reader = MvccReader::new(snapshot, None, true); 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); } }