feat: refactor manifest to get snapshot in memory (#825)

* support return snapshot in `TableVersion`.

* introdce `TableSnapshotProvider` to get manifest snapshot data from memory.

* refactor `Snapshotter` to support from memory or storage.

* correct the order between do snapshot and write wal.

* fix test in manifest.

* let table snapshot provider return None when drop table.

* remove `origin_logs_num` from `Snapshot`.

* fix compile and clippy.

* split `Snapshotter` to `Snapshotter` and `SnapshotRecoverer`, remove the now unused `SnapshotBuilder`.

* move `TableSnapshotProviderImpl` to space module.

Author: Rachelint

analytic_engine/src/instance/engine.rs

@@ -205,6 +205,11 @@ pub enum Error {
         backtrace,
     ))]
     CreateOpenFailedTable { table: String, backtrace: Backtrace },
+
+    #[snafu(display("Failed to open manifest, err:{}", source))]
+    OpenManifest {
+        source: crate::manifest::details::Error,
+    },
 }
 
 define_result!(Error);
@@ -234,7 +239,8 @@ impl From<Error> for table_engine::engine::Error {
             | Error::StoreVersionEdit { .. }
             | Error::EncodePayloads { .. }
             | Error::CreateOpenFailedTable { .. }
-            | Error::DoManifestSnapshot { .. } => Self::Unexpected {
+            | Error::DoManifestSnapshot { .. }
+            | Error::OpenManifest { .. } => Self::Unexpected {
                 source: Box::new(err),
             },
         }

analytic_engine/src/instance/mod.rs

@@ -17,10 +17,7 @@ mod read;
 pub(crate) mod serial_executor;
 pub(crate) mod write;
 
-use std::{
-    collections::HashMap,
-    sync::{Arc, RwLock},
-};
+use std::sync::Arc;
 
 use common_types::table::TableId;
 use common_util::{
@@ -40,7 +37,7 @@ use crate::{
     compaction::{scheduler::CompactionSchedulerRef, TableCompactionRequest},
     manifest::ManifestRef,
     row_iter::IterOptions,
-    space::{SpaceId, SpaceRef},
+    space::{SpaceId, SpaceRef, SpacesRef},
     sst::{
         factory::{FactoryRef as SstFactoryRef, ObjectStorePickerRef, ScanOptions},
         file::FilePurger,
@@ -70,41 +67,9 @@ pub enum Error {
 
 define_result!(Error);
 
-/// Spaces states
-#[derive(Default)]
-struct Spaces {
-    /// Id to space
-    id_to_space: HashMap<SpaceId, SpaceRef>,
-}
-
-impl Spaces {
-    /// Insert space by name, and also insert id to space mapping
-    fn insert(&mut self, space: SpaceRef) {
-        let space_id = space.id;
-        self.id_to_space.insert(space_id, space);
-    }
-
-    fn get_by_id(&self, id: SpaceId) -> Option<&SpaceRef> {
-        self.id_to_space.get(&id)
-    }
-
-    /// List all tables of all spaces
-    fn list_all_tables(&self, tables: &mut Vec<TableDataRef>) {
-        let total_tables = self.id_to_space.values().map(|s| s.table_num()).sum();
-        tables.reserve(total_tables);
-        for space in self.id_to_space.values() {
-            space.list_all_tables(tables);
-        }
-    }
-
-    fn list_all_spaces(&self) -> Vec<SpaceRef> {
-        self.id_to_space.values().cloned().collect()
-    }
-}
-
 pub struct SpaceStore {
     /// All spaces of the engine.
-    spaces: RwLock<Spaces>,
+    spaces: SpacesRef,
     /// Manifest (or meta) stores meta data of the engine instance.
     manifest: ManifestRef,
     /// Wal of all tables

analytic_engine/src/instance/open.rs

@@ -9,6 +9,7 @@ use std::{
 
 use common_types::schema::IndexInWriterSchema;
 use log::{debug, error, info, trace};
+use object_store::ObjectStoreRef;
 use snafu::ResultExt;
 use table_engine::engine::OpenTableRequest;
 use wal::{
@@ -22,17 +23,20 @@ use crate::{
     engine,
     instance::{
         self,
-        engine::{ApplyMemTable, FlushTable, ReadMetaUpdate, ReadWal, RecoverTableData, Result},
+        engine::{
+            ApplyMemTable, FlushTable, OpenManifest, ReadMetaUpdate, ReadWal, RecoverTableData,
+            Result,
+        },
         flush_compaction::TableFlushOptions,
         mem_collector::MemUsageCollector,
         serial_executor::TableOpSerialExecutor,
         write::MemTableWriter,
-        Instance, SpaceStore, Spaces,
+        Instance, SpaceStore,
     },
-    manifest::{meta_data::TableManifestData, LoadRequest, ManifestRef},
+    manifest::{details::ManifestImpl, meta_data::TableManifestData, LoadRequest},
     payload::{ReadPayload, WalDecoder},
     row_iter::IterOptions,
-    space::{Space, SpaceContext, SpaceId, SpaceRef},
+    space::{Space, SpaceContext, SpaceId, SpaceRef, Spaces, TableSnapshotProviderImpl},
     sst::{
         factory::{FactoryRef as SstFactoryRef, ObjectStorePickerRef, ScanOptions},
         file::FilePurger,
@@ -42,18 +46,36 @@ use crate::{
 
 const MAX_RECORD_BATCHES_IN_FLIGHT_WHEN_COMPACTION_READ: usize = 64;
 
+pub(crate) struct ManifestStorages {
+    pub wal_manager: WalManagerRef,
+    pub oss_storage: ObjectStoreRef,
+}
+
 impl Instance {
     /// Open a new instance
-    pub async fn open(
+    pub(crate) async fn open(
         ctx: OpenContext,
-        manifest: ManifestRef,
+        manifest_storages: ManifestStorages,
         wal_manager: WalManagerRef,
         store_picker: ObjectStorePickerRef,
         sst_factory: SstFactoryRef,
     ) -> Result<Arc<Self>> {
+        let spaces: Arc<RwLock<Spaces>> = Arc::new(RwLock::new(Spaces::default()));
+        let table_snapshot_provider = Arc::new(TableSnapshotProviderImpl {
+            spaces: spaces.clone(),
+        });
+        let manifest = ManifestImpl::open(
+            ctx.config.manifest.clone(),
+            manifest_storages.wal_manager,
+            manifest_storages.oss_storage,
+            table_snapshot_provider,
+        )
+        .await
+        .context(OpenManifest)?;
+
         let space_store = Arc::new(SpaceStore {
-            spaces: RwLock::new(Spaces::default()),
-            manifest,
+            spaces,
+            manifest: Arc::new(manifest),
             wal_manager: wal_manager.clone(),
             store_picker: store_picker.clone(),
             sst_factory,