refactor(meta-service): merge Marked and SeqMarked into one (#18350)

* refactor(meta-service): merge `Marked` and `SeqMarked` into one

These two structures provide similar functionality and should be just
one.

Internally, `SeqMarked` is a type for value with sequence number and
tombstone mark. Since this commit it is defined in a shared crate
`seqmarked` and reused by all depending crates:

- `rotbl`: the snapshot storage impl;
- `map-api`: abstract map like API;
- databend raft-store impl

* M  Cargo.lock

* M  Cargo.lock

* M  src/meta/raft-store/Cargo.toml

* M  Cargo.toml

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

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

Author: drmingdrmer

Cargo.lock

@@ -381,7 +381,7 @@ logforth = { git = "https://github.com/datafuse-extras/logforth", branch = "glob
     'fastrace',
 ] }
 lz4 = "1.24.0"
-map-api = { version = "0.2.3" }
+map-api = { version = "0.2.5" }
 maplit = "1.0.2"
 match-template = "0.0.1"
 md-5 = "0.10.5"
@@ -466,7 +466,7 @@ reqwest-hickory-resolver = "0.2"
 ringbuffer = "0.14.2"
 rmp-serde = "1.1.1"
 roaring = { version = "^0.10", features = ["serde"] }
-rotbl = { version = "0.2.3", features = [] }
+rotbl = { version = "0.2.6", features = [] }
 rust_decimal = "1.26"
 rustix = "0.38.37"
 rustls = { version = "0.23.27", features = ["ring", "tls12"], default-features = false }
@@ -476,6 +476,7 @@ rustyline = "14"
 scroll = "0.12.0"
 self_cell = "1.2.0"
 semver = "1.0.14"
+seq-marked = { version = "0.3.1", features = ["seq-marked-serde", "seq-marked-bincode", "seqv-serde"] }
 serde = { version = "1.0.164", features = ["derive", "rc"] }
 serde_derive = "1"
 serde_ignored = "0.1.10"
@@ -657,7 +658,7 @@ 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.2.3" }
+map-api = { git = "https://github.com/databendlabs/map-api", tag = "v0.2.6" }
 openai_api_rust = { git = "https://github.com/datafuse-extras/openai-api", rev = "819a0ed" }
 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" }

Cargo.toml

@@ -86,7 +86,7 @@ where
     let x: Result<_, PbDecodeError> = try {
         let p: T::PB = prost::Message::decode(buf.as_ref())?;
         let v: T = FromToProto::from_pb(p)?;
-        SeqV::with_meta(seqv.seq, seqv.meta, v)
+        SeqV::new_with_meta(seqv.seq, seqv.meta, v)
     };
 
     x.map_err(|e| e.with_context(context()))

src/meta/api/src/kv_pb_api/codec.rs

