Merge branch 'main' into ST_YMAX

Author: kkk25641463

Cargo.lock

@@ -31,6 +31,7 @@ use anyhow::anyhow;
 use databend_common_base::base::tokio;
 use databend_common_meta_raft_store::config::RaftConfig;
 use databend_common_meta_raft_store::key_spaces::RaftStoreEntry;
+use databend_common_meta_raft_store::key_spaces::SMEntry;
 use databend_common_meta_raft_store::ondisk::DataVersion;
 use databend_common_meta_raft_store::ondisk::OnDisk;
 use databend_common_meta_raft_store::sm_v002::leveled_store::sys_data_api::SysDataApiRO;
@@ -175,17 +176,26 @@ async fn import_v002(
 
     let snapshot_store = SnapshotStoreV002::new(DataVersion::V002, raft_config);
 
-    let (tx, join_handle) = snapshot_store.spawn_writer_thread("import_v002");
+    let writer = snapshot_store.new_writer()?;
+
+    let (tx, join_handle) = writer.spawn_writer_thread("import_v002");
+
+    let mut last_applied = None;
 
     for line in lines {
         let l = line?;
         let (tree_name, kv_entry): (String, RaftStoreEntry) = serde_json::from_str(&l)?;
 
         if tree_name.starts_with("state_machine/") {
             // Write to snapshot
-            let sm_entry = kv_entry.try_into().map_err(|err_str| {
+            let sm_entry: SMEntry = kv_entry.try_into().map_err(|err_str| {
                 anyhow::anyhow!("Failed to convert RaftStoreEntry to SMEntry: {}", err_str)
             })?;
+
+            if let Some(last) = sm_entry.last_applied() {
+                last_applied = Some(last);
+            }
+
             tx.send(WriteEntry::Data(sm_entry)).await?;
         } else {
             // Write to sled tree
@@ -214,9 +224,21 @@ async fn import_v002(
 
     tx.send(WriteEntry::Finish).await?;
 
-    let (_snapshot_store, snapshot_stat) = join_handle.await??;
-
-    eprintln!("Imported {} records, snapshot: {}", n, snapshot_stat,);
+    let (temp_snapshot_data, snapshot_stat) = join_handle.await??;
+
+    let (snapshot_id, snapshot_data) = snapshot_store.commit_snapshot_data_gen_id(
+        temp_snapshot_data,
+        last_applied,
+        snapshot_stat.entry_cnt,
+    )?;
+
+    eprintln!(
+        "Imported {} records, snapshot: {}; snapshot_path: {}; snapshot_stat: {}",
+        n,
+        snapshot_id.to_string(),
+        snapshot_data.path(),
+        snapshot_stat,
+    );
     Ok(max_log_id)
 }
 

src/binaries/metactl/snapshot.rs

@@ -96,11 +96,11 @@ History versions that are not included in the above chart:
 
 ## Compatibility between databend-meta
 
-| Meta version         | Backward compatible with |
-|:---------------------|:-------------------------|
-| [0.9.41,   1.2.212)  | [0.9.41,  1.2.212)       |
-| [1.2.212,  1.2.476?) | [0.9.41,  1.2.476?)      |
-| [1.2.476?, +∞)       | [1.2.212, +∞)            |
+| Meta version        | Backward compatible with |
+|:--------------------|:-------------------------|
+| [0.9.41,   1.2.212) | [0.9.41,  1.2.212)       |
+| [1.2.212,  1.2.479) | [0.9.41,  1.2.479)      |
+| [1.2.479, +∞)       | [1.2.212, +∞)            |
 
 TODO: fix the above version when merged
 
@@ -116,7 +116,7 @@ TODO: fix the above version when merged
   In this version, databend-meta raft-server introduced a new API `install_snapshot_v1()`.
   The raft-client will try to use either this new API or the original `install_snapshot()`.
 
-- `1.2.476?` Remove: `install_snapshot()`(v0) from client and server.
+- `1.2.479` Remove: `install_snapshot()`(v0) from client and server.
   The `install_snapshot_v1()` is the only API to install snapshot, and becomes **REQUIRED** for the client.
 
 

src/meta/README.md

@@ -194,7 +194,7 @@ impl RaftConfig {
     ///
     /// Raft will choose a random timeout in this range for next election.
     pub fn election_timeout(&self) -> (u64, u64) {
-        (self.heartbeat_interval * 5, self.heartbeat_interval * 7)
+        (self.heartbeat_interval * 2, self.heartbeat_interval * 3)
     }
 
     pub fn check(&self) -> std::result::Result<(), MetaStartupError> {

src/meta/raft-store/src/config.rs

@@ -21,6 +21,7 @@ use databend_common_meta_sled_store::SledOrderedSerde;
 use databend_common_meta_sled_store::SledSerde;
 use databend_common_meta_stoerr::MetaStorageError;
 use databend_common_meta_types::Entry;
+use databend_common_meta_types::LogId;
 use databend_common_meta_types::LogIndex;
 use databend_common_meta_types::Node;
 use databend_common_meta_types::NodeId;
@@ -209,6 +210,20 @@ impl SMEntry {
 
         unreachable!("unknown prefix: {}", prefix);
     }
+
+    pub fn last_applied(&self) -> Option<LogId> {
+        match self {
+            Self::StateMachineMeta { key, value } => {
+                if *key == StateMachineMetaKey::LastApplied {
+                    let last: LogId = value.clone().try_into().unwrap();
+                    Some(last)
+                } else {
+                    None
+                }
+            }
+            _ => None,
+        }
+    }
 }
 
 /// Enum of key-value pairs that are used in the raft storage impl for meta-service.

src/meta/raft-store/src/key_spaces.rs

@@ -285,8 +285,13 @@ impl OnDisk {
         let tree = self.db.open_tree(sm_tree_name)?;
 
         let snapshot_store = SnapshotStoreV002::new(DataVersion::V002, self.config.clone());
+        let mut last_applied = None;
 
-        let (tx, join_handle) = snapshot_store.spawn_writer_thread("upgrade-v001-to-v002-snapshot");
+        let writer = snapshot_store
+            .new_writer()
+            .map_err(|e| snap_err(e, "new snapshot writer"))?;
+
+        let (tx, join_handle) = writer.spawn_writer_thread("upgrade-v001-to-v002-snapshot");
 
         for ivec_pair_res in tree.iter() {
             let sm_entry = {
@@ -311,6 +316,11 @@ impl OnDisk {
                 continue;
             }
 
+            if let Some(last) = sm_entry.last_applied() {
+                self.progress(format_args!("found state machine last_applied: {}", last));
+                last_applied = Some(last);
+            }
+
             tx.send(WriteEntry::Data(sm_entry))
                 .await
                 .map_err(|e| snap_err(e, "send SMEntry"))?;
@@ -320,15 +330,24 @@ impl OnDisk {
             .await
             .map_err(|e| snap_err(e, "send Commit"))?;
 
-        let (snapshot_store, snapshot_stat) = join_handle
+        let (temp_snapshot_data, snapshot_stat) = join_handle
             .await
             .map_err(|e| snap_err(e, "join snapshot writer thread"))?
             .map_err(|e| snap_err(e, "writer error"))?;
 
+        if snapshot_stat.entry_cnt > 0 {
+            assert!(last_applied.is_some(), "last_applied must be Some");
+        }
+
+        let (snapshot_id, snapshot_data) = snapshot_store
+            .commit_snapshot_data_gen_id(temp_snapshot_data, last_applied, snapshot_stat.entry_cnt)
+            .map_err(|e| snap_err(e, "commit snapshot data"))?;
+
         self.progress(format_args!(
-            "Written to snapshot: {}, path: {}",
+            "Written to snapshot: {}, {}; path: {}",
+            snapshot_id.to_string(),
             snapshot_stat,
-            snapshot_store.snapshot_path(&snapshot_stat.snapshot_id.to_string())
+            snapshot_data.path()
         ));
 
         Ok(())

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

@@ -14,6 +14,7 @@
 
 use std::borrow::Borrow;
 use std::io;
+use std::ops::Deref;
 use std::ops::RangeBounds;
 use std::sync::Arc;
 
@@ -27,10 +28,50 @@ use crate::sm_v002::leveled_store::map_api::MarkedOf;
 use crate::sm_v002::marked::Marked;
 use crate::state_machine::ExpireKey;
 
-impl Level {
+/// A single **immutable** level of state machine data.
+///
+/// Immutable level implement only [`MapApiRO`], but not [`MapApi`].
+///
+/// [`MapApi`]: crate::sm_v002::leveled_store::map_api::MapApi
+#[derive(Debug, Clone)]
+pub struct Immutable {
+    level: Arc<Level>,
+}
+
+impl Immutable {
+    pub fn new(level: Arc<Level>) -> Self {
+        Self { level }
+    }
+
+    pub fn new_from_level(level: Level) -> Self {
+        Self {
+            level: Arc::new(level),
+        }
+    }
+
+    pub fn inner(&self) -> &Arc<Level> {
+        &self.level
+    }
+}
+
+impl AsRef<Level> for Immutable {
+    fn as_ref(&self) -> &Level {
+        self.level.as_ref()
+    }
+}
+
+impl Deref for Immutable {
+    type Target = Level;
+
+    fn deref(&self) -> &Self::Target {
+        self.level.as_ref()
+    }
+}
+
+impl Immutable {
     /// Build a static stream that yields key values for primary index
     #[futures_async_stream::try_stream(boxed, ok = MapKV<String>, error = io::Error)]
-    async fn str_range<Q, R>(self: Arc<Level>, range: R)
+    async fn str_range<Q, R>(self: Immutable, range: R)
     where
         String: Borrow<Q>,
         Q: Ord + Send + Sync + ?Sized,
@@ -45,7 +86,7 @@ impl Level {
 
     /// Build a static stream that yields expire key and key for the secondary expiration index
     #[futures_async_stream::try_stream(boxed, ok = MapKV<ExpireKey>, error = io::Error)]
-    async fn expire_range<Q, R>(self: Arc<Level>, range: R)
+    async fn expire_range<Q, R>(self: Immutable, range: R)
     where
         ExpireKey: Borrow<Q>,
         Q: Ord + Send + Sync + ?Sized,
@@ -60,7 +101,7 @@ impl Level {
 }
 
 #[async_trait::async_trait]
-impl MapApiRO<String> for Arc<Level> {
+impl MapApiRO<String> for Immutable {
     async fn get<Q>(&self, key: &Q) -> Result<Marked<<String as MapKey>::V>, io::Error>
     where
         String: Borrow<Q>,
@@ -78,7 +119,7 @@ impl MapApiRO<String> for Arc<Level> {
 }
 
 #[async_trait::async_trait]
-impl MapApiRO<ExpireKey> for Arc<Level> {
+impl MapApiRO<ExpireKey> for Immutable {
     async fn get<Q>(&self, key: &Q) -> Result<MarkedOf<ExpireKey>, io::Error>
     where
         ExpireKey: Borrow<Q>,

…/sm_v002/leveled_store/arc_level_impl.rs …e/src/sm_v002/leveled_store/immutable.rssrc/meta/raft-store/src/sm_v002/leveled_store/arc_level_impl.rs renamed to src/meta/raft-store/src/sm_v002/leveled_store/immutable.rs src/meta/raft-store/src/sm_v002/leveled_store/arc_level_impl.rs renamed to src/meta/raft-store/src/sm_v002/leveled_store/immutable.rs

@@ -15,8 +15,8 @@
 use std::borrow::Borrow;
 use std::io;
 use std::ops::RangeBounds;
-use std::sync::Arc;
 
+use crate::sm_v002::leveled_store::immutable::Immutable;
 use crate::sm_v002::leveled_store::level::Level;
 use crate::sm_v002::leveled_store::map_api::compacted_get;
 use crate::sm_v002::leveled_store::map_api::compacted_range;
@@ -28,20 +28,20 @@ use crate::sm_v002::marked::Marked;
 
 /// A readonly leveled map that owns the data.
 #[derive(Debug, Default, Clone)]
-pub struct StaticLevels {
+pub struct ImmutableLevels {
     /// From oldest to newest, i.e., levels[0] is the oldest
-    levels: Vec<Arc<Level>>,
+    levels: Vec<Immutable>,
 }
 
-impl StaticLevels {
-    pub(in crate::sm_v002) fn new(levels: impl IntoIterator<Item = Arc<Level>>) -> Self {
+impl ImmutableLevels {
+    pub(in crate::sm_v002) fn new(levels: impl IntoIterator<Item = Immutable>) -> Self {
         Self {
             levels: levels.into_iter().collect(),
         }
     }
 
     /// Return an iterator of all Arc of levels from newest to oldest.
-    pub(in crate::sm_v002) fn iter_arc_levels(&self) -> impl Iterator<Item = &Arc<Level>> {
+    pub(in crate::sm_v002) fn iter_immutable_levels(&self) -> impl Iterator<Item = &Immutable> {
         self.levels.iter().rev()
     }
 
@@ -50,11 +50,11 @@ impl StaticLevels {
         self.levels.iter().map(|x| x.as_ref()).rev()
     }
 
-    pub(in crate::sm_v002) fn newest(&self) -> Option<&Arc<Level>> {
+    pub(in crate::sm_v002) fn newest(&self) -> Option<&Immutable> {
         self.levels.last()
     }
 
-    pub(in crate::sm_v002) fn push(&mut self, level: Arc<Level>) {
+    pub(in crate::sm_v002) fn push(&mut self, level: Immutable) {
         self.levels.push(level);
     }
 
@@ -69,24 +69,24 @@ impl StaticLevels {
 }
 
 #[async_trait::async_trait]
-impl<K> MapApiRO<K> for StaticLevels
+impl<K> MapApiRO<K> for ImmutableLevels
 where
     K: MapKey,
     Level: MapApiRO<K>,
-    Arc<Level>: MapApiRO<K>,
+    Immutable: MapApiRO<K>,
 {
     async fn get<Q>(&self, key: &Q) -> Result<Marked<K::V>, io::Error>
     where
         K: Borrow<Q>,
         Q: Ord + Send + Sync + ?Sized,
     {
-        let levels = self.iter_arc_levels();
+        let levels = self.iter_immutable_levels();
         compacted_get(key, levels).await
     }
 
     async fn range<R>(&self, range: R) -> Result<KVResultStream<K>, io::Error>
     where R: RangeBounds<K> + Clone + Send + Sync + 'static {
-        let levels = self.iter_arc_levels();
+        let levels = self.iter_immutable_levels();
         compacted_range::<_, _, _, Level>(range, None, levels).await
     }
 }