Merge branch 'master' into make-extra-states-writebatch

Signed-off-by: Jay Lee <BusyJayLee@gmail.com>

README.md

@@ -151,7 +151,7 @@ A third-party security auditing was performed by Cure53. See the full report [he
 
 To report a security vulnerability, please send an email to [TiKV-security](mailto:tikv-security@lists.cncf.io) group.
 
-See [Security](./security/SECURITY.md) for the process and policy followed by the TiKV project.
+See [Security](SECURITY.md) for the process and policy followed by the TiKV project.
 
 ## Communication
 

security/SECURITY.md → SECURITY.md

@@ -18,6 +18,8 @@ The following are the versions that we support for security updates
 
 | Version | Supported          |
 | ------- | ------------------ |
+| 6.x   | :white_check_mark: |
+| 5.x   | :white_check_mark: |
 | 4.x   | :white_check_mark: |
 | 3.x   | :white_check_mark: |
 | 2.x   | :white_check_mark: |
@@ -94,4 +96,4 @@ IvCICV7zG1cyuM/Z2Y7/TJ+upvahP46nM3s3G15b8FYuTSmRN1Kp9+mBt2BHqOy1
 ulx+VF4Lf9n3ydf593Nha9bMJ/rnSp01
 =XbYK
 -----END PGP PUBLIC KEY BLOCK-----
-```
+```

components/engine_panic/src/raft_engine.rs

@@ -47,11 +47,23 @@ impl RaftEngineReadOnly for PanicEngine {
         panic!()
     }
 
-    fn get_region_state(&self, raft_group_id: u64) -> Result<Option<RegionLocalState>> {
+    fn get_region_state(
+        &self,
+        raft_group_id: u64,
+        apply_index: u64,
+    ) -> Result<Option<RegionLocalState>> {
+        panic!()
+    }
+
+    fn get_apply_state(
+        &self,
+        raft_group_id: u64,
+        apply_index: u64,
+    ) -> Result<Option<RaftApplyState>> {
         panic!()
     }
 
-    fn get_apply_state(&self, raft_group_id: u64) -> Result<Option<RaftApplyState>> {
+    fn get_flushed_index(&self, raft_group_id: u64, cf: &str) -> Result<Option<u64>> {
         panic!()
     }
 
@@ -186,11 +198,31 @@ impl RaftLogBatch for PanicWriteBatch {
         panic!()
     }
 
-    fn put_region_state(&mut self, raft_group_id: u64, state: &RegionLocalState) -> Result<()> {
+    fn put_region_state(
+        &mut self,
+        raft_group_id: u64,
+        apply_index: u64,
+        state: &RegionLocalState,
+    ) -> Result<()> {
+        panic!()
+    }
+
+    fn put_apply_state(
+        &mut self,
+        raft_group_id: u64,
+        apply_index: u64,
+        state: &RaftApplyState,
+    ) -> Result<()> {
         panic!()
     }
 
-    fn put_apply_state(&mut self, raft_group_id: u64, state: &RaftApplyState) -> Result<()> {
+    fn put_flushed_index(
+        &mut self,
+        raft_group_id: u64,
+        cf: &str,
+        tablet_index: u64,
+        apply_index: u64,
+    ) -> Result<()> {
         panic!()
     }
 

components/engine_rocks/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::{
+    FlushState, IterOptions, Iterable, KvEngine, Peekable, ReadOptions, Result, SyncMutable,
+};
 use rocksdb::{DBIterator, Writable, DB};
 
 use crate::{
@@ -24,6 +26,7 @@ use crate::{
 pub struct RocksEngine {
     db: Arc<DB>,
     support_multi_batch_write: bool,
+    flush_state: Option<Arc<FlushState>>,
 }
 
 impl RocksEngine {
@@ -35,6 +38,7 @@ impl RocksEngine {
         RocksEngine {
             db: db.clone(),
             support_multi_batch_write: db.get_db_options().is_enable_multi_batch_write(),
+            flush_state: None,
         }
     }
 
@@ -49,6 +53,14 @@ impl RocksEngine {
     pub fn support_multi_batch_write(&self) -> bool {
         self.support_multi_batch_write
     }
+
+    pub fn set_flush_state(&mut self, flush_state: Arc<FlushState>) {
+        self.flush_state = Some(flush_state);
+    }
+
+    pub fn flush_state(&self) -> Option<Arc<FlushState>> {
+        self.flush_state.clone()
+    }
 }
 
 impl KvEngine for RocksEngine {

components/engine_rocks/src/event_listener.rs

@@ -1,10 +1,11 @@
 // Copyright 2020 TiKV Project Authors. Licensed under Apache-2.0.
 
+use engine_traits::{PersistenceListener, RaftEngine};
 use file_system::{get_io_type, set_io_type, IoType};
 use regex::Regex;
 use rocksdb::{
-    CompactionJobInfo, DBBackgroundErrorReason, FlushJobInfo, IngestionInfo, MutableStatus,
-    SubcompactionJobInfo, WriteStallInfo,
+    CompactionJobInfo, DBBackgroundErrorReason, FlushJobInfo, IngestionInfo, MemTableInfo,
+    MutableStatus, SubcompactionJobInfo, WriteStallInfo,
 };
 use tikv_util::{error, metrics::CRITICAL_ERROR, set_panic_mark, warn, worker::Scheduler};
 
@@ -178,6 +179,26 @@ fn resolve_sst_filename_from_err(err: &str) -> Option<String> {
     Some(filename)
 }
 
+pub struct RocksPersistenceListener<ER>(PersistenceListener<ER>);
+
+impl<ER> RocksPersistenceListener<ER> {
+    pub fn new(listener: PersistenceListener<ER>) -> RocksPersistenceListener<ER> {
+        RocksPersistenceListener(listener)
+    }
+}
+
+impl<ER: RaftEngine> rocksdb::EventListener for RocksPersistenceListener<ER> {
+    fn on_memtable_sealed(&self, info: &MemTableInfo) {
+        self.0
+            .on_memtable_sealed(info.cf_name().to_string(), info.first_seqno());
+    }
+
+    fn on_flush_completed(&self, job: &FlushJobInfo) {
+        self.0
+            .on_flush_completed(job.cf_name(), job.smallest_seqno());
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

components/engine_rocks/src/raft_engine.rs

@@ -144,14 +144,26 @@ impl RaftEngineReadOnly for RocksEngine {
         self.get_msg_cf(CF_DEFAULT, keys::PREPARE_BOOTSTRAP_KEY)
     }
 
-    fn get_region_state(&self, raft_group_id: u64) -> Result<Option<RegionLocalState>> {
-        let key = keys::region_state_key(raft_group_id);
-        self.get_msg_cf(CF_DEFAULT, &key)
+    // Following methods are used by raftstore v2 only, which always use raft log
+    // engine.
+    fn get_region_state(
+        &self,
+        _raft_group_id: u64,
+        _apply_index: u64,
+    ) -> Result<Option<RegionLocalState>> {
+        panic!()
     }
 
-    fn get_apply_state(&self, raft_group_id: u64) -> Result<Option<RaftApplyState>> {
-        let key = keys::apply_state_key(raft_group_id);
-        self.get_msg_cf(CF_DEFAULT, &key)
+    fn get_apply_state(
+        &self,
+        _raft_group_id: u64,
+        _apply_index: u64,
+    ) -> Result<Option<RaftApplyState>> {
+        panic!()
+    }
+
+    fn get_flushed_index(&self, _raft_group_id: u64, _cf: &str) -> Result<Option<u64>> {
+        panic!()
     }
 
     fn get_recover_state(&self) -> Result<Option<StoreRecoverState>> {
@@ -405,12 +417,34 @@ impl RaftLogBatch for RocksWriteBatchVec {
         self.delete(keys::PREPARE_BOOTSTRAP_KEY)
     }
 
-    fn put_region_state(&mut self, raft_group_id: u64, state: &RegionLocalState) -> Result<()> {
-        self.put_msg(&keys::region_state_key(raft_group_id), state)
+    // Following methods are used by raftstore v2 only, which always use raft log
+    // engine.
+    fn put_region_state(
+        &mut self,
+        _raft_group_id: u64,
+        _apply_index: u64,
+        _state: &RegionLocalState,
+    ) -> Result<()> {
+        panic!()
+    }
+
+    fn put_apply_state(
+        &mut self,
+        _raft_group_id: u64,
+        _apply_index: u64,
+        _state: &RaftApplyState,
+    ) -> Result<()> {
+        panic!()
     }
 
-    fn put_apply_state(&mut self, raft_group_id: u64, state: &RaftApplyState) -> Result<()> {
-        self.put_msg(&keys::apply_state_key(raft_group_id), state)
+    fn put_flushed_index(
+        &mut self,
+        _raft_group_id: u64,
+        _cf: &str,
+        _tablet_index: u64,
+        _apply_index: u64,
+    ) -> Result<()> {
+        panic!()
     }
 
     fn put_recover_state(&mut self, state: &StoreRecoverState) -> Result<()> {