feat: impl recovering version from manifest for region (#127)

* feat: impl recovering version from manifest for region

* refactor: rename try_apply_edit to replay_edit

* fix: remove println

* fix: address CR problems

* feat: remove Metadata in manifest trait and update region manifest state after recovering

src/storage/src/error.rs

@@ -210,6 +210,12 @@ pub enum Error {
         message: String,
         backtrace: Backtrace,
     },
+
+    #[snafu(display("Region version not found in manifest, the region: {}", region_name))]
+    VersionNotFound {
+        region_name: String,
+        backtrace: Backtrace,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -235,7 +241,8 @@ impl ErrorExt for Error {
             | Readline { .. }
             | InvalidParquetSchema { .. }
             | SequenceColumnNotFound { .. }
-            | WalDataCorrupted { .. } => StatusCode::Unexpected,
+            | WalDataCorrupted { .. }
+            | VersionNotFound { .. } => StatusCode::Unexpected,
 
             FlushIo { .. }
             | InitBackend { .. }

src/storage/src/flush.rs

@@ -4,8 +4,7 @@ use async_trait::async_trait;
 use common_telemetry::logging;
 use common_time::RangeMillis;
 use store_api::logstore::LogStore;
-use store_api::manifest::Manifest;
-use store_api::manifest::ManifestVersion;
+use store_api::manifest::{Manifest, ManifestVersion, MetaAction};
 use store_api::storage::SequenceNumber;
 use uuid::Uuid;
 
@@ -200,18 +199,23 @@ impl<S: LogStore> FlushJob<S> {
 
     async fn write_to_manifest(&self, file_metas: &[FileMeta]) -> Result<ManifestVersion> {
         let edit = RegionEdit {
-            region_id: self.shared.id,
             region_version: self.shared.version_control.metadata().version,
-            flush_sequence: self.flush_sequence,
+            flushed_sequence: self.flush_sequence,
             files_to_add: file_metas.to_vec(),
             files_to_remove: Vec::default(),
         };
-        logging::debug!("Write region edit: {:?} to manifest.", edit);
-        self.manifest
-            .update(RegionMetaActionList::with_action(RegionMetaAction::Edit(
-                edit,
-            )))
-            .await
+        let prev_version = self.shared.version_control.current_manifest_version();
+
+        logging::debug!(
+            "Write region edit: {:?} to manifest, prev_version: {}.",
+            edit,
+            prev_version,
+        );
+
+        let mut action_list = RegionMetaActionList::with_action(RegionMetaAction::Edit(edit));
+        action_list.set_prev_version(prev_version);
+
+        self.manifest.update(action_list).await
     }
 
     /// Generates random SST file name in format: `^[a-f\d]{8}(-[a-f\d]{4}){3}-[a-f\d]{12}.parquet$`

src/storage/src/manifest.rs

@@ -3,3 +3,5 @@ pub(crate) mod action;
 pub(crate) mod checkpoint;
 pub mod region;
 pub(crate) mod storage;
+#[cfg(test)]
+pub mod test_utils;

src/storage/src/manifest/action.rs

@@ -8,7 +8,6 @@ use store_api::manifest::action::ProtocolAction;
 use store_api::manifest::action::ProtocolVersion;
 use store_api::manifest::ManifestVersion;
 use store_api::manifest::MetaAction;
-use store_api::manifest::Metadata;
 use store_api::storage::RegionId;
 use store_api::storage::SequenceNumber;
 
@@ -31,20 +30,12 @@ pub struct RegionRemove {
 
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
 pub struct RegionEdit {
-    pub region_id: RegionId,
     pub region_version: VersionNumber,
-    pub flush_sequence: SequenceNumber,
+    pub flushed_sequence: SequenceNumber,
     pub files_to_add: Vec<FileMeta>,
     pub files_to_remove: Vec<FileMeta>,
 }
 
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
-pub struct RegionManifestData {
-    pub region_meta: RegionMetadataRef,
-    // TODO(dennis): [open_region] 1. load version metadata 2. The `region_meta` field could be removed if we
-    // have a `version` field.
-}
-
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
 pub struct RegionMetaActionList {
     pub actions: Vec<RegionMetaAction>,
@@ -157,8 +148,6 @@ impl RegionMetaActionList {
     }
 }
 
-impl Metadata for RegionManifestData {}
-
 impl MetaAction for RegionMetaActionList {
     fn set_prev_version(&mut self, version: ManifestVersion) {
         self.prev_version = version;
@@ -170,6 +159,7 @@ mod tests {
     use common_telemetry::logging;
 
     use super::*;
+    use crate::manifest::test_utils;
 
     #[test]
     fn test_encode_decode_action_list() {
@@ -178,32 +168,18 @@ mod tests {
         protocol.min_reader_version = 1;
         let mut action_list = RegionMetaActionList::new(vec![
             RegionMetaAction::Protocol(protocol.clone()),
-            RegionMetaAction::Edit(RegionEdit {
-                region_id: 1,
-                region_version: 10,
-                flush_sequence: 99,
-                files_to_add: vec![
-                    FileMeta {
-                        file_name: "test1".to_string(),
-                        level: 1,
-                    },
-                    FileMeta {
-                        file_name: "test2".to_string(),
-                        level: 2,
-                    },
-                ],
-                files_to_remove: vec![FileMeta {
-                    file_name: "test0".to_string(),
-                    level: 0,
-                }],
-            }),
+            RegionMetaAction::Edit(test_utils::build_region_edit(
+                99,
+                &["test1", "test2"],
+                &["test3"],
+            )),
         ]);
         action_list.set_prev_version(3);
 
         let bs = action_list.encode().unwrap();
         // {"prev_version":3}
         // {"Protocol":{"min_reader_version":1,"min_writer_version":0}}
-        // {"Edit":{"region_id":1,"region_version":10,"flush_sequence":99,"files_to_add":[{"file_name":"test1","level":1},{"file_name":"test2","level":2}],"files_to_remove":[{"file_name":"test0","level":0}]}}
+        // {"Edit":{"region_version":0,"flush_sequence":99,"files_to_add":[{"file_name":"test1","level":1},{"file_name":"test2","level":2}],"files_to_remove":[{"file_name":"test0","level":0}]}}
 
         logging::debug!(
             "Encoded action list: \r\n{}",

src/storage/src/manifest/region.rs

@@ -28,48 +28,38 @@ impl RegionManifest {
             inner: Arc::new(RegionManifestInner::new(manifest_dir, object_store)),
         }
     }
+
+    /// Update inner state.
+    pub fn update_state(&self, version: ManifestVersion, protocol: Option<ProtocolAction>) {
+        self.inner.update_state(version, protocol);
+    }
 }
 
 #[async_trait]
 impl Manifest for RegionManifest {
     type Error = Error;
     type MetaAction = RegionMetaActionList;
-    type Metadata = RegionManifestData;
+    type MetaActionIterator = RegionMetaActionListIterator;
 
     async fn update(&self, action_list: RegionMetaActionList) -> Result<ManifestVersion> {
         self.inner.save(action_list).await
     }
 
-    async fn load(&self) -> Result<Option<RegionManifestData>> {
-        let last_version = self.inner.last_version();
-
-        let start_bound = if last_version == MIN_VERSION {
-            // No actions have ever saved
-            MIN_VERSION
-        } else {
-            last_version - 1
-        };
-
-        let mut iter = self.inner.scan(start_bound, MAX_VERSION).await?;
-
-        // TODO(yingwen): [open_region] 1. Create Version from metadata 2. Load VersionEdits
-        // and apply to the Version by `Version::apply_edit`.
-        while let Some((_v, action_list)) = iter.next_action().await? {
-            for action in action_list.actions {
-                if let RegionMetaAction::Change(c) = action {
-                    return Ok(Some(RegionManifestData {
-                        region_meta: c.metadata,
-                    }));
-                }
-            }
-        }
-
-        Ok(None)
+    async fn scan(
+        &self,
+        start: ManifestVersion,
+        end: ManifestVersion,
+    ) -> Result<RegionMetaActionListIterator> {
+        self.inner.scan(start, end).await
     }
 
     async fn checkpoint(&self) -> Result<ManifestVersion> {
         unimplemented!();
     }
+
+    fn last_version(&self) -> ManifestVersion {
+        self.inner.last_version()
+    }
 }
 
 #[derive(Debug)]
@@ -83,18 +73,33 @@ struct RegionManifestInner {
     supported_writer_version: ProtocolVersion,
 }
 
-struct RegionMetaActionListIterator {
+pub struct RegionMetaActionListIterator {
     log_iter: ObjectStoreLogIterator,
     reader_version: ProtocolVersion,
+    last_protocol: Option<ProtocolAction>,
 }
 
 impl RegionMetaActionListIterator {
+    pub fn last_protocol(&self) -> &Option<ProtocolAction> {
+        &self.last_protocol
+    }
+}
+
+#[async_trait]
+impl MetaActionIterator for RegionMetaActionListIterator {
+    type Error = Error;
+    type MetaAction = RegionMetaActionList;
+
     async fn next_action(&mut self) -> Result<Option<(ManifestVersion, RegionMetaActionList)>> {
         match self.log_iter.next_log().await? {
             Some((v, bytes)) => {
-                //TODO(dennis): save protocol into inner's protocol when recovering
-                let (action_list, _protocol) =
+                let (action_list, protocol) =
                     RegionMetaActionList::decode(&bytes, self.reader_version)?;
+
+                if protocol.is_some() {
+                    self.last_protocol = protocol;
+                }
+
                 Ok(Some((v, action_list)))
             }
             None => Ok(None),
@@ -108,7 +113,6 @@ impl RegionManifestInner {
 
         Self {
             store: Arc::new(ManifestObjectStore::new(manifest_dir, object_store)),
-            // TODO(dennis): recover the last version from history
             version: AtomicU64::new(0),
             protocol: ArcSwap::new(Arc::new(ProtocolAction::new())),
             supported_reader_version: reader_version,
@@ -121,6 +125,13 @@ impl RegionManifestInner {
         self.version.fetch_add(1, Ordering::Relaxed)
     }
 
+    fn update_state(&self, version: ManifestVersion, protocol: Option<ProtocolAction>) {
+        self.version.store(version, Ordering::Relaxed);
+        if let Some(p) = protocol {
+            self.protocol.store(Arc::new(p));
+        }
+    }
+
     #[inline]
     fn last_version(&self) -> ManifestVersion {
         self.version.load(Ordering::Relaxed)
@@ -158,19 +169,18 @@ impl RegionManifestInner {
         Ok(RegionMetaActionListIterator {
             log_iter: self.store.scan(start, end).await?,
             reader_version: self.supported_reader_version,
+            last_protocol: None,
         })
     }
 }
 
 #[cfg(test)]
 mod tests {
-    use datatypes::type_id::LogicalTypeId;
     use object_store::{backend::fs, ObjectStore};
     use tempdir::TempDir;
 
     use super::*;
-    use crate::metadata::RegionMetadata;
-    use crate::test_util::descriptor_util::RegionDescBuilder;
+    use crate::manifest::test_utils::*;
 
     #[tokio::test]
     async fn test_region_manifest() {
@@ -186,16 +196,16 @@ mod tests {
 
         let manifest = RegionManifest::new("/manifest/", object_store);
 
-        let region_name = "region-0";
-        let desc = RegionDescBuilder::new(region_name)
-            .id(0)
-            .push_key_column(("k1", LogicalTypeId::Int32, false))
-            .push_value_column(("v1", LogicalTypeId::Float32, true))
-            .build();
-        let metadata: RegionMetadata = desc.try_into().unwrap();
-        let region_meta = Arc::new(metadata);
+        let region_meta = Arc::new(build_region_meta());
 
-        assert!(manifest.load().await.unwrap().is_none());
+        assert!(manifest
+            .scan(0, MAX_VERSION)
+            .await
+            .unwrap()
+            .next_action()
+            .await
+            .unwrap()
+            .is_none());
 
         manifest
             .update(RegionMetaActionList::with_action(RegionMetaAction::Change(
@@ -206,30 +216,48 @@ mod tests {
             .await
             .unwrap();
 
-        let manifest_data = manifest.load().await.unwrap().unwrap();
-        assert_eq!(manifest_data.region_meta, region_meta);
+        let mut iter = manifest.scan(0, MAX_VERSION).await.unwrap();
 
-        // save another metadata
-        let region_name = "region-0";
-        let desc = RegionDescBuilder::new(region_name)
-            .id(0)
-            .push_key_column(("k1", LogicalTypeId::Int32, false))
-            .push_key_column(("k2", LogicalTypeId::Int64, false))
-            .push_value_column(("v1", LogicalTypeId::Float32, true))
-            .push_value_column(("bool", LogicalTypeId::Boolean, true))
-            .build();
-        let metadata: RegionMetadata = desc.try_into().unwrap();
-        let region_meta = Arc::new(metadata);
+        let (v, action_list) = iter.next_action().await.unwrap().unwrap();
+        assert_eq!(0, v);
+        assert_eq!(1, action_list.actions.len());
+        let action = &action_list.actions[0];
+
+        match action {
+            RegionMetaAction::Change(c) => {
+                assert_eq!(c.metadata, region_meta);
+            }
+            _ => unreachable!(),
+        }
+
+        // Save some actions
         manifest
-            .update(RegionMetaActionList::with_action(RegionMetaAction::Change(
-                RegionChange {
-                    metadata: region_meta.clone(),
-                },
-            )))
+            .update(RegionMetaActionList::new(vec![
+                RegionMetaAction::Edit(build_region_edit(1, &["f1"], &[])),
+                RegionMetaAction::Edit(build_region_edit(2, &["f2", "f3"], &[])),
+            ]))
             .await
             .unwrap();
 
-        let manifest_data = manifest.load().await.unwrap().unwrap();
-        assert_eq!(manifest_data.region_meta, region_meta);
+        let mut iter = manifest.scan(0, MAX_VERSION).await.unwrap();
+        let (v, action_list) = iter.next_action().await.unwrap().unwrap();
+        assert_eq!(0, v);
+        assert_eq!(1, action_list.actions.len());
+        let action = &action_list.actions[0];
+        match action {
+            RegionMetaAction::Change(c) => {
+                assert_eq!(c.metadata, region_meta);
+            }
+            _ => unreachable!(),
+        }
+
+        let (v, action_list) = iter.next_action().await.unwrap().unwrap();
+        assert_eq!(1, v);
+        assert_eq!(2, action_list.actions.len());
+        assert!(matches!(&action_list.actions[0], RegionMetaAction::Edit(_)));
+        assert!(matches!(&action_list.actions[1], RegionMetaAction::Edit(_)));
+
+        // Reach end
+        assert!(iter.next_action().await.unwrap().is_none());
     }
 }

src/storage/src/manifest/test_utils.rs

@@ -0,0 +1,42 @@
+use datatypes::type_id::LogicalTypeId;
+use store_api::storage::SequenceNumber;
+
+use crate::manifest::action::*;
+use crate::metadata::RegionMetadata;
+use crate::sst::FileMeta;
+use crate::test_util::descriptor_util::RegionDescBuilder;
+
+pub fn build_region_meta() -> RegionMetadata {
+    let region_name = "region-0";
+    let desc = RegionDescBuilder::new(region_name)
+        .id(0)
+        .push_key_column(("k1", LogicalTypeId::Int32, false))
+        .push_value_column(("v1", LogicalTypeId::Float32, true))
+        .build();
+    desc.try_into().unwrap()
+}
+
+pub fn build_region_edit(
+    sequence: SequenceNumber,
+    files_to_add: &[&str],
+    files_to_remove: &[&str],
+) -> RegionEdit {
+    RegionEdit {
+        region_version: 0,
+        flushed_sequence: sequence,
+        files_to_add: files_to_add
+            .iter()
+            .map(|f| FileMeta {
+                file_name: f.to_string(),
+                level: 0,
+            })
+            .collect(),
+        files_to_remove: files_to_remove
+            .iter()
+            .map(|f| FileMeta {
+                file_name: f.to_string(),
+                level: 0,
+            })
+            .collect(),
+    }
+}

src/storage/src/region.rs

@@ -6,21 +6,22 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use datatypes::schema::SchemaRef;
-use snafu::ensure;
+use snafu::{ensure, OptionExt};
 use store_api::logstore::LogStore;
-use store_api::manifest::Manifest;
+use store_api::manifest::{self, Manifest, ManifestVersion, MetaActionIterator};
 use store_api::storage::{
     OpenOptions, ReadContext, Region, RegionId, RegionMeta, WriteContext, WriteResponse,
 };
 
 use crate::error::{self, Error, Result};
 use crate::flush::{FlushSchedulerRef, FlushStrategyRef};
-use crate::manifest::region::RegionManifest;
+use crate::manifest::{action::RegionMetaAction, region::RegionManifest};
 use crate::memtable::MemtableBuilderRef;
 use crate::metadata::{RegionMetaImpl, RegionMetadata};
 pub use crate::region::writer::{RegionWriter, RegionWriterRef, WriterContext};
 use crate::snapshot::SnapshotImpl;
 use crate::sst::AccessLayerRef;
+use crate::version::VersionEdit;
 use crate::version::{Version, VersionControl, VersionControlRef};
 use crate::wal::Wal;
 use crate::write_batch::WriteBatch;
@@ -115,7 +116,7 @@ impl<S: LogStore> RegionImpl<S> {
         opts: &OpenOptions,
     ) -> Result<RegionImpl<S>> {
         // Load version meta data from manifest.
-        let version = Self::recover_from_manifest(&store_config.manifest).await?;
+        let version = Self::recover_from_manifest(&name, &store_config.manifest).await?;
         let metadata = version.metadata().clone();
         let version_control = Arc::new(VersionControl::with_version(version));
         let wal = Wal::new(name.clone(), store_config.log_store);
@@ -140,11 +141,76 @@ impl<S: LogStore> RegionImpl<S> {
         unimplemented!()
     }
 
-    async fn recover_from_manifest(manifest: &RegionManifest) -> Result<Version> {
-        let _metadata = manifest.load().await?;
+    async fn recover_from_manifest(
+        region_name: &str,
+        manifest: &RegionManifest,
+    ) -> Result<Version> {
+        let (start, end) = Self::manifest_scan_range();
+        let mut iter = manifest.scan(start, end).await?;
 
-        // TODO(yingwen): [open_region] Get version from metadata.
-        unimplemented!()
+        let mut version = None;
+        let mut actions = Vec::new();
+        let mut last_manifest_version = manifest::MIN_VERSION;
+        while let Some((manifest_version, action_list)) = iter.next_action().await? {
+            last_manifest_version = manifest_version;
+
+            for action in action_list.actions {
+                match (action, version) {
+                    (RegionMetaAction::Change(c), None) => {
+                        version = Some(Version::new(c.metadata));
+                        for (manifest_version, action) in actions.drain(..) {
+                            version = Self::replay_edit(manifest_version, action, version);
+                        }
+                    }
+                    (RegionMetaAction::Change(_), Some(_)) => {
+                        unimplemented!("alter schema is not implemented")
+                    }
+                    (action, None) => {
+                        actions.push((manifest_version, action));
+                        version = None;
+                    }
+                    (action, Some(v)) => {
+                        version = Self::replay_edit(manifest_version, action, Some(v));
+                    }
+                }
+            }
+        }
+
+        assert!(actions.is_empty() || version.is_none());
+
+        if version.is_some() {
+            // update manifest state after recovering
+            let protocol = iter.last_protocol();
+            manifest.update_state(last_manifest_version + 1, protocol.clone());
+        }
+
+        version.context(error::VersionNotFoundSnafu { region_name })
+    }
+
+    fn manifest_scan_range() -> (ManifestVersion, ManifestVersion) {
+        // TODO(dennis): use manifest version in WAL
+        (manifest::MIN_VERSION, manifest::MAX_VERSION)
+    }
+
+    fn replay_edit(
+        manifest_version: ManifestVersion,
+        action: RegionMetaAction,
+        version: Option<Version>,
+    ) -> Option<Version> {
+        if let RegionMetaAction::Edit(e) = action {
+            let edit = VersionEdit {
+                files_to_add: e.files_to_add,
+                flushed_sequence: Some(e.flushed_sequence),
+                manifest_version,
+                max_memtable_id: None,
+            };
+            version.map(|mut v| {
+                v.apply_edit(edit);
+                v
+            })
+        } else {
+            version
+        }
     }
 }
 

src/storage/src/region/tests.rs

@@ -4,10 +4,15 @@ mod flush;
 mod read_write;
 
 use datatypes::type_id::LogicalTypeId;
+use log_store::fs::noop::NoopLogStore;
+use object_store::{backend::fs, ObjectStore};
 use store_api::storage::consts;
 use tempdir::TempDir;
 
 use super::*;
+use crate::manifest::action::RegionChange;
+use crate::manifest::action::RegionMetaActionList;
+use crate::manifest::test_utils::*;
 use crate::test_util::{self, config_util, descriptor_util::RegionDescBuilder, schema_util};
 
 #[tokio::test]
@@ -43,3 +48,65 @@ async fn test_new_region() {
     assert_eq!(region_name, region.name());
     assert_eq!(expect_schema, *region.in_memory_metadata().schema());
 }
+
+#[tokio::test]
+async fn test_recover_region_manifets() {
+    let tmp_dir = TempDir::new("test_new_region").unwrap();
+
+    let object_store = ObjectStore::new(
+        fs::Backend::build()
+            .root(&tmp_dir.path().to_string_lossy())
+            .finish()
+            .await
+            .unwrap(),
+    );
+
+    let manifest = RegionManifest::new("/manifest/", object_store);
+    let region_name = "region-0";
+    let region_meta = Arc::new(build_region_meta());
+
+    // Recover from empty
+    assert!(
+        RegionImpl::<NoopLogStore>::recover_from_manifest(region_name, &manifest)
+            .await
+            .is_err()
+    );
+
+    {
+        // save some actions into region_meta
+        manifest
+            .update(RegionMetaActionList::with_action(RegionMetaAction::Change(
+                RegionChange {
+                    metadata: region_meta.clone(),
+                },
+            )))
+            .await
+            .unwrap();
+
+        manifest
+            .update(RegionMetaActionList::new(vec![
+                RegionMetaAction::Edit(build_region_edit(1, &["f1"], &[])),
+                RegionMetaAction::Edit(build_region_edit(2, &["f2", "f3"], &[])),
+            ]))
+            .await
+            .unwrap();
+    }
+
+    // try to recover
+    let version = RegionImpl::<NoopLogStore>::recover_from_manifest(region_name, &manifest)
+        .await
+        .unwrap();
+    assert_eq!(*version.metadata(), region_meta);
+    assert_eq!(version.flushed_sequence(), 2);
+    assert_eq!(version.manifest_version(), 1);
+    let ssts = version.ssts();
+    let files = ssts.levels()[0].files();
+    assert_eq!(3, files.len());
+    for (i, file) in files.iter().enumerate() {
+        assert_eq!(format!("f{}", i + 1), file.file_name());
+    }
+    assert!(version.mutable_memtables().is_empty());
+
+    // check manifest state
+    assert_eq!(2, manifest.last_version());
+}

src/storage/src/sst.rs

@@ -70,6 +70,11 @@ impl LevelMetas {
 
         Ok(())
     }
+
+    #[cfg(test)]
+    pub fn levels(&self) -> &[LevelMeta] {
+        &self.levels
+    }
 }
 
 impl Default for LevelMetas {
@@ -96,6 +101,11 @@ impl LevelMeta {
     fn visit_level<V: Visitor>(&self, visitor: &mut V) -> Result<()> {
         visitor.visit(self.level.into(), &self.files)
     }
+
+    #[cfg(test)]
+    pub fn files(&self) -> &[FileHandle] {
+        &self.files
+    }
 }
 
 fn new_level_meta_vec() -> LevelMetaVec {

src/storage/src/version.rs

@@ -39,7 +39,7 @@ impl VersionControl {
     /// Construct a new version control from `metadata`.
     pub fn new(metadata: RegionMetadata) -> VersionControl {
         VersionControl {
-            version: CowCell::new(Version::new(metadata)),
+            version: CowCell::new(Version::new(Arc::new(metadata))),
             committed_sequence: AtomicU64::new(0),
         }
     }
@@ -58,6 +58,11 @@ impl VersionControl {
         self.version.get()
     }
 
+    #[inline]
+    pub fn current_manifest_version(&self) -> ManifestVersion {
+        self.current().manifest_version
+    }
+
     /// Metadata of current version.
     pub fn metadata(&self) -> RegionMetadataRef {
         let version = self.current();
@@ -147,9 +152,9 @@ pub struct Version {
 }
 
 impl Version {
-    pub fn new(metadata: RegionMetadata) -> Version {
+    pub fn new(metadata: RegionMetadataRef) -> Version {
         Version {
-            metadata: Arc::new(metadata),
+            metadata,
             memtables: Arc::new(MemtableVersion::new()),
             ssts: Arc::new(LevelMetas::new()),
             flushed_sequence: 0,

src/store-api/src/manifest.rs

@@ -12,24 +12,38 @@ pub type ManifestVersion = u64;
 pub const MIN_VERSION: u64 = 0;
 pub const MAX_VERSION: u64 = u64::MAX;
 
-pub trait Metadata: Clone {}
-
 pub trait MetaAction: Serialize + DeserializeOwned {
     fn set_prev_version(&mut self, version: ManifestVersion);
 }
 
+#[async_trait]
+pub trait MetaActionIterator {
+    type MetaAction: MetaAction;
+    type Error: ErrorExt + Send + Sync;
+
+    async fn next_action(
+        &mut self,
+    ) -> Result<Option<(ManifestVersion, Self::MetaAction)>, Self::Error>;
+}
+
 /// Manifest service
 #[async_trait]
 pub trait Manifest: Send + Sync + Clone + 'static {
     type Error: ErrorExt + Send + Sync;
     type MetaAction: MetaAction;
-    type Metadata: Metadata;
+    type MetaActionIterator: MetaActionIterator<Error = Self::Error, MetaAction = Self::MetaAction>;
 
     /// Update metadata by the action
     async fn update(&self, action: Self::MetaAction) -> Result<ManifestVersion, Self::Error>;
 
-    /// Retrieve the latest metadata
-    async fn load(&self) -> Result<Option<Self::Metadata>, Self::Error>;
+    /// Scan actions which version in range [start, end)
+    async fn scan(
+        &self,
+        start: ManifestVersion,
+        end: ManifestVersion,
+    ) -> Result<Self::MetaActionIterator, Self::Error>;
 
     async fn checkpoint(&self) -> Result<ManifestVersion, Self::Error>;
+
+    fn last_version(&self) -> ManifestVersion;
 }