refactor(meta): simplify Hummock metadata initialization (#14259)

risingwavelabs · Jan 4, 2024 · 2824bdf · 2824bdf
1 parent c8cdb9f
commit 2824bdf
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 42 deletions.
diff --git a/src/meta/src/hummock/manager/checkpoint.rs b/src/meta/src/hummock/manager/checkpoint.rs
@@ -30,9 +30,6 @@ use crate::hummock::error::Result;
 use crate::hummock::manager::{read_lock, write_lock};
 use crate::hummock::metrics_utils::trigger_gc_stat;
 use crate::hummock::HummockManager;
-use crate::storage::{MetaStore, MetaStoreError, DEFAULT_COLUMN_FAMILY};
-
-const HUMMOCK_INIT_FLAG_KEY: &[u8] = b"hummock_init_flag";
 
 #[derive(Default)]
 pub struct HummockVersionCheckpoint {
@@ -63,9 +60,8 @@ impl HummockVersionCheckpoint {
 /// A hummock version checkpoint compacts previous hummock version delta logs, and stores stale
 /// objects from those delta logs.
 impl HummockManager {
-    /// # Panics
-    /// if checkpoint is not found.
-    pub async fn read_checkpoint(&self) -> Result<HummockVersionCheckpoint> {
+    /// Returns Ok(None) if not found.
+    pub async fn try_read_checkpoint(&self) -> Result<Option<HummockVersionCheckpoint>> {
         use prost::Message;
         let data = match self
             .object_store
@@ -75,16 +71,13 @@ impl HummockManager {
             Ok(data) => data,
             Err(e) => {
                 if e.is_object_not_found_error() {
-                    panic!(
-                        "Hummock version checkpoints do not exist in object store, path: {}",
-                        self.version_checkpoint_path
-                    );
+                    return Ok(None);
                 }
                 return Err(e.into());
             }
         };
         let ckpt = PbHummockVersionCheckpoint::decode(data).map_err(|e| anyhow::anyhow!(e))?;
-        Ok(HummockVersionCheckpoint::from_protobuf(&ckpt))
+        Ok(Some(HummockVersionCheckpoint::from_protobuf(&ckpt)))
     }
 
     pub(super) async fn write_checkpoint(
@@ -173,31 +166,6 @@ impl HummockManager {
         Ok(new_checkpoint_id - old_checkpoint_id)
     }
 
-    pub(super) async fn need_init(&self) -> Result<bool> {
-        match self
-            .env
-            .meta_store()
-            .get_cf(DEFAULT_COLUMN_FAMILY, HUMMOCK_INIT_FLAG_KEY)
-            .await
-        {
-            Ok(_) => Ok(false),
-            Err(MetaStoreError::ItemNotFound(_)) => Ok(true),
-            Err(e) => Err(e.into()),
-        }
-    }
-
-    pub(super) async fn mark_init(&self) -> Result<()> {
-        self.env
-            .meta_store()
-            .put_cf(
-                DEFAULT_COLUMN_FAMILY,
-                HUMMOCK_INIT_FLAG_KEY.to_vec(),
-                memcomparable::to_vec(&0).unwrap(),
-            )
-            .await
-            .map_err(Into::into)
-    }
-
     pub fn pause_version_checkpoint(&self) {
         self.pause_version_checkpoint.store(true, Ordering::Relaxed);
         tracing::info!("hummock version checkpoint is paused.");

diff --git a/src/meta/src/hummock/manager/mod.rs b/src/meta/src/hummock/manager/mod.rs
@@ -479,14 +479,19 @@ impl HummockManager {
                 .map(|version_delta| (version_delta.id, version_delta))
                 .collect();
 
-        let mut redo_state = if self.need_init().await? {
+        let checkpoint = self.try_read_checkpoint().await?;
+        let mut redo_state = if let Some(c) = checkpoint {
+            versioning_guard.checkpoint = c;
+            versioning_guard.checkpoint.version.clone()
+        } else {
             let default_compaction_config = self
                 .compaction_group_manager
                 .read()
                 .await
                 .default_compaction_config();
             let checkpoint_version = create_init_version(default_compaction_config);
             tracing::info!("init hummock version checkpoint");
+            // This write to meta store is idempotent. So if `write_checkpoint` fails, restarting meta node is fine.
             HummockVersionStats::default()
                 .insert(self.env.meta_store())
                 .await?;
@@ -495,13 +500,9 @@ impl HummockManager {
                 stale_objects: Default::default(),
             };
             self.write_checkpoint(&versioning_guard.checkpoint).await?;
-            self.mark_init().await?;
             checkpoint_version
-        } else {
-            // Read checkpoint from object store.
-            versioning_guard.checkpoint = self.read_checkpoint().await?;
-            versioning_guard.checkpoint.version.clone()
         };
+
         versioning_guard.version_stats = HummockVersionStats::list(self.env.meta_store())
             .await?
             .into_iter()