diff --git a/components/server/src/server.rs b/components/server/src/server.rs
index d49dbfdb650..bb810b8482e 100644
--- a/components/server/src/server.rs
+++ b/components/server/src/server.rs
@@ -31,8 +31,8 @@ use concurrency_manager::ConcurrencyManager;
 use encryption_export::{data_key_manager_from_config, DataKeyManager};
 use engine_rocks::{from_rocks_compression_type, get_env, FlowInfo, RocksEngine};
 use engine_traits::{
-    compaction_job::CompactionJobInfo, CFOptionsExt, ColumnFamilyOptions, Engines, MiscExt,
-    RaftEngine, CF_DEFAULT, CF_LOCK, CF_WRITE,
+    compaction_job::CompactionJobInfo, CFOptionsExt, ColumnFamilyOptions, Engines, KvEngine,
+    MiscExt, RaftEngine, CF_DEFAULT, CF_LOCK, CF_WRITE,
 };
 use error_code::ErrorCodeExt;
 use file_system::{
@@ -180,8 +180,8 @@ struct TiKVServer<ER: RaftEngine> {
     store_path: PathBuf,
     snap_mgr: Option<SnapManager>, // Will be filled in `init_servers`.
     encryption_key_manager: Option<Arc<DataKeyManager>>,
-    engines: Option<TiKVEngines<ER>>,
-    servers: Option<Servers<ER>>,
+    engines: Option<TiKVEngines<RocksEngine, ER>>,
+    servers: Option<Servers<RocksEngine, ER>>,
     region_info_accessor: RegionInfoAccessor,
     coprocessor_host: Option<CoprocessorHost<RocksEngine>>,
     to_stop: Vec<Box<dyn Stop>>,
@@ -191,24 +191,24 @@ struct TiKVServer<ER: RaftEngine> {
     background_worker: Worker,
 }
 
-struct TiKVEngines<ER: RaftEngine> {
-    engines: Engines<RocksEngine, ER>,
+struct TiKVEngines<EK: KvEngine, ER: RaftEngine> {
+    engines: Engines<EK, ER>,
     store_meta: Arc<Mutex<StoreMeta>>,
-    engine: RaftKv<RocksEngine, ServerRaftStoreRouter<RocksEngine, ER>>,
+    engine: RaftKv<EK, ServerRaftStoreRouter<EK, ER>>,
 }
 
-struct Servers<ER: RaftEngine> {
+struct Servers<EK: KvEngine, ER: RaftEngine> {
     lock_mgr: LockManager,
-    server: LocalServer<ER>,
-    node: Node<RpcClient, ER>,
+    server: LocalServer<EK, ER>,
+    node: Node<RpcClient, EK, ER>,
     importer: Arc<SSTImporter>,
     cdc_scheduler: tikv_util::worker::Scheduler<cdc::Task>,
     cdc_memory_quota: MemoryQuota,
 }
 
-type LocalServer<ER> =
-    Server<RaftRouter<RocksEngine, ER>, resolve::PdStoreAddrResolver, LocalRaftKv<ER>>;
-type LocalRaftKv<ER> = RaftKv<RocksEngine, ServerRaftStoreRouter<RocksEngine, ER>>;
+type LocalServer<EK, ER> =
+    Server<RaftRouter<EK, ER>, resolve::PdStoreAddrResolver, LocalRaftKv<EK, ER>>;
+type LocalRaftKv<EK, ER> = RaftKv<EK, ServerRaftStoreRouter<EK, ER>>;
 
 impl<ER: RaftEngine> TiKVServer<ER> {
     fn init(mut config: TiKvConfig) -> TiKVServer<ER> {
@@ -1054,8 +1054,9 @@ impl<ER: RaftEngine> TiKVServer<ER> {
         fetcher: BytesFetcher,
         engines_info: Arc<EnginesResourceInfo>,
     ) {
-        let mut engine_metrics =
-            EngineMetricsManager::new(self.engines.as_ref().unwrap().engines.clone());
+        let mut engine_metrics = EngineMetricsManager::<RocksEngine, ER>::new(
+            self.engines.as_ref().unwrap().engines.clone(),
+        );
         let mut io_metrics = IOMetricsManager::new(fetcher);
         let engines_info_clone = engines_info.clone();
         self.background_worker
@@ -1473,13 +1474,13 @@ impl<T: fmt::Display + Send + 'static> Stop for LazyWorker<T> {
     }
 }
 
-pub struct EngineMetricsManager<R: RaftEngine> {
-    engines: Engines<RocksEngine, R>,
+pub struct EngineMetricsManager<EK: KvEngine, R: RaftEngine> {
+    engines: Engines<EK, R>,
     last_reset: Instant,
 }
 
-impl<R: RaftEngine> EngineMetricsManager<R> {
-    pub fn new(engines: Engines<RocksEngine, R>) -> Self {
+impl<EK: KvEngine, R: RaftEngine> EngineMetricsManager<EK, R> {
+    pub fn new(engines: Engines<EK, R>) -> Self {
         EngineMetricsManager {
             engines,
             last_reset: Instant::now(),
@@ -1487,10 +1488,10 @@ impl<R: RaftEngine> EngineMetricsManager<R> {
     }
 
     pub fn flush(&mut self, now: Instant) {
-        self.engines.kv.flush_metrics("kv");
+        KvEngine::flush_metrics(&self.engines.kv, "kv");
         self.engines.raft.flush_metrics("raft");
         if now.saturating_duration_since(self.last_reset) >= DEFAULT_ENGINE_METRICS_RESET_INTERVAL {
-            self.engines.kv.reset_statistics();
+            KvEngine::reset_statistics(&self.engines.kv);
             self.engines.raft.reset_statistics();
             self.last_reset = now;
         }
diff --git a/components/server/src/signal_handler.rs b/components/server/src/signal_handler.rs
index 6c59cc38e55..75feaa4319b 100644
--- a/components/server/src/signal_handler.rs
+++ b/components/server/src/signal_handler.rs
@@ -4,15 +4,14 @@ pub use self::imp::wait_for_signal;
 
 #[cfg(unix)]
 mod imp {
-    use engine_rocks::RocksEngine;
-    use engine_traits::{Engines, MiscExt, RaftEngine};
+    use engine_traits::{Engines, KvEngine, MiscExt, RaftEngine};
     use libc::c_int;
     use nix::sys::signal::{SIGHUP, SIGINT, SIGTERM, SIGUSR1, SIGUSR2};
     use signal::trap::Trap;
     use tikv_util::metrics;
 
     #[allow(dead_code)]
-    pub fn wait_for_signal<ER: RaftEngine>(engines: Option<Engines<RocksEngine, ER>>) {
+    pub fn wait_for_signal(engines: Option<Engines<impl KvEngine, impl RaftEngine>>) {
         let trap = Trap::trap(&[SIGTERM, SIGINT, SIGHUP, SIGUSR1, SIGUSR2]);
         for sig in trap {
             match sig {
@@ -37,8 +36,7 @@ mod imp {
 
 #[cfg(not(unix))]
 mod imp {
-    use engine_rocks::RocksEngine;
-    use engine_traits::Engines;
+    use engine_traits::{Engines, KvEngine, RaftEngine};
 
-    pub fn wait_for_signal(_: Option<Engines<RocksEngine, RocksEngine>>) {}
+    pub fn wait_for_signal(_: Option<Engines<impl KvEngine, impl RaftEngine>>) {}
 }
diff --git a/components/test_raftstore/src/node.rs b/components/test_raftstore/src/node.rs
index c0e693ec9d1..aba4d1aa7fa 100644
--- a/components/test_raftstore/src/node.rs
+++ b/components/test_raftstore/src/node.rs
@@ -134,7 +134,7 @@ type SimulateChannelTransport = SimulateTransport<ChannelTransport>;
 pub struct NodeCluster {
     trans: ChannelTransport,
     pd_client: Arc<TestPdClient>,
-    nodes: HashMap<u64, Node<TestPdClient, RocksEngine>>,
+    nodes: HashMap<u64, Node<TestPdClient, RocksEngine, RocksEngine>>,
     snap_mgrs: HashMap<u64, SnapManager>,
     simulate_trans: HashMap<u64, SimulateChannelTransport>,
     concurrency_managers: HashMap<u64, ConcurrencyManager>,
@@ -183,7 +183,10 @@ impl NodeCluster {
         self.post_create_coprocessor_host = Some(op)
     }
 
-    pub fn get_node(&mut self, node_id: u64) -> Option<&mut Node<TestPdClient, RocksEngine>> {
+    pub fn get_node(
+        &mut self,
+        node_id: u64,
+    ) -> Option<&mut Node<TestPdClient, RocksEngine, RocksEngine>> {
         self.nodes.get_mut(&node_id)
     }
 
diff --git a/components/test_raftstore/src/server.rs b/components/test_raftstore/src/server.rs
index 7a82a0abc1b..6089b5fd47d 100644
--- a/components/test_raftstore/src/server.rs
+++ b/components/test_raftstore/src/server.rs
@@ -108,7 +108,7 @@ impl StoreAddrResolver for AddressMap {
 }
 
 struct ServerMeta {
-    node: Node<TestPdClient, RocksEngine>,
+    node: Node<TestPdClient, RocksEngine, RocksEngine>,
     server: Server<SimulateStoreTransport, PdStoreAddrResolver, SimulateEngine>,
     sim_router: SimulateStoreTransport,
     sim_trans: SimulateServerTransport,
diff --git a/src/server/node.rs b/src/server/node.rs
index 4ecfde42430..b2e53173598 100644
--- a/src/server/node.rs
+++ b/src/server/node.rs
@@ -14,8 +14,7 @@ use crate::storage::kv::FlowStatsReporter;
 use crate::storage::txn::flow_controller::FlowController;
 use crate::storage::{config::Config as StorageConfig, Storage};
 use concurrency_manager::ConcurrencyManager;
-use engine_rocks::RocksEngine;
-use engine_traits::{Engines, KvEngine, Peekable, RaftEngine};
+use engine_traits::{Engines, KvEngine, RaftEngine};
 use kvproto::metapb;
 use kvproto::raft_serverpb::StoreIdent;
 use kvproto::replication_modepb::ReplicationStatus;
@@ -64,11 +63,11 @@ where
 
 /// A wrapper for the raftstore which runs Multi-Raft.
 // TODO: we will rename another better name like RaftStore later.
-pub struct Node<C: PdClient + 'static, ER: RaftEngine> {
+pub struct Node<C: PdClient + 'static, EK: KvEngine, ER: RaftEngine> {
     cluster_id: u64,
     store: metapb::Store,
     store_cfg: Arc<VersionTrack<StoreConfig>>,
-    system: RaftBatchSystem<RocksEngine, ER>,
+    system: RaftBatchSystem<EK, ER>,
     has_started: bool,
 
     pd_client: Arc<C>,
@@ -76,20 +75,21 @@ pub struct Node<C: PdClient + 'static, ER: RaftEngine> {
     bg_worker: Worker,
 }
 
-impl<C, ER> Node<C, ER>
+impl<C, EK, ER> Node<C, EK, ER>
 where
     C: PdClient,
+    EK: KvEngine,
     ER: RaftEngine,
 {
     /// Creates a new Node.
     pub fn new(
-        system: RaftBatchSystem<RocksEngine, ER>,
+        system: RaftBatchSystem<EK, ER>,
         cfg: &ServerConfig,
         store_cfg: Arc<VersionTrack<StoreConfig>>,
         pd_client: Arc<C>,
         state: Arc<Mutex<GlobalReplicationState>>,
         bg_worker: Worker,
-    ) -> Node<C, ER> {
+    ) -> Node<C, EK, ER> {
         let mut store = metapb::Store::default();
         store.set_id(INVALID_ID);
         if cfg.advertise_addr.is_empty() {
@@ -138,7 +138,7 @@ where
         }
     }
 
-    pub fn try_bootstrap_store(&mut self, engines: Engines<RocksEngine, ER>) -> Result<()> {
+    pub fn try_bootstrap_store(&mut self, engines: Engines<EK, ER>) -> Result<()> {
         let mut store_id = self.check_store(&engines)?;
         if store_id == INVALID_ID {
             store_id = self.alloc_id()?;
@@ -158,12 +158,12 @@ where
     #[allow(clippy::too_many_arguments)]
     pub fn start<T>(
         &mut self,
-        engines: Engines<RocksEngine, ER>,
+        engines: Engines<EK, ER>,
         trans: T,
         snap_mgr: SnapManager,
-        pd_worker: LazyWorker<PdTask<RocksEngine>>,
+        pd_worker: LazyWorker<PdTask<EK>>,
         store_meta: Arc<Mutex<StoreMeta>>,
-        coprocessor_host: CoprocessorHost<RocksEngine>,
+        coprocessor_host: CoprocessorHost<EK>,
         importer: Arc<SSTImporter>,
         split_check_scheduler: Scheduler<SplitCheckTask>,
         auto_split_controller: AutoSplitController,
@@ -215,17 +215,17 @@ where
 
     /// Gets a transmission end of a channel which is used to send `Msg` to the
     /// raftstore.
-    pub fn get_router(&self) -> RaftRouter<RocksEngine, ER> {
+    pub fn get_router(&self) -> RaftRouter<EK, ER> {
         self.system.router()
     }
     /// Gets a transmission end of a channel which is used send messages to apply worker.
-    pub fn get_apply_router(&self) -> ApplyRouter<RocksEngine> {
+    pub fn get_apply_router(&self) -> ApplyRouter<EK> {
         self.system.apply_router()
     }
 
     // check store, return store id for the engine.
     // If the store is not bootstrapped, use INVALID_ID.
-    fn check_store(&self, engines: &Engines<RocksEngine, ER>) -> Result<u64> {
+    fn check_store(&self, engines: &Engines<EK, ER>) -> Result<u64> {
         let res = engines.kv.get_msg::<StoreIdent>(keys::STORE_IDENT_KEY)?;
         if res.is_none() {
             return Ok(INVALID_ID);
@@ -274,7 +274,7 @@ where
     #[doc(hidden)]
     pub fn prepare_bootstrap_cluster(
         &self,
-        engines: &Engines<RocksEngine, ER>,
+        engines: &Engines<EK, ER>,
         store_id: u64,
     ) -> Result<metapb::Region> {
         let region_id = self.alloc_id()?;
@@ -298,7 +298,7 @@ where
 
     fn check_or_prepare_bootstrap_cluster(
         &self,
-        engines: &Engines<RocksEngine, ER>,
+        engines: &Engines<EK, ER>,
         store_id: u64,
     ) -> Result<Option<metapb::Region>> {
         if let Some(first_region) = engines.kv.get_msg(keys::PREPARE_BOOTSTRAP_KEY)? {
@@ -312,7 +312,7 @@ where
 
     fn bootstrap_cluster(
         &mut self,
-        engines: &Engines<RocksEngine, ER>,
+        engines: &Engines<EK, ER>,
         first_region: metapb::Region,
     ) -> Result<()> {
         let region_id = first_region.get_id();
@@ -374,12 +374,12 @@ where
     fn start_store<T>(
         &mut self,
         store_id: u64,
-        engines: Engines<RocksEngine, ER>,
+        engines: Engines<EK, ER>,
         trans: T,
         snap_mgr: SnapManager,
-        pd_worker: LazyWorker<PdTask<RocksEngine>>,
+        pd_worker: LazyWorker<PdTask<EK>>,
         store_meta: Arc<Mutex<StoreMeta>>,
-        coprocessor_host: CoprocessorHost<RocksEngine>,
+        coprocessor_host: CoprocessorHost<EK>,
         importer: Arc<SSTImporter>,
         split_check_scheduler: Scheduler<SplitCheckTask>,
         auto_split_controller: AutoSplitController,
diff --git a/src/storage/config.rs b/src/storage/config.rs
index d1c2be72bc8..9b1ac08dde5 100644
--- a/src/storage/config.rs
+++ b/src/storage/config.rs
@@ -7,8 +7,7 @@ use crate::server::ttl::TTLCheckerTask;
 use crate::server::CONFIG_ROCKSDB_GAUGE;
 use crate::storage::txn::flow_controller::FlowController;
 use engine_rocks::raw::{Cache, LRUCacheOptions, MemoryAllocator};
-use engine_rocks::RocksEngine;
-use engine_traits::{CFNamesExt, CFOptionsExt, ColumnFamilyOptions, CF_DEFAULT};
+use engine_traits::{ColumnFamilyOptions, KvEngine, CF_DEFAULT};
 use file_system::{get_io_rate_limiter, IOPriority, IORateLimitMode, IORateLimiter, IOType};
 use libc::c_int;
 use online_config::{ConfigChange, ConfigManager, ConfigValue, OnlineConfig, Result as CfgResult};
@@ -105,20 +104,20 @@ impl Config {
     }
 }
 
-pub struct StorageConfigManger {
-    kvdb: RocksEngine,
+pub struct StorageConfigManger<EK: KvEngine> {
+    kvdb: EK,
     shared_block_cache: bool,
     ttl_checker_scheduler: Scheduler<TTLCheckerTask>,
     flow_controller: Arc<FlowController>,
 }
 
-impl StorageConfigManger {
+impl<EK: KvEngine> StorageConfigManger<EK> {
     pub fn new(
-        kvdb: RocksEngine,
+        kvdb: EK,
         shared_block_cache: bool,
         ttl_checker_scheduler: Scheduler<TTLCheckerTask>,
         flow_controller: Arc<FlowController>,
-    ) -> StorageConfigManger {
+    ) -> Self {
         StorageConfigManger {
             kvdb,
             shared_block_cache,
@@ -128,7 +127,7 @@ impl StorageConfigManger {
     }
 }
 
-impl ConfigManager for StorageConfigManger {
+impl<EK: KvEngine> ConfigManager for StorageConfigManger<EK> {
     fn dispatch(&mut self, mut change: ConfigChange) -> CfgResult<()> {
         if let Some(ConfigValue::Module(mut block_cache)) = change.remove("block_cache") {
             if !self.shared_block_cache {