refactor(raft-store): update trait interfaces and restructure leveled map (#18719)

* refactor(raft-store): update trait interfaces and restructure leveled map

Upgrade map-api to 0.4.0

* M  Cargo.lock

* A  src/meta/raft-store/src/leveled_store/leveled_map/impl_scoped_seq_bounded_get.rs

* M  src/meta/raft-store/src/leveled_store/leveled_map/impl_scoped_seq_bounded_get.rs

* M  src/meta/types/src/snapshot_db.rs

Author: drmingdrmer

Cargo.lock

@@ -376,7 +376,7 @@ logforth = { git = "https://github.com/datafuse-extras/logforth", branch = "main
     'fastrace',
 ] }
 lz4 = "1.24.0"
-map-api = { version = "0.3.1" }
+map-api = { version = "0.4.1" }
 maplit = "1.0.2"
 match-template = "0.0.1"
 md-5 = "0.10.5"
@@ -498,7 +498,7 @@ socket2 = "0.5.3"
 span-map = { version = "0.2.0" }
 sqlx = { version = "0.8", features = ["mysql", "runtime-tokio"] }
 state = "0.6.0"
-state-machine-api = { version = "0.2.1" }
+state-machine-api = { version = "0.3.1" }
 stream-more = "0.1.3"
 strength_reduce = "0.2.4"
 stringslice = "0.2.0"
