risingwave_meta/backup_restore/
meta_snapshot_builder.rs

1// Copyright 2025 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::future::Future;
16
17use itertools::Itertools;
18use risingwave_backup::MetaSnapshotId;
19use risingwave_backup::error::{BackupError, BackupResult};
20use risingwave_backup::meta_snapshot_v2::{MetaSnapshotV2, MetadataV2};
21use risingwave_hummock_sdk::version::{HummockVersion, HummockVersionDelta};
22use risingwave_meta_model as model;
23use risingwave_pb::hummock::PbHummockVersionDelta;
24use sea_orm::{DbErr, EntityTrait, QueryOrder, TransactionTrait};
25
26use crate::controller::SqlMetaStore;
27
28const VERSION: u32 = 2;
29
30fn map_db_err(e: DbErr) -> BackupError {
31    BackupError::MetaStorage(e.into())
32}
33
34macro_rules! define_set_metadata {
35    ($( {$name:ident, $mod_path:ident::$mod_name:ident} ),*) => {
36        async fn set_metadata(
37            metadata: &mut MetadataV2,
38            txn: &sea_orm::DatabaseTransaction,
39        ) -> BackupResult<()> {
40          $(
41              metadata.$name = $mod_path::$mod_name::Entity::find()
42                                    .all(txn)
43                                    .await
44                                    .map_err(map_db_err)?;
45          )*
46          Ok(())
47        }
48    };
49}
50
51risingwave_backup::for_all_metadata_models_v2!(define_set_metadata);
52
53pub struct MetaSnapshotV2Builder {
54    snapshot: MetaSnapshotV2,
55    meta_store: SqlMetaStore,
56}
57
58impl MetaSnapshotV2Builder {
59    pub fn new(meta_store: SqlMetaStore) -> Self {
60        Self {
61            snapshot: MetaSnapshotV2::default(),
62            meta_store,
63        }
64    }
65
66    pub async fn build(
67        &mut self,
68        id: MetaSnapshotId,
69        hummock_version_builder: impl Future<Output = HummockVersion>,
70    ) -> BackupResult<()> {
71        self.snapshot.format_version = VERSION;
72        self.snapshot.id = id;
73        // Get `hummock_version` before `meta_store_snapshot`.
74        // We have ensure the required delta logs for replay is available, see
75        // `HummockManager::delete_version_deltas`.
76        let hummock_version = hummock_version_builder.await;
77        let txn = self
78            .meta_store
79            .conn
80            .begin_with_config(
81                Some(sea_orm::IsolationLevel::Serializable),
82                Some(sea_orm::AccessMode::ReadOnly),
83            )
84            .await
85            .map_err(map_db_err)?;
86        let version_deltas = model::prelude::HummockVersionDelta::find()
87            .order_by_asc(model::hummock_version_delta::Column::Id)
88            .all(&txn)
89            .await
90            .map_err(map_db_err)?
91            .into_iter()
92            .map_into::<PbHummockVersionDelta>()
93            .map(|pb_delta| HummockVersionDelta::from_persisted_protobuf(&pb_delta));
94        let hummock_version = {
95            let mut redo_state = hummock_version;
96            let mut max_log_id = None;
97            for version_delta in version_deltas {
98                if version_delta.prev_id == redo_state.id {
99                    redo_state.apply_version_delta(&version_delta);
100                }
101                max_log_id = Some(version_delta.id);
102            }
103            if let Some(max_log_id) = max_log_id {
104                if max_log_id != redo_state.id {
105                    return Err(BackupError::Other(anyhow::anyhow!(format!(
106                        "inconsistent hummock version: expected {}, actual {}",
107                        max_log_id, redo_state.id
108                    ))));
109                }
110            }
111            redo_state
112        };
113        let mut metadata = MetadataV2 {
114            hummock_version,
115            ..Default::default()
116        };
117        set_metadata(&mut metadata, &txn).await?;
118
119        txn.commit().await.map_err(map_db_err)?;
120        self.snapshot.metadata = metadata;
121        Ok(())
122    }
123
124    pub fn finish(self) -> BackupResult<MetaSnapshotV2> {
125        // Any sanity check goes here.
126        Ok(self.snapshot)
127    }
128}
risingwave_meta/backup_restore/meta_snapshot_builder.rs

risingwave_meta/backup_restore/
meta_snapshot_builder.rs
