In-memory Engine: implement read flow part 1 (tikv#16163)

ref tikv#16141

implement in-memory engine read flow -- part 1

Signed-off-by: SpadeA-Tang <[email protected]>

Author: SpadeA-Tang

Cargo.lock

@@ -996,7 +996,7 @@ mod test {
 
         let (items, size) = super::with_record_read_throughput(|| {
             let mut items = vec![];
-            let snap = engine.snapshot();
+            let snap = engine.snapshot(None);
             snap.scan(CF_DEFAULT, b"", b"", false, |k, v| {
                 items.push((k.to_owned(), v.to_owned()));
                 Ok(true)

components/backup-stream/src/utils.rs

@@ -121,7 +121,8 @@ impl<E: KvEngine> CmdObserver<E> for CdcObserver {
         // Create a snapshot here for preventing the old value was GC-ed.
         // TODO: only need it after enabling old value, may add a flag to indicate
         // whether to get it.
-        let snapshot = RegionSnapshot::from_snapshot(Arc::new(engine.snapshot()), Arc::new(region));
+        let snapshot =
+            RegionSnapshot::from_snapshot(Arc::new(engine.snapshot(None)), Arc::new(region));
         let get_old_value = move |key,
                                   query_ts,
                                   old_value_cache: &mut OldValueCache,

components/cdc/src/observer.rs

@@ -308,7 +308,7 @@ mod tests {
         value: Option<Value>,
     ) -> Statistics {
         let key = key.clone().append_ts(ts.into());
-        let snapshot = Arc::new(kv_engine.snapshot());
+        let snapshot = Arc::new(kv_engine.snapshot(None));
         let mut cursor = new_write_cursor_on_key(&snapshot, &key);
         let load_default = Either::Left(&snapshot);
         let mut stats = Statistics::default();
@@ -527,7 +527,7 @@ mod tests {
             must_commit(&mut engine, &key, 200, 201);
         }
 
-        let snapshot = Arc::new(kv_engine.snapshot());
+        let snapshot = Arc::new(kv_engine.snapshot(None));
         let mut cursor = new_old_value_cursor(&snapshot, CF_WRITE);
         let mut default_cursor = new_old_value_cursor(&snapshot, CF_DEFAULT);
         let mut load_default = |use_default_cursor: bool| {
@@ -598,7 +598,7 @@ mod tests {
         }
 
         let key = format!("zkey-{:0>3}", 0).into_bytes();
-        let snapshot = Arc::new(kv_engine.snapshot());
+        let snapshot = Arc::new(kv_engine.snapshot(None));
         let perf_instant = ReadPerfInstant::new();
         let value = get_old_value(
             &snapshot,

components/cdc/src/old_value.rs

@@ -1,7 +1,7 @@
 // Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
 
 use engine_traits::{
-    IterOptions, Iterable, Iterator, KvEngine, Peekable, ReadOptions, Result, SyncMutable,
+    IterOptions, Iterable, Iterator, KvEngine, Peekable, ReadOptions, Result, SnapCtx, SyncMutable,
     WriteOptions,
 };
 
@@ -13,7 +13,7 @@ pub struct PanicEngine;
 impl KvEngine for PanicEngine {
     type Snapshot = PanicSnapshot;
 
-    fn snapshot(&self) -> Self::Snapshot {
+    fn snapshot(&self, _: Option<SnapCtx>) -> Self::Snapshot {
         panic!()
     }
     fn sync(&self) -> Result<()> {

components/engine_panic/src/engine.rs

@@ -2,7 +2,9 @@
 
 use std::{any::Any, sync::Arc};
 
-use engine_traits::{IterOptions, Iterable, KvEngine, Peekable, ReadOptions, Result, SyncMutable};
+use engine_traits::{
+    IterOptions, Iterable, KvEngine, Peekable, ReadOptions, Result, SnapCtx, SyncMutable,
+};
 use rocksdb::{DBIterator, Writable, DB};
 
 use crate::{
@@ -182,7 +184,7 @@ impl RocksEngine {
 impl KvEngine for RocksEngine {
     type Snapshot = RocksSnapshot;
 
-    fn snapshot(&self) -> RocksSnapshot {
+    fn snapshot(&self, _: Option<SnapCtx>) -> RocksSnapshot {
         RocksSnapshot::new(self.db.clone())
     }
 
@@ -292,7 +294,7 @@ mod tests {
         engine.put_msg(key, &r).unwrap();
         engine.put_msg_cf(cf, key, &r).unwrap();
 
-        let snap = engine.snapshot();
+        let snap = engine.snapshot(None);
 
         let mut r1: Region = engine.get_msg(key).unwrap().unwrap();
         assert_eq!(r, r1);

components/engine_rocks/src/engine.rs

@@ -39,7 +39,10 @@ pub trait KvEngine:
     type Snapshot: Snapshot;
 
     /// Create a snapshot
-    fn snapshot(&self) -> Self::Snapshot;
+    ///
+    /// SnapCtx will only be used by some type of trait implementors (ex:
+    /// HybridEngine)
+    fn snapshot(&self, snap_ctx: Option<SnapCtx>) -> Self::Snapshot;
 
     /// Syncs any writes to disk
     fn sync(&self) -> Result<()>;
@@ -78,3 +81,8 @@ pub trait KvEngine:
     #[cfg(feature = "testexport")]
     fn inner_refcount(&self) -> usize;
 }
+
+pub struct SnapCtx {
+    pub region_id: u64,
+    pub read_ts: u64,
+}

components/engine_traits/src/engine.rs

@@ -149,6 +149,8 @@ pub enum Error {
     EntriesUnavailable,
     #[error("The entries of region is compacted")]
     EntriesCompacted,
+    #[error("Iterator of RegionCacheSnapshot is only supported with boundary set")]
+    BoundaryNotSet,
 }
 
 pub type Result<T> = result::Result<T, Error>;
@@ -165,6 +167,7 @@ impl ErrorCodeExt for Error {
             Error::Other(_) => error_code::UNKNOWN,
             Error::EntriesUnavailable => error_code::engine::DATALOSS,
             Error::EntriesCompacted => error_code::engine::DATACOMPACTED,
+            Error::BoundaryNotSet => error_code::engine::BOUNDARY_NOT_SET,
         }
     }
 }

components/engine_traits/src/errors.rs

@@ -41,7 +41,9 @@ fn iter_empty_engine() {
 #[test]
 fn iter_empty_snapshot() {
     let db = default_engine();
-    iter_empty(&db.engine, |e| e.snapshot().iterator(CF_DEFAULT).unwrap());
+    iter_empty(&db.engine, |e| {
+        e.snapshot(None).iterator(CF_DEFAULT).unwrap()
+    });
 }
 
 fn iter_forward<E, I, IF>(e: &E, i: IF)
@@ -99,7 +101,9 @@ fn iter_forward_engine() {
 #[test]
 fn iter_forward_snapshot() {
     let db = default_engine();
-    iter_forward(&db.engine, |e| e.snapshot().iterator(CF_DEFAULT).unwrap());
+    iter_forward(&db.engine, |e| {
+        e.snapshot(None).iterator(CF_DEFAULT).unwrap()
+    });
 }
 
 fn iter_reverse<E, I, IF>(e: &E, i: IF)
@@ -157,7 +161,9 @@ fn iter_reverse_engine() {
 #[test]
 fn iter_reverse_snapshot() {
     let db = default_engine();
-    iter_reverse(&db.engine, |e| e.snapshot().iterator(CF_DEFAULT).unwrap());
+    iter_reverse(&db.engine, |e| {
+        e.snapshot(None).iterator(CF_DEFAULT).unwrap()
+    });
 }
 
 fn seek_to_key_then_forward<E, I, IF>(e: &E, i: IF)
@@ -198,7 +204,9 @@ fn seek_to_key_then_forward_engine() {
 #[test]
 fn seek_to_key_then_forward_snapshot() {
     let db = default_engine();
-    seek_to_key_then_forward(&db.engine, |e| e.snapshot().iterator(CF_DEFAULT).unwrap());
+    seek_to_key_then_forward(&db.engine, |e| {
+        e.snapshot(None).iterator(CF_DEFAULT).unwrap()
+    });
 }
 
 fn seek_to_key_then_reverse<E, I, IF>(e: &E, i: IF)
@@ -239,7 +247,9 @@ fn seek_to_key_then_reverse_engine() {
 #[test]
 fn seek_to_key_then_reverse_snapshot() {
     let db = default_engine();
-    seek_to_key_then_reverse(&db.engine, |e| e.snapshot().iterator(CF_DEFAULT).unwrap());
+    seek_to_key_then_reverse(&db.engine, |e| {
+        e.snapshot(None).iterator(CF_DEFAULT).unwrap()
+    });
 }
 
 fn iter_forward_then_reverse<E, I, IF>(e: &E, i: IF)
@@ -300,7 +310,9 @@ fn iter_forward_then_reverse_engine() {
 #[test]
 fn iter_forward_then_reverse_snapshot() {
     let db = default_engine();
-    iter_forward_then_reverse(&db.engine, |e| e.snapshot().iterator(CF_DEFAULT).unwrap());
+    iter_forward_then_reverse(&db.engine, |e| {
+        e.snapshot(None).iterator(CF_DEFAULT).unwrap()
+    });
 }
 
 fn iter_reverse_then_forward<E, I, IF>(e: &E, i: IF)
@@ -361,7 +373,9 @@ fn iter_reverse_then_forward_engine() {
 #[test]
 fn iter_reverse_then_forward_snapshot() {
     let db = default_engine();
-    iter_reverse_then_forward(&db.engine, |e| e.snapshot().iterator(CF_DEFAULT).unwrap());
+    iter_reverse_then_forward(&db.engine, |e| {
+        e.snapshot(None).iterator(CF_DEFAULT).unwrap()
+    });
 }
 
 // When seek finds an exact key then seek_for_prev behaves just like seek
@@ -405,7 +419,9 @@ fn seek_for_prev_engine() {
 #[test]
 fn seek_for_prev_snapshot() {
     let db = default_engine();
-    seek_for_prev(&db.engine, |e| e.snapshot().iterator(CF_DEFAULT).unwrap());
+    seek_for_prev(&db.engine, |e| {
+        e.snapshot(None).iterator(CF_DEFAULT).unwrap()
+    });
 }
 
 // When Seek::Key doesn't find an exact match,
@@ -440,7 +456,9 @@ fn seek_key_miss_engine() {
 #[test]
 fn seek_key_miss_snapshot() {
     let db = default_engine();
-    seek_key_miss(&db.engine, |e| e.snapshot().iterator(CF_DEFAULT).unwrap());
+    seek_key_miss(&db.engine, |e| {
+        e.snapshot(None).iterator(CF_DEFAULT).unwrap()
+    });
 }
 
 fn seek_key_prev_miss<E, I, IF>(e: &E, i: IF)
@@ -472,5 +490,7 @@ fn seek_key_prev_miss_engine() {
 #[test]
 fn seek_key_prev_miss_snapshot() {
     let db = default_engine();
-    seek_key_prev_miss(&db.engine, |e| e.snapshot().iterator(CF_DEFAULT).unwrap());
+    seek_key_prev_miss(&db.engine, |e| {
+        e.snapshot(None).iterator(CF_DEFAULT).unwrap()
+    });
 }

components/engine_traits_tests/src/iterator.rs

@@ -12,7 +12,7 @@ fn snapshot_with_writes() {
 
     db.engine.put(b"a", b"aa").unwrap();
 
-    let snapshot = db.engine.snapshot();
+    let snapshot = db.engine.snapshot(None);
 
     assert_eq!(snapshot.get_value(b"a").unwrap().unwrap(), b"aa");
 
@@ -77,5 +77,7 @@ fn iterator_with_writes_engine() {
 #[test]
 fn iterator_with_writes_snapshot() {
     let db = default_engine();
-    iterator_with_writes(&db.engine, |e| e.snapshot().iterator(CF_DEFAULT).unwrap());
+    iterator_with_writes(&db.engine, |e| {
+        e.snapshot(None).iterator(CF_DEFAULT).unwrap()
+    });
 }