@@ -657,12 +657,12 @@ backtrace = { git = "https://github.com/rust-lang/backtrace-rs.git", rev = "7226
 color-eyre = { git = "https://github.com/eyre-rs/eyre.git", rev = "e5d92c3" }
 deltalake = { git = "https://github.com/delta-io/delta-rs", rev = "9954bff" }
 display-more = { git = "https://github.com/databendlabs/display-more", tag = "v0.2.0" }
-map-api = { git = "https://github.com/databendlabs/map-api", tag = "v0.3.1" }
+map-api = { git = "https://github.com/databendlabs/map-api", tag = "v0.4.1" }
 openraft = { git = "https://github.com/databendlabs/openraft", tag = "v0.10.0-alpha.9" }
 orc-rust = { git = "https://github.com/datafuse-extras/orc-rust", rev = "d82aa6d" }
 recursive = { git = "https://github.com/datafuse-extras/recursive.git", rev = "16e433a" }
 sled = { git = "https://github.com/datafuse-extras/sled", tag = "v0.34.7-datafuse.1" }
-state-machine-api = { git = "https://github.com/databendlabs/state-machine-api.git", tag = "v0.2.1" }
+state-machine-api = { git = "https://github.com/databendlabs/state-machine-api.git", tag = "v0.3.1" }
 sub-cache = { git = "https://github.com/databendlabs/sub-cache", tag = "v0.2.1" }
 tantivy = { git = "https://github.com/datafuse-extras/tantivy", rev = "7502370" }
 tantivy-common = { git = "https://github.com/datafuse-extras/tantivy", rev = "7502370", package = "tantivy-common" }

Cargo.toml

@@ -16,22 +16,10 @@ use std::sync::Arc;
 use std::sync::Mutex;
 use std::time::Duration;
 
-use map_api::mvcc;
-
 use crate::leveled_store::leveled_map::applier_acquirer::WriterPermit;
-use crate::leveled_store::leveled_map::leveled_map_data::LeveledMapData;
-use crate::leveled_store::types::Key;
-use crate::leveled_store::types::Namespace;
-use crate::leveled_store::types::Value;
+use crate::leveled_store::view::StateMachineView;
 use crate::sm_v003::OnChange;
 
-mod impl_expire_scoped_view;
-mod impl_expire_scoped_view_readonly;
-mod impl_user_scoped_view;
-mod impl_user_scoped_view_readonly;
-
-pub(crate) type StateMachineView = mvcc::View<Namespace, Key, Value, Arc<LeveledMapData>>;
-
 pub(crate) struct ApplierData {
     /// Hold a unique permit to serialize all apply operations to the state machine.
     pub(crate) _permit: WriterPermit,

src/meta/raft-store/src/applier/applier_data/impl_expire_scoped_view_readonly.rs

@@ -23,13 +23,14 @@ use databend_common_meta_types::SeqV;
 use futures_util::StreamExt;
 use futures_util::TryStreamExt;
 use log::info;
-use map_api::map_api_ro::MapApiRO;
+use map_api::mvcc::ScopedSeqBoundedRange;
 use map_api::IOResultStream;
+use state_machine_api::ExpireKey;
 use state_machine_api::ExpireValue;
+use state_machine_api::UserKey;
 
 use crate::key_spaces::SMEntry;
-use crate::leveled_store::db_map_api_ro_impl::MapView;
-use crate::leveled_store::map_api::AsMap;
+use crate::leveled_store::db_impl_scoped_seq_bounded_read::ScopedSeqBoundedRead;
 use crate::state_machine::StateMachineMetaKey;
 use crate::state_machine::StateMachineMetaValue;
 
@@ -92,7 +93,9 @@ impl<'a> DBExporter<'a> {
 
         // expire index
 
-        let strm = MapView(self.db).as_expire_map().range(..).await?;
+        let strm = ScopedSeqBoundedRead(self.db)
+            .range(ExpireKey::default().., u64::MAX)
+            .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 +105,9 @@ impl<'a> DBExporter<'a> {
 
         // kv
 
-        let strm = MapView(self.db).as_user_map().range(..).await?;
+        let strm = ScopedSeqBoundedRead(self.db)
+            .range(UserKey::default().., u64::MAX)
+            .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/applier/applier_data/impl_user_scoped_view_readonly.rs

@@ -13,34 +13,40 @@
 // limitations under the License.
 
 use std::io;
+use std::io::Error;
 use std::ops::RangeBounds;
 
 use databend_common_meta_types::snapshot_db::DB;
 use futures_util::StreamExt;
-use map_api::map_api_ro::MapApiRO;
+use map_api::mvcc;
+use map_api::mvcc::ViewKey;
+use map_api::mvcc::ViewValue;
+use map_api::IOResultStream;
 use rotbl::v001::SeqMarked;
 
-use crate::leveled_store::map_api::KVResultStream;
 use crate::leveled_store::map_api::MapKey;
 use crate::leveled_store::map_api::MapKeyDecode;
 use crate::leveled_store::map_api::MapKeyEncode;
 use crate::leveled_store::rotbl_codec::RotblCodec;
 use crate::leveled_store::value_convert::ValueConvert;
 
-/// A wrapper that implements the `MapApiRO` trait for the `DB`.
+/// A wrapper that implements the `ScopedSnapshot*` trait for the `DB`.
 #[derive(Debug, Clone)]
-pub struct MapView<'a>(pub &'a DB);
+pub struct ScopedSeqBoundedRead<'a>(pub &'a DB);
 
+// TODO: test
 #[async_trait::async_trait]
-impl<K> MapApiRO<K> for MapView<'_>
+impl<K> mvcc::ScopedSeqBoundedGet<K, K::V> for ScopedSeqBoundedRead<'_>
 where
     K: MapKey,
-    K: MapKeyEncode,
-    K: MapKeyDecode,
+    K: ViewKey,
+    K: MapKeyEncode + MapKeyDecode,
+    K::V: ViewValue,
     SeqMarked<K::V>: ValueConvert<SeqMarked>,
 {
-    async fn get(&self, key: &K) -> Result<SeqMarked<K::V>, io::Error> {
-        let key = RotblCodec::encode_key(key)
+    async fn get(&self, key: K, _snapshot_seq: u64) -> Result<SeqMarked<K::V>, io::Error> {
+        // TODO: DB does not consider snapshot_seq
+        let key = RotblCodec::encode_key(&key)
             .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
 
         let res = self.0.rotbl.get(&key).await?;
@@ -52,9 +58,28 @@ where
         let marked = SeqMarked::<K::V>::conv_from(seq_marked)?;
         Ok(marked)
     }
+}
+
+// TODO: test
+#[async_trait::async_trait]
+impl<K> mvcc::ScopedSeqBoundedRange<K, K::V> for ScopedSeqBoundedRead<'_>
+where
+    K: MapKey,
+    K: ViewKey,
+    K: MapKeyEncode + MapKeyDecode,
+    K::V: ViewValue,
+    SeqMarked<K::V>: ValueConvert<SeqMarked>,
+{
+    async fn range<R>(
+        &self,
+        range: R,
+        _snapshot_seq: u64,
+    ) -> Result<IOResultStream<(K, SeqMarked<K::V>)>, Error>
+    where
+        R: RangeBounds<K> + Send + Sync + Clone + 'static,
+    {
+        // TODO: DB does not consider snapshot_seq
 
-    async fn range<R>(&self, range: R) -> Result<KVResultStream<K>, io::Error>
-    where R: RangeBounds<K> + Clone + Send + Sync + 'static {
         let rng = RotblCodec::encode_range(&range)
             .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;