risingwave_meta/hummock/manager/
mod.rs

1// Copyright 2022 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::{BTreeMap, HashMap};
16use std::ops::{Deref, DerefMut};
17use std::sync::Arc;
18use std::sync::atomic::AtomicBool;
19
20use anyhow::anyhow;
21use bytes::Bytes;
22use futures::FutureExt;
23use itertools::Itertools;
24use parking_lot::lock_api::RwLock;
25use risingwave_common::catalog::{TableId, TableOption};
26use risingwave_common::monitor::MonitoredRwLock;
27use risingwave_common::system_param::reader::SystemParamsRead;
28use risingwave_hummock_sdk::change_log::TableChangeLog;
29use risingwave_hummock_sdk::version::{HummockVersion, HummockVersionDelta};
30use risingwave_hummock_sdk::{
31    CompactionGroupId, HummockCompactionTaskId, HummockContextId, HummockVersionId,
32    version_archive_dir, version_checkpoint_path,
33};
34use risingwave_meta_model::{
35    compaction_status, compaction_task, hummock_pinned_version, hummock_version_delta,
36    hummock_version_stats,
37};
38use risingwave_pb::hummock::compact_task::TaskStatus;
39use risingwave_pb::hummock::{
40    HummockVersionStats, PbCompactionGroupInfo, SubscribeCompactionEventRequest,
41};
42use table_write_throughput_statistic::TableWriteThroughputStatisticManager;
43use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel};
44use tokio::sync::{Mutex, Semaphore, oneshot};
45use tonic::Streaming;
46
47use crate::MetaResult;
48use crate::hummock::CompactorManagerRef;
49use crate::hummock::compaction::CompactStatus;
50use crate::hummock::error::Result;
51use crate::hummock::manager::checkpoint::HummockVersionCheckpoint;
52use crate::hummock::manager::context::ContextInfo;
53use crate::hummock::manager::gc::{FullGcState, GcManager};
54use crate::hummock::manager::sequence::PrefetchedSequence;
55use crate::hummock::model::ext::{compaction_task_model_to_assignment, to_table_change_log};
56use crate::manager::{MetaSrvEnv, MetadataManager};
57use crate::model::{ClusterId, MetadataModelError};
58use crate::rpc::metrics::MetaMetrics;
59
60mod context;
61mod gc;
62mod tests;
63mod versioning;
64pub use context::HummockVersionSafePoint;
65use versioning::*;
66pub(crate) mod checkpoint;
67mod commit_epoch;
68mod compaction;
69pub mod sequence;
70mod table_change_log;
71pub use table_change_log::CommittedTableChangeLogProgress;
72pub mod table_write_throughput_statistic;
73pub mod time_travel;
74mod timer_task;
75mod transaction;
76mod utils;
77mod worker;
78
79pub use commit_epoch::{CommitEpochInfo, NewTableFragmentInfo};
80pub use compaction::compaction_event_loop::*;
81use compaction::*;
82pub use compaction::{
83    GroupState, GroupStateValidator, ManualCompactionTriggerResult, ScheduleTrigger,
84};
85pub(crate) use utils::*;
86
87struct TableCommittedEpochNotifiers {
88    txs: HashMap<TableId, Vec<UnboundedSender<u64>>>,
89}
90
91impl TableCommittedEpochNotifiers {
92    fn notify_deltas(&mut self, deltas: &[HummockVersionDelta]) {
93        self.txs.retain(|table_id, txs| {
94            let mut is_dropped = false;
95            let mut committed_epoch = None;
96            for delta in deltas {
97                if delta.removed_table_ids.contains(table_id) {
98                    is_dropped = true;
99                    break;
100                }
101                if let Some(info) = delta.state_table_info_delta.get(table_id) {
102                    committed_epoch = Some(info.committed_epoch);
103                }
104            }
105            if is_dropped {
106                false
107            } else if let Some(committed_epoch) = committed_epoch {
108                txs.retain(|tx| tx.send(committed_epoch).is_ok());
109                !txs.is_empty()
110            } else {
111                true
112            }
113        })
114    }
115}
116
117#[derive(Clone, Debug)]
118struct CompactionTaskReportResult {
119    task_id: HummockCompactionTaskId,
120
121    task_status: TaskStatus,
122    reported: bool,
123}
124
125struct CompactionTaskReportNotifiers {
126    txs: HashMap<HummockCompactionTaskId, Vec<oneshot::Sender<CompactionTaskReportResult>>>,
127}
128
129impl CompactionTaskReportNotifiers {
130    fn register(
131        &mut self,
132        task_id: HummockCompactionTaskId,
133        tx: oneshot::Sender<CompactionTaskReportResult>,
134    ) {
135        self.txs.entry(task_id).or_default().push(tx);
136    }
137
138    fn remove(&mut self, task_id: HummockCompactionTaskId) {
139        self.txs.remove(&task_id);
140    }
141
142    fn notify(&mut self, result: CompactionTaskReportResult) {
143        if let Some(txs) = self.txs.remove(&result.task_id) {
144            for tx in txs {
145                let _ = tx.send(result.clone());
146            }
147        }
148    }
149}
150// Update to states are performed as follow:
151// - Initialize ValTransaction for the meta state to update
152// - Make changes on the ValTransaction.
153// - Call `commit_multi_var` to commit the changes via meta store transaction. If transaction
154//   succeeds, the in-mem state will be updated by the way.
155pub struct HummockManager {
156    pub env: MetaSrvEnv,
157
158    metadata_manager: MetadataManager,
159    /// Lock order: `compaction`, `versioning`, `compaction_group_manager`, `context_info`
160    /// - Lock `compaction` first, then `versioning`, then `compaction_group_manager` and finally `context_info`.
161    /// - This order should be strictly followed to prevent deadlock.
162    compaction: MonitoredRwLock<Compaction>,
163    versioning: MonitoredRwLock<Versioning>,
164    /// `CompactionGroupManager` manages compaction configs for compaction groups.
165    compaction_group_manager: MonitoredRwLock<CompactionGroupManager>,
166    context_info: MonitoredRwLock<ContextInfo>,
167
168    pub metrics: Arc<MetaMetrics>,
169
170    pub compactor_manager: CompactorManagerRef,
171    pub iceberg_compactor_manager: Arc<IcebergCompactorManager>,
172    event_sender: HummockManagerEventSender,
173    object_store: ObjectStoreRef,
174    version_checkpoint_path: String,
175    version_archive_dir: String,
176    pause_version_checkpoint: AtomicBool,
177    table_write_throughput_statistic_manager:
178        parking_lot::RwLock<TableWriteThroughputStatisticManager>,
179    table_committed_epoch_notifiers: parking_lot::Mutex<TableCommittedEpochNotifiers>,
180    compaction_task_report_notifiers: parking_lot::Mutex<CompactionTaskReportNotifiers>,
181    version_stat_tx: UnboundedSender<Arc<HummockVersion>>,
182
183    // for compactor
184    // `compactor_streams_change_tx` is used to pass the mapping from `context_id` to event_stream
185    // and is maintained in memory. All event_streams are consumed through a separate event loop
186    compactor_streams_change_tx:
187        UnboundedSender<(HummockContextId, Streaming<SubscribeCompactionEventRequest>)>,
188
189    // `compaction_state` will record the types of compact tasks that can be triggered in `hummock`
190    // and suggest types with a certain priority.
191    pub(super) compaction_state: CompactionState,
192    full_gc_state: Arc<FullGcState>,
193    now: Mutex<u64>,
194    inflight_time_travel_query: Semaphore,
195    gc_manager: GcManager,
196
197    /// In-memory cache of prefetched compaction task ids to reduce per-task DB round-trips.
198    prefetched_compaction_task_ids: PrefetchedSequence,
199    table_id_to_table_option: parking_lot::RwLock<HashMap<TableId, TableOption>>,
200    /// Progress table id to its upstream table id and latest safe truncate epoch.
201    table_change_log_progress: parking_lot::RwLock<HashMap<TableId, (TableId, u64)>>,
202}
203
204pub type HummockManagerRef = Arc<HummockManager>;
205
206use risingwave_object_store::object::{ObjectError, ObjectStoreRef, build_remote_object_store};
207use risingwave_pb::catalog::Table;
208
209use super::IcebergCompactorManager;
210use crate::controller::SqlMetaStore;
211use crate::hummock::manager::compaction_group_manager::CompactionGroupManager;
212use crate::hummock::manager::worker::HummockManagerEventSender;
213
214impl HummockManager {
215    pub async fn new(
216        env: MetaSrvEnv,
217        metadata_manager: MetadataManager,
218        metrics: Arc<MetaMetrics>,
219        compactor_manager: CompactorManagerRef,
220        compactor_streams_change_tx: UnboundedSender<(
221            HummockContextId,
222            Streaming<SubscribeCompactionEventRequest>,
223        )>,
224    ) -> Result<HummockManagerRef> {
225        let compaction_group_manager = CompactionGroupManager::new(&env).await?;
226        Self::new_impl(
227            env,
228            metadata_manager,
229            metrics,
230            compactor_manager,
231            compaction_group_manager,
232            compactor_streams_change_tx,
233        )
234        .await
235    }
236
237    #[cfg(any(test, feature = "test"))]
238    pub(super) async fn with_config(
239        env: MetaSrvEnv,
240        cluster_controller: crate::controller::cluster::ClusterControllerRef,
241        catalog_controller: crate::controller::catalog::CatalogControllerRef,
242        metrics: Arc<MetaMetrics>,
243        compactor_manager: CompactorManagerRef,
244        config: risingwave_pb::hummock::CompactionConfig,
245        compactor_streams_change_tx: UnboundedSender<(
246            HummockContextId,
247            Streaming<SubscribeCompactionEventRequest>,
248        )>,
249    ) -> HummockManagerRef {
250        let compaction_group_manager = CompactionGroupManager::new_with_config(&env, config)
251            .await
252            .unwrap();
253        let metadata_manager = MetadataManager::new(cluster_controller, catalog_controller);
254        Self::new_impl(
255            env,
256            metadata_manager,
257            metrics,
258            compactor_manager,
259            compaction_group_manager,
260            compactor_streams_change_tx,
261        )
262        .await
263        .unwrap()
264    }
265
266    async fn new_impl(
267        env: MetaSrvEnv,
268        metadata_manager: MetadataManager,
269        metrics: Arc<MetaMetrics>,
270        compactor_manager: CompactorManagerRef,
271        compaction_group_manager: CompactionGroupManager,
272        compactor_streams_change_tx: UnboundedSender<(
273            HummockContextId,
274            Streaming<SubscribeCompactionEventRequest>,
275        )>,
276    ) -> Result<HummockManagerRef> {
277        let sys_params = env.system_params_reader().await;
278        let state_store_url = sys_params.state_store();
279        let state_store_url = state_store_url.expose();
280
281        let state_store_dir: &str = sys_params.data_directory();
282        let use_new_object_prefix_strategy: bool = sys_params.use_new_object_prefix_strategy();
283        let deterministic_mode = env.opts.compaction_deterministic_test;
284        let mut object_store_config = env.opts.object_store_config.clone();
285        // For fs and hdfs object store, operations are not always atomic.
286        // We should manually enable atomicity guarantee by setting the atomic_write_dir config when building services.
287        object_store_config.set_atomic_write_dir();
288        let object_store = Arc::new(
289            build_remote_object_store(
290                state_store_url.strip_prefix("hummock+").unwrap_or("memory"),
291                metrics.object_store_metric.clone(),
292                "Version Checkpoint",
293                Arc::new(object_store_config),
294            )
295            .await,
296        );
297        // Make sure data dir is not used by another cluster.
298        // Skip this check in e2e compaction test, which needs to start a secondary cluster with
299        // same bucket
300        if !deterministic_mode {
301            write_exclusive_cluster_id(
302                state_store_dir,
303                env.cluster_id().clone(),
304                object_store.clone(),
305            )
306            .await?;
307
308            // config bucket lifecycle for new cluster.
309            if let risingwave_object_store::object::ObjectStoreImpl::S3(s3) = object_store.as_ref()
310                && !env.opts.do_not_config_object_storage_lifecycle
311            {
312                let is_bucket_expiration_configured =
313                    s3.inner().configure_bucket_lifecycle(state_store_dir).await;
314                if is_bucket_expiration_configured {
315                    return Err(ObjectError::internal("Cluster cannot start with object expiration configured for bucket because RisingWave data will be lost when object expiration kicks in.
316                    Please disable object expiration and restart the cluster.")
317                    .into());
318                }
319            }
320        }
321        let version_checkpoint_path = version_checkpoint_path(state_store_dir);
322        let version_archive_dir = version_archive_dir(state_store_dir);
323        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
324        let (version_stat_tx, mut version_stat_rx) = tokio::sync::mpsc::unbounded_channel();
325        let inflight_time_travel_query = env.opts.max_inflight_time_travel_query;
326        let gc_manager = GcManager::new(
327            object_store.clone(),
328            state_store_dir,
329            use_new_object_prefix_strategy,
330        );
331
332        let table_statistic_retention = env.opts.table_write_throughput_retention_seconds;
333
334        let iceberg_compactor_manager = Arc::new(IcebergCompactorManager::new());
335
336        let instance = HummockManager {
337            env,
338            versioning: MonitoredRwLock::new(
339                metrics.hummock_manager_lock_time.clone(),
340                metrics.hummock_manager_real_process_time.clone(),
341                Default::default(),
342                "hummock_manager::versioning",
343            ),
344            compaction: MonitoredRwLock::new(
345                metrics.hummock_manager_lock_time.clone(),
346                metrics.hummock_manager_real_process_time.clone(),
347                Default::default(),
348                "hummock_manager::compaction",
349            ),
350            compaction_group_manager: MonitoredRwLock::new(
351                metrics.hummock_manager_lock_time.clone(),
352                metrics.hummock_manager_real_process_time.clone(),
353                compaction_group_manager,
354                "hummock_manager::compaction_group_manager",
355            ),
356            context_info: MonitoredRwLock::new(
357                metrics.hummock_manager_lock_time.clone(),
358                metrics.hummock_manager_real_process_time.clone(),
359                Default::default(),
360                "hummock_manager::context_info",
361            ),
362            metrics,
363            metadata_manager,
364            compactor_manager,
365            iceberg_compactor_manager,
366            event_sender: tx,
367            object_store,
368            version_checkpoint_path,
369            version_archive_dir,
370            pause_version_checkpoint: AtomicBool::new(false),
371            table_write_throughput_statistic_manager: parking_lot::RwLock::new(
372                TableWriteThroughputStatisticManager::new(table_statistic_retention),
373            ),
374            table_committed_epoch_notifiers: parking_lot::Mutex::new(
375                TableCommittedEpochNotifiers {
376                    txs: Default::default(),
377                },
378            ),
379            compaction_task_report_notifiers: parking_lot::Mutex::new(
380                CompactionTaskReportNotifiers {
381                    txs: Default::default(),
382                },
383            ),
384            version_stat_tx,
385            compactor_streams_change_tx,
386            compaction_state: CompactionState::new(),
387            full_gc_state: FullGcState::new().into(),
388            now: Mutex::new(0),
389            inflight_time_travel_query: Semaphore::new(inflight_time_travel_query as usize),
390            gc_manager,
391            prefetched_compaction_task_ids: PrefetchedSequence::new(),
392            table_id_to_table_option: RwLock::new(HashMap::new()),
393            table_change_log_progress: RwLock::new(HashMap::new()),
394        };
395        let instance = Arc::new(instance);
396        let version_stat_metrics = instance.metrics.clone();
397        tokio::spawn(async move {
398            while let Some(mut version) = version_stat_rx.recv().await {
399                while let Some(Some(next_version)) = version_stat_rx.recv().now_or_never() {
400                    version = next_version;
401                }
402                transaction::trigger_version_stat(&version_stat_metrics, version.as_ref());
403            }
404        });
405        instance.init_time_travel_state().await?;
406        instance.load_meta_store_state().await?;
407        instance.start_worker(rx);
408        instance.release_invalid_contexts().await?;
409        // Release snapshots pinned by meta on restarting.
410        instance.release_meta_context().await?;
411        Ok(instance)
412    }
413
414    fn meta_store_ref(&self) -> &SqlMetaStore {
415        self.env.meta_store_ref()
416    }
417
418    /// Load state from meta store.
419    async fn load_meta_store_state(&self) -> Result<()> {
420        let now = self.load_now().await?;
421        *self.now.lock().await = now.unwrap_or(0);
422
423        let mut compaction_guard = self
424            .compaction
425            .write_with_process_name("load_meta_store_state")
426            .await;
427        let mut versioning_guard = self
428            .versioning
429            .write_with_process_name("load_meta_store_state")
430            .await;
431        let mut context_info_guard = self
432            .context_info
433            .write_with_process_name("load_meta_store_state")
434            .await;
435        self.load_meta_store_state_impl(
436            &mut compaction_guard,
437            &mut versioning_guard,
438            &mut context_info_guard,
439        )
440        .await
441    }
442
443    /// Load state from meta store.
444    async fn load_meta_store_state_impl(
445        &self,
446        compaction_guard: &mut Compaction,
447        versioning_guard: &mut Versioning,
448        context_info: &mut ContextInfo,
449    ) -> Result<()> {
450        use sea_orm::EntityTrait;
451        let meta_store = self.meta_store_ref();
452        let compaction_statuses: BTreeMap<CompactionGroupId, CompactStatus> =
453            compaction_status::Entity::find()
454                .all(&meta_store.conn)
455                .await
456                .map_err(MetadataModelError::from)?
457                .into_iter()
458                .map(|m| (m.compaction_group_id as CompactionGroupId, m.into()))
459                .collect();
460        if !compaction_statuses.is_empty() {
461            compaction_guard.compaction_statuses = compaction_statuses;
462        }
463
464        compaction_guard.compact_task_assignment = compaction_task::Entity::find()
465            .all(&meta_store.conn)
466            .await
467            .map_err(MetadataModelError::from)?
468            .into_iter()
469            .map(|m| {
470                (
471                    m.id as HummockCompactionTaskId,
472                    compaction_task_model_to_assignment(m),
473                )
474            })
475            .collect();
476
477        let hummock_version_deltas: BTreeMap<HummockVersionId, HummockVersionDelta> =
478            hummock_version_delta::Entity::find()
479                .all(&meta_store.conn)
480                .await
481                .map_err(MetadataModelError::from)?
482                .into_iter()
483                .map(|m| {
484                    (
485                        m.id,
486                        HummockVersionDelta::from_persisted_protobuf_owned(m.into()),
487                    )
488                })
489                .collect();
490
491        let checkpoint = self.try_read_checkpoint().await?;
492        let mut redo_state = if let Some(c) = checkpoint {
493            versioning_guard.checkpoint = c;
494            versioning_guard.checkpoint.version.as_ref().clone()
495        } else {
496            let default_compaction_config = self
497                .compaction_group_manager
498                .read_with_process_name("load_meta_store_state")
499                .await
500                .default_compaction_config();
501            let checkpoint_version = HummockVersion::create_init_version(default_compaction_config);
502            tracing::info!("init hummock version checkpoint");
503            versioning_guard.checkpoint = HummockVersionCheckpoint {
504                version: Arc::new(checkpoint_version.clone()),
505                stale_objects: Default::default(),
506            };
507            self.write_checkpoint(&versioning_guard.checkpoint).await?;
508            checkpoint_version
509        };
510        let mut applied_delta_count = 0;
511        let total_to_apply = hummock_version_deltas.range(redo_state.id + 1..).count();
512        tracing::info!(
513            total_delta = hummock_version_deltas.len(),
514            total_to_apply,
515            "Start redo Hummock version."
516        );
517        for version_delta in hummock_version_deltas
518            .range(redo_state.id + 1..)
519            .map(|(_, v)| v)
520        {
521            assert_eq!(
522                version_delta.prev_id, redo_state.id,
523                "delta prev_id {}, redo state id {}",
524                version_delta.prev_id, redo_state.id
525            );
526            redo_state.apply_version_delta(version_delta);
527            applied_delta_count += 1;
528            if applied_delta_count % 1000 == 0 {
529                tracing::info!("Redo progress {applied_delta_count}/{total_to_apply}.");
530            }
531        }
532        tracing::info!("Finish redo Hummock version.");
533        let pruned_stale_table_id_count = redo_state.prune_stale_table_ids_from_ssts();
534        if pruned_stale_table_id_count > 0 {
535            tracing::warn!(
536                pruned_stale_table_id_count,
537                version_id = ?redo_state.id,
538                "Pruned stale table ids from recovered Hummock SST metadata."
539            );
540        }
541        versioning_guard.version_stats = hummock_version_stats::Entity::find()
542            .one(&meta_store.conn)
543            .await
544            .map_err(MetadataModelError::from)?
545            .map(HummockVersionStats::from)
546            .unwrap_or_else(|| HummockVersionStats {
547                // version_stats.hummock_version_id is always 0 in meta store.
548                hummock_version_id: 0.into(),
549                ..Default::default()
550            });
551
552        versioning_guard.current_version = Arc::new(redo_state);
553        versioning_guard.hummock_version_deltas = hummock_version_deltas;
554        versioning_guard.table_change_log =
555            risingwave_meta_model::hummock_table_change_log::Entity::find()
556                .all(&self.env.meta_store_ref().conn)
557                .await
558                .map_err(MetadataModelError::from)?
559                .into_iter()
560                .map(|m| (m.table_id, to_table_change_log(m)))
561                .into_group_map()
562                .into_iter()
563                .map(|(table_id, unordered_change_logs)| {
564                    (
565                        table_id,
566                        TableChangeLog::new(
567                            unordered_change_logs
568                                .into_iter()
569                                .sorted_by_key(|l| l.checkpoint_epoch),
570                        ),
571                    )
572                })
573                .collect();
574
575        context_info.pinned_versions = hummock_pinned_version::Entity::find()
576            .all(&meta_store.conn)
577            .await
578            .map_err(MetadataModelError::from)?
579            .into_iter()
580            .map(|m| (m.context_id as HummockContextId, m.into()))
581            .collect();
582
583        self.initial_compaction_group_config_after_load(
584            versioning_guard,
585            self.compaction_group_manager
586                .write_with_process_name("load_meta_store_state")
587                .await
588                .deref_mut(),
589        )
590        .await?;
591
592        Ok(())
593    }
594
595    pub fn init_metadata_for_version_replay(
596        &self,
597        _table_catalogs: Vec<Table>,
598        _compaction_groups: Vec<PbCompactionGroupInfo>,
599    ) -> Result<()> {
600        unimplemented!("kv meta store is deprecated");
601    }
602
603    /// Replay a version delta to current hummock version.
604    /// Returns the `version_id`, `max_committed_epoch` of the new version and the modified
605    /// compaction groups
606    #[cfg(any(test, feature = "test"))]
607    pub async fn replay_version_delta(
608        &self,
609        mut version_delta: HummockVersionDelta,
610    ) -> Result<(HummockVersion, Vec<CompactionGroupId>)> {
611        let mut versioning_guard = self
612            .versioning
613            .write_with_process_name("replay_version_delta")
614            .await;
615        // ensure the version id is ascending after replay
616        version_delta.id = versioning_guard.current_version.next_version_id();
617        version_delta.prev_id = versioning_guard.current_version.id;
618        let mut version_new = versioning_guard.current_version.as_ref().clone();
619        version_new.apply_version_delta(&version_delta);
620
621        let compaction_group_ids = version_delta.group_deltas.keys().cloned().collect();
622        versioning_guard.current_version = Arc::new(version_new.clone());
623        Ok((version_new, compaction_group_ids))
624    }
625
626    pub async fn disable_commit_epoch(&self) -> Arc<HummockVersion> {
627        let mut versioning_guard = self
628            .versioning
629            .write_with_process_name("disable_commit_epoch")
630            .await;
631        versioning_guard.disable_commit_epochs = true;
632        versioning_guard.current_version.clone()
633    }
634
635    pub fn metadata_manager(&self) -> &MetadataManager {
636        &self.metadata_manager
637    }
638
639    pub fn object_store_media_type(&self) -> &'static str {
640        self.object_store.media_type()
641    }
642
643    pub fn update_table_id_to_table_option(
644        &self,
645        new_table_id_to_table_option: HashMap<TableId, TableOption>,
646    ) {
647        *self.table_id_to_table_option.write() = new_table_id_to_table_option;
648    }
649
650    pub fn metadata_manager_ref(&self) -> &MetadataManager {
651        &self.metadata_manager
652    }
653
654    pub async fn subscribe_table_committed_epoch(
655        &self,
656        table_id: TableId,
657    ) -> MetaResult<(u64, UnboundedReceiver<u64>)> {
658        let version = self
659            .versioning
660            .read_with_process_name("subscribe_table_committed_epoch")
661            .await;
662        if let Some(epoch) = version.current_version.table_committed_epoch(table_id) {
663            let (tx, rx) = unbounded_channel();
664            self.table_committed_epoch_notifiers
665                .lock()
666                .txs
667                .entry(table_id)
668                .or_default()
669                .push(tx);
670            Ok((epoch, rx))
671        } else {
672            Err(anyhow!("table {} does not exist", table_id).into())
673        }
674    }
675}
676
677async fn write_exclusive_cluster_id(
678    state_store_dir: &str,
679    cluster_id: ClusterId,
680    object_store: ObjectStoreRef,
681) -> Result<()> {
682    const CLUSTER_ID_DIR: &str = "cluster_id";
683    const CLUSTER_ID_NAME: &str = "0";
684    let cluster_id_dir = format!("{}/{}/", state_store_dir, CLUSTER_ID_DIR);
685    let cluster_id_full_path = format!("{}{}", cluster_id_dir, CLUSTER_ID_NAME);
686    tracing::info!("try reading cluster_id");
687    match object_store.read(&cluster_id_full_path, ..).await {
688        Ok(stored_cluster_id) => {
689            let stored_cluster_id = String::from_utf8(stored_cluster_id.to_vec()).unwrap();
690            if cluster_id.deref() == stored_cluster_id {
691                return Ok(());
692            }
693
694            Err(ObjectError::internal(format!(
695                "Data directory is already used by another cluster with id {:?}, path {}.",
696                stored_cluster_id, cluster_id_full_path,
697            ))
698            .into())
699        }
700        Err(e) => {
701            if e.is_object_not_found_error() {
702                tracing::info!("cluster_id not found, writing cluster_id");
703                object_store
704                    .upload(&cluster_id_full_path, Bytes::from(String::from(cluster_id)))
705                    .await?;
706                return Ok(());
707            }
708            Err(e.into())
709        }
710    }
711}
risingwave_meta/hummock/manager/mod.rs

risingwave_meta/hummock/manager/
mod.rs
