refactor(meta): introduce meta store ref and avoid pass generic parameter (#12114)

Author: wenym1

src/meta/src/backup_restore/backup_manager.rs

@@ -33,7 +33,6 @@ use crate::backup_restore::metrics::BackupManagerMetrics;
 use crate::hummock::{HummockManagerRef, HummockVersionSafePoint};
 use crate::manager::{IdCategory, LocalNotification, MetaSrvEnv};
 use crate::rpc::metrics::MetaMetrics;
-use crate::storage::MetaStore;
 use crate::MetaResult;
 
 pub enum BackupJobResult {
@@ -59,25 +58,25 @@ impl BackupJobHandle {
     }
 }
 
-pub type BackupManagerRef<S> = Arc<BackupManager<S>>;
+pub type BackupManagerRef = Arc<BackupManager>;
 /// (url, dir)
 type StoreConfig = (String, String);
 
 /// `BackupManager` manages lifecycle of all existent backups and the running backup job.
-pub struct BackupManager<S: MetaStore> {
-    env: MetaSrvEnv<S>,
-    hummock_manager: HummockManagerRef<S>,
+pub struct BackupManager {
+    env: MetaSrvEnv,
+    hummock_manager: HummockManagerRef,
     backup_store: ArcSwap<(BoxedMetaSnapshotStorage, StoreConfig)>,
     /// Tracks the running backup job. Concurrent jobs is not supported.
     running_backup_job: tokio::sync::Mutex<Option<BackupJobHandle>>,
     metrics: BackupManagerMetrics,
     meta_metrics: Arc<MetaMetrics>,
 }
 
-impl<S: MetaStore> BackupManager<S> {
+impl BackupManager {
     pub async fn new(
-        env: MetaSrvEnv<S>,
-        hummock_manager: HummockManagerRef<S>,
+        env: MetaSrvEnv,
+        hummock_manager: HummockManagerRef,
         metrics: Arc<MetaMetrics>,
         store_url: &str,
         store_dir: &str,
@@ -139,8 +138,8 @@ impl<S: MetaStore> BackupManager<S> {
     }
 
     fn with_store(
-        env: MetaSrvEnv<S>,
-        hummock_manager: HummockManagerRef<S>,
+        env: MetaSrvEnv,
+        hummock_manager: HummockManagerRef,
         meta_metrics: Arc<MetaMetrics>,
         backup_store: (BoxedMetaSnapshotStorage, StoreConfig),
     ) -> Self {
@@ -167,7 +166,7 @@ impl<S: MetaStore> BackupManager<S> {
     }
 
     #[cfg(test)]
-    pub fn for_test(env: MetaSrvEnv<S>, hummock_manager: HummockManagerRef<S>) -> Self {
+    pub fn for_test(env: MetaSrvEnv, hummock_manager: HummockManagerRef) -> Self {
         Self::with_store(
             env,
             hummock_manager,
@@ -326,12 +325,12 @@ impl<S: MetaStore> BackupManager<S> {
 }
 
 /// `BackupWorker` creates a database snapshot.
-struct BackupWorker<S: MetaStore> {
-    backup_manager: BackupManagerRef<S>,
+struct BackupWorker {
+    backup_manager: BackupManagerRef,
 }
 
-impl<S: MetaStore> BackupWorker<S> {
-    fn new(backup_manager: BackupManagerRef<S>) -> Self {
+impl BackupWorker {
+    fn new(backup_manager: BackupManagerRef) -> Self {
         Self { backup_manager }
     }
 

src/meta/src/backup_restore/meta_snapshot_builder.rs

@@ -14,7 +14,6 @@
 
 use std::collections::HashMap;
 use std::future::Future;
-use std::sync::Arc;
 
 use anyhow::anyhow;
 use risingwave_backup::error::{BackupError, BackupResult};
@@ -36,11 +35,11 @@ const VERSION: u32 = 1;
 
 pub struct MetaSnapshotBuilder<S> {
     snapshot: MetaSnapshot,
-    meta_store: Arc<S>,
+    meta_store: S,
 }
 
 impl<S: MetaStore> MetaSnapshotBuilder<S> {
-    pub fn new(meta_store: Arc<S>) -> Self {
+    pub fn new(meta_store: S) -> Self {
         Self {
             snapshot: MetaSnapshot::default(),
             meta_store,
@@ -161,8 +160,6 @@ impl<S: MetaStore> MetaSnapshotBuilder<S> {
 
 #[cfg(test)]
 mod tests {
-    use std::ops::Deref;
-    use std::sync::Arc;
 
     use assert_matches::assert_matches;
     use itertools::Itertools;
@@ -179,7 +176,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_snapshot_builder() {
-        let meta_store = Arc::new(MemStore::new());
+        let meta_store = MemStore::new();
 
         let mut builder = MetaSnapshotBuilder::new(meta_store.clone());
         let hummock_version = HummockVersion {
@@ -190,7 +187,7 @@ mod tests {
             let v_ = v.clone();
             async move { v_ }
         };
-        hummock_version.insert(meta_store.deref()).await.unwrap();
+        hummock_version.insert(&meta_store).await.unwrap();
         let err = builder
             .build(1, get_ckpt_builder(&hummock_version))
             .await
@@ -205,21 +202,15 @@ mod tests {
             hummock_version_id: hummock_version.id,
             ..Default::default()
         };
-        hummock_version_stats
-            .insert(meta_store.deref())
-            .await
-            .unwrap();
+        hummock_version_stats.insert(&meta_store).await.unwrap();
         let err = builder
             .build(1, get_ckpt_builder(&hummock_version))
             .await
             .unwrap_err();
         let err = assert_matches!(err, BackupError::Other(e) => e);
         assert_eq!("system params not found in meta store", err.to_error_str());
 
-        system_params_for_test()
-            .insert(meta_store.deref())
-            .await
-            .unwrap();
+        system_params_for_test().insert(&meta_store).await.unwrap();
 
         let err = builder
             .build(1, get_ckpt_builder(&hummock_version))
@@ -229,7 +220,7 @@ mod tests {
         assert_eq!("cluster id not found in meta store", err.to_error_str());
 
         ClusterId::new()
-            .put_at_meta_store(meta_store.deref())
+            .put_at_meta_store(&meta_store)
             .await
             .unwrap();
 

src/meta/src/barrier/command.rs

@@ -37,7 +37,6 @@ use super::trace::TracedEpoch;
 use crate::barrier::CommandChanges;
 use crate::manager::{FragmentManagerRef, WorkerId};
 use crate::model::{ActorId, DispatcherId, FragmentId, PausedReason, TableFragments};
-use crate::storage::MetaStore;
 use crate::stream::{build_actor_connector_splits, SourceManagerRef, SplitAssignment};
 use crate::MetaResult;
 
@@ -214,8 +213,8 @@ impl Command {
 
 /// [`CommandContext`] is used for generating barrier and doing post stuffs according to the given
 /// [`Command`].
-pub struct CommandContext<S: MetaStore> {
-    fragment_manager: FragmentManagerRef<S>,
+pub struct CommandContext {
+    fragment_manager: FragmentManagerRef,
 
     client_pool: StreamClientPoolRef,
 
@@ -232,7 +231,7 @@ pub struct CommandContext<S: MetaStore> {
 
     pub kind: BarrierKind,
 
-    source_manager: SourceManagerRef<S>,
+    source_manager: SourceManagerRef,
 
     /// The tracing span of this command.
     ///
@@ -242,18 +241,18 @@ pub struct CommandContext<S: MetaStore> {
     pub span: tracing::Span,
 }
 
-impl<S: MetaStore> CommandContext<S> {
+impl CommandContext {
     #[allow(clippy::too_many_arguments)]
     pub(super) fn new(
-        fragment_manager: FragmentManagerRef<S>,
+        fragment_manager: FragmentManagerRef,
         client_pool: StreamClientPoolRef,
         info: BarrierActorInfo,
         prev_epoch: TracedEpoch,
         curr_epoch: TracedEpoch,
         current_paused_reason: Option<PausedReason>,
         command: Command,
         kind: BarrierKind,
-        source_manager: SourceManagerRef<S>,
+        source_manager: SourceManagerRef,
         span: tracing::Span,
     ) -> Self {
         Self {
@@ -271,10 +270,7 @@ impl<S: MetaStore> CommandContext<S> {
     }
 }
 
-impl<S> CommandContext<S>
-where
-    S: MetaStore,
-{
+impl CommandContext {
     /// Generate a mutation for the given command.
     pub async fn to_mutation(&self) -> MetaResult<Option<Mutation>> {
         let mutation = match &self.command {