Merge branch 'main' into xxh/bigquery_retry

src/meta/node/src/server.rs

@@ -659,7 +659,7 @@ pub async fn start_service_as_election_leader(
     );
     let health_srv = HealthServiceImpl::new();
     let backup_srv = BackupServiceImpl::new(backup_manager);
-    let telemetry_srv = TelemetryInfoServiceImpl::new(env.meta_store_ref());
+    let telemetry_srv = TelemetryInfoServiceImpl::new(env.meta_store());
     let system_params_srv = SystemParamsServiceImpl::new(env.system_params_manager_impl_ref());
     let session_params_srv = SessionParamsServiceImpl::new(env.session_params_manager_impl_ref());
     let serving_srv =

src/meta/src/controller/session_params.rs

@@ -151,7 +151,7 @@ mod tests {
         use sea_orm::QueryFilter;
 
         let env = MetaSrvEnv::for_test_with_sql_meta_store().await;
-        let meta_store = env.meta_store().as_sql();
+        let meta_store = env.meta_store_ref().as_sql();
         let init_params = SessionConfig::default();
 
         // init system parameter controller as first launch.

src/meta/src/hummock/compactor_manager.rs

@@ -138,7 +138,7 @@ impl CompactorManagerInner {
         use risingwave_meta_model_v2::compaction_task;
         use sea_orm::EntityTrait;
         // Retrieve the existing task assignments from metastore.
-        let task_assignment: Vec<CompactTaskAssignment> = match env.meta_store() {
+        let task_assignment: Vec<CompactTaskAssignment> = match env.meta_store_ref() {
             MetaStoreImpl::Kv(meta_store) => CompactTaskAssignment::list(meta_store).await?,
             MetaStoreImpl::Sql(sql_meta_store) => compaction_task::Entity::find()
                 .all(&sql_meta_store.conn)

src/meta/src/hummock/manager/compaction.rs

@@ -704,15 +704,15 @@ impl HummockManager {
             // When the last table of a compaction group is deleted, the compaction group (and its
             // config) is destroyed as well. Then a compaction task for this group may come later and
             // cannot find its config.
-            let group_config = match self
-                .compaction_group_manager
-                .read()
-                .await
-                .try_get_compaction_group_config(compaction_group_id)
-            {
-                Some(config) => config,
-                None => continue,
+            let group_config = {
+                let config_manager = self.compaction_group_manager.read().await;
+
+                match config_manager.try_get_compaction_group_config(compaction_group_id) {
+                    Some(config) => config,
+                    None => continue,
+                }
             };
+
             // StoredIdGenerator already implements ids pre-allocation by ID_PREALLOCATE_INTERVAL.
             let task_id = next_compaction_task_id(&self.env).await?;