refactor(meta-service): simplify StateMachineApi (#18366)

Author: drmingdrmer

src/meta/raft-store/src/leveled_store/db_exporter.rs

@@ -92,7 +92,7 @@ impl<'a> DBExporter<'a> {
 
         // expire index
 
-        let strm = MapView(self.db).expire_map().range(..).await?;
+        let strm = MapView(self.db).as_expire_map().range(..).await?;
         let expire_strm = strm.try_filter_map(|(exp_k, marked)| {
             // Tombstone will be converted to None and be ignored.
             let exp_val = ExpireValue::from_marked(marked);
@@ -102,7 +102,7 @@ impl<'a> DBExporter<'a> {
 
         // kv
 
-        let strm = MapView(self.db).user_map().range(..).await?;
+        let strm = MapView(self.db).as_user_map().range(..).await?;
         let kv_strm = strm.try_filter_map(|(user_key, seq_marked)| {
             // Tombstone will be converted to None and be ignored.
             let seqv: Option<SeqV<_>> = seq_marked.into();

src/meta/raft-store/src/leveled_store/db_map_api_ro_test.rs

@@ -68,7 +68,7 @@ async fn test_db_map_api_ro() -> anyhow::Result<()> {
     // Test kv map
 
     let binding = MapView(&db);
-    let smap = binding.user_map();
+    let smap = binding.as_user_map();
     assert_eq!(
         SeqMarked::new_normal(4, (Some(KVMeta::new(Some(15))), b("a1"))),
         smap.get(&user_key("a")).await?
@@ -102,7 +102,7 @@ async fn test_db_map_api_ro() -> anyhow::Result<()> {
     // Test expire index
 
     let binding = MapView(&db);
-    let emap = binding.expire_map();
+    let emap = binding.as_expire_map();
 
     assert_eq!(
         SeqMarked::new_normal(4, s("a")),

src/meta/raft-store/src/leveled_store/immutable.rs

@@ -122,7 +122,7 @@ impl Immutable {
 impl MapApiRO<UserKey> for Immutable {
     async fn get(&self, key: &UserKey) -> Result<SeqMarked<MetaValue>, io::Error> {
         // get() is just delegated
-        self.as_ref().user_map().get(key).await
+        self.as_ref().as_user_map().get(key).await
     }
 
     async fn range<R>(&self, range: R) -> Result<KVResultStream<UserKey>, io::Error>
@@ -136,7 +136,7 @@ impl MapApiRO<UserKey> for Immutable {
 impl MapApiRO<ExpireKey> for Immutable {
     async fn get(&self, key: &ExpireKey) -> Result<SeqMarkedOf<ExpireKey>, io::Error> {
         // get() is just delegated
-        self.as_ref().expire_map().get(key).await
+        self.as_ref().as_expire_map().get(key).await
     }
 
     async fn range<R>(&self, range: R) -> Result<KVResultStream<ExpireKey>, io::Error>

src/meta/raft-store/src/leveled_store/level.rs

@@ -131,7 +131,7 @@ impl MapApi<UserKey> for Level {
             SeqMarked::new_tombstone(seq)
         };
 
-        let prev = (*self).user_map().get(&key).await?;
+        let prev = (*self).as_user_map().get(&key).await?;
         self.kv.insert(key, marked.clone());
         Ok((prev, marked))
     }
@@ -184,7 +184,7 @@ impl MapApi<ExpireKey> for Level {
             SeqMarked::new_tombstone(seq)
         };
 
-        let prev = (*self).expire_map().get(&key).await?;
+        let prev = (*self).as_expire_map().get(&key).await?;
         self.expire.insert(key, marked.clone());
         Ok((prev, marked))
     }

src/meta/raft-store/src/leveled_store/leveled_map/compacting_data.rs

@@ -71,12 +71,12 @@ impl<'a> CompactingData<'a> {
         let mut data = newest.new_level();
 
         // Copy all expire data and keep tombstone.
-        let strm = (*immutable_levels).expire_map().range(..).await?;
+        let strm = (*immutable_levels).as_expire_map().range(..).await?;
         let bt = strm.try_collect().await?;
         data.replace_expire(bt);
 
         // Copy all kv data and keep tombstone.
-        let strm = (*immutable_levels).user_map().range(..).await?;
+        let strm = (*immutable_levels).as_user_map().range(..).await?;
         let bt = strm.try_collect().await?;
         data.replace_kv(bt);
 
@@ -110,7 +110,7 @@ impl<'a> CompactingData<'a> {
 
         // expire index: prefix `exp-/`.
 
-        let strm = (*self.immutable_levels).expire_map().range(..).await?;
+        let strm = (*self.immutable_levels).as_expire_map().range(..).await?;
         let expire_strm = strm.map(|item: Result<(ExpireKey, SeqMarked<String>), io::Error>| {
             let (expire_key, marked_string) = item?;
 
@@ -120,7 +120,7 @@ impl<'a> CompactingData<'a> {
 
         // kv: prefix: `kv--/`
 
-        let strm = (*self.immutable_levels).user_map().range(..).await?;
+        let strm = (*self.immutable_levels).as_user_map().range(..).await?;
         let kv_strm = strm.map(|item: Result<MapKV<UserKey>, io::Error>| {
             let (k, v) = item?;