@@ -89,7 +89,7 @@ where
     if let Some(pb_seqv) = resp.value {
         let seqv = SeqV::from(pb_seqv);
         let value = deserialize_struct::<K::ValueType>(&seqv.data)?;
-        let seqv = SeqV::with_meta(seqv.seq, seqv.meta, value);
+        let seqv = SeqV::new_with_meta(seqv.seq, seqv.meta, value);
         Ok((key, Some(seqv)))
     } else {
         Ok((key, None))
@@ -111,7 +111,7 @@ where K: kvapi::Key {
     if let Some(pb_seqv) = resp.value {
         let seqv = SeqV::from(pb_seqv);
         let id = deserialize_u64(&seqv.data)?;
-        let seqv = SeqV::with_meta(seqv.seq, seqv.meta, id);
+        let seqv = SeqV::new_with_meta(seqv.seq, seqv.meta, id);
         Ok((key, Some(seqv)))
     } else {
         Ok((key, None))

src/meta/api/src/util.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::time::Duration;
+use std::time::SystemTime;
 
 use databend_common_meta_kvapi::kvapi;
 use databend_common_meta_types::protobuf as pb;
@@ -281,7 +282,7 @@ impl TestSuite {
             assert!(res.is_none(), "got expired");
         }
 
-        let now_sec = SeqV::<()>::now_sec();
+        let now_sec = since_epoch_secs();
 
         info!("--- expired entry act as if it does not exist, an ADD op should apply");
         {
@@ -293,7 +294,7 @@ impl TestSuite {
                         .with(MetaSpec::new_expire(now_sec - 1)),
                 )
                 .await?;
-            // dbg!("update expired k1", _res);
+            // dbg!("update on expired k1", _res);
 
             let _res = kv
                 .upsert_kv(
@@ -305,8 +306,14 @@ impl TestSuite {
             // dbg!("update non expired k2", _res);
 
             info!("--- mget should not return expired");
-            let mut res = kv.mget_kv(&["k1".to_string(), "k2".to_string()]).await?;
             {
+                // let got = kv.get_kv("k1").await?;
+                // dbg!("k1", got);
+                // let got = kv.get_kv("k2").await?;
+                // dbg!("k2", got);
+
+                let mut res = kv.mget_kv(&["k1".to_string(), "k2".to_string()]).await?;
+                // dbg!(&res);
                 assert_eq!(res[0], None);
 
                 let v2 = res.remove(1).unwrap();
@@ -362,7 +369,7 @@ impl TestSuite {
 
         info!("--- expired in milliseconds");
         {
-            let now_sec = SeqV::<()>::now_sec();
+            let now_sec = since_epoch_secs();
 
             // expire time in milliseconds
             kv.upsert_kv(
@@ -451,7 +458,7 @@ impl TestSuite {
 
         let test_key = "test_key_for_update_meta";
 
-        let now_sec = SeqV::<()>::now_sec();
+        let now_sec = since_epoch_secs();
 
         let r = kv.upsert_kv(UpsertKV::update(test_key, b"v1")).await?;
         assert_eq!(Some(SeqV::new(1, b("v1"))), r.result);
@@ -872,7 +879,7 @@ impl TestSuite {
                 TxnOpResponse {
                     response: Some(txn_op_response::Response::Get(TxnGetResponse {
                         key: txn_key1.clone(),
-                        value: Some(pb::SeqV::from(SeqV::with_meta(
+                        value: Some(pb::SeqV::from(SeqV::new_with_meta(
                             6,
                             Some(KVMeta::default()),
                             val1_new.clone(),
@@ -884,7 +891,7 @@ impl TestSuite {
                     response: Some(txn_op_response::Response::Delete(TxnDeleteResponse {
                         key: txn_key1.clone(),
                         success: true,
-                        prev_value: Some(pb::SeqV::from(SeqV::with_meta(
+                        prev_value: Some(pb::SeqV::from(SeqV::new_with_meta(
                             6,
                             Some(KVMeta::default()),
                             val1_new.clone(),
@@ -1427,7 +1434,7 @@ impl TestSuite {
 
         info!("--- {}", func_path!());
 
-        let now_ms = SeqV::<()>::now_ms();
+        let now_ms = since_epoch_millis();
 
         let txn = TxnRequest::new(vec![], vec![
             TxnOp::put_sequential("k1/", "seq1", b("v1")).with_expires_at_ms(Some(now_ms + 1000)),
@@ -1906,3 +1913,17 @@ fn normalize_txn_response(vs: Vec<TxnOpResponse>) -> Vec<TxnOpResponse> {
 fn b(x: impl ToString) -> Vec<u8> {
     x.to_string().as_bytes().to_vec()
 }
+
+fn since_epoch_secs() -> u64 {
+    since_epoch().as_secs()
+}
+
+fn since_epoch_millis() -> u64 {
+    since_epoch().as_millis() as u64
+}
+
+fn since_epoch() -> Duration {
+    SystemTime::now()
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .unwrap()
+}

src/meta/kvapi-test-suite/src/kvapi_test_suite.rs

@@ -42,6 +42,7 @@ raft-log = { workspace = true }
 rmp-serde = { workspace = true }
 rotbl = { workspace = true }
 semver = { workspace = true }
+seq-marked = { workspace = true, features = ["seq-marked-serde", "seq-marked-bincode", "seqv-serde"] }
 serde = { workspace = true }
 serde_json = { workspace = true }
 stream-more = { workspace = true }

src/meta/raft-store/Cargo.toml

@@ -675,6 +675,7 @@ where SM: StateMachineApi + 'static
 
         for (expire_key, key) in to_clean {
             let curr = self.sm.get_maybe_expired_kv(&key).await?;
+
             if let Some(seq_v) = &curr {
                 assert_eq!(expire_key.seq, seq_v.seq);
                 info!("clean expired: {}, {}", key, expire_key);

src/meta/raft-store/src/applier/mod.rs

@@ -102,11 +102,14 @@ impl<'a> DBExporter<'a> {
 
         // kv
 
-        let strm = MapView(self.db).str_map().range(..).await?;
-        let kv_strm = strm.try_filter_map(|(str_k, marked)| {
+        let strm = MapView(self.db).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<_>> = marked.into();
-            let ent = seqv.map(|value| SMEntry::GenericKV { key: str_k, value });
+            let seqv: Option<SeqV<_>> = seq_marked.into();
+            let ent = seqv.map(|value| SMEntry::GenericKV {
+                key: user_key.to_string(),
+                value,
+            });
             future::ready(Ok(ent))
         });
 

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

@@ -16,7 +16,6 @@ use std::io;
 use std::ops::RangeBounds;
 
 use databend_common_meta_types::snapshot_db::DB;
-use databend_common_meta_types::KVMeta;
 use futures_util::StreamExt;
 use map_api::map_api_ro::MapApiRO;
 use rotbl::v001::SeqMarked;
@@ -27,31 +26,30 @@ 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;
-use crate::marked::Marked;
 
 /// A wrapper that implements the `MapApiRO` trait for the `DB`.
 #[derive(Debug, Clone)]
 pub struct MapView<'a>(pub &'a DB);
 
 #[async_trait::async_trait]
-impl<K> MapApiRO<K, KVMeta> for MapView<'_>
+impl<K> MapApiRO<K> for MapView<'_>
 where
-    K: MapKey<KVMeta>,
+    K: MapKey,
     K: MapKeyEncode,
     K: MapKeyDecode,
-    Marked<K::V>: ValueConvert<SeqMarked>,
+    SeqMarked<K::V>: ValueConvert<SeqMarked>,
 {
-    async fn get(&self, key: &K) -> Result<Marked<K::V>, io::Error> {
+    async fn get(&self, key: &K) -> Result<SeqMarked<K::V>, io::Error> {
         let key = RotblCodec::encode_key(key)
             .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
 
         let res = self.0.rotbl.get(&key).await?;
 
         let Some(seq_marked) = res else {
-            return Ok(Marked::empty());
+            return Ok(SeqMarked::new_not_found());
         };
 
-        let marked = Marked::<K::V>::conv_from(seq_marked)?;
+        let marked = SeqMarked::<K::V>::conv_from(seq_marked)?;
         Ok(marked)
     }
 
@@ -65,7 +63,7 @@ where
         let strm = strm.map(|res_item: Result<(String, SeqMarked), io::Error>| {
             let (str_k, seq_marked) = res_item?;
             let key = RotblCodec::decode_key(&str_k)?;
-            let marked = Marked::conv_from(seq_marked)?;
+            let marked = SeqMarked::conv_from(seq_marked)?;
             Ok((key, marked))
         });