risingwave_backup/
meta_snapshot_v2.rs

1// Copyright 2023 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::fmt::{Display, Formatter};
16
17use bytes::{Buf, BufMut};
18use risingwave_hummock_sdk::version::HummockVersion;
19use serde::{Deserialize, Serialize};
20
21use crate::meta_snapshot::{MetaSnapshot, Metadata};
22use crate::{BackupError, BackupResult};
23pub type MetaSnapshotV2 = MetaSnapshot<MetadataV2>;
24
25impl From<serde_json::Error> for BackupError {
26    fn from(value: serde_json::Error) -> Self {
27        BackupError::Other(value.into())
28    }
29}
30
31/// Add new item in the end. Do not change the order.
32#[macro_export]
33macro_rules! for_all_metadata_models_v2 {
34    ($macro:ident) => {
35        $macro! {
36            {seaql_migrations, risingwave_meta_model::serde_seaql_migration},
37            {version_stats, risingwave_meta_model::hummock_version_stats},
38            {compaction_configs, risingwave_meta_model::compaction_config},
39            {clusters, risingwave_meta_model::cluster},
40            {fragment_relation, risingwave_meta_model::fragment_relation},
41            {catalog_versions, risingwave_meta_model::catalog_version},
42            {connections, risingwave_meta_model::connection},
43            {databases, risingwave_meta_model::database},
44            {fragments, risingwave_meta_model::fragment},
45            {functions, risingwave_meta_model::function},
46            {indexes, risingwave_meta_model::index},
47            {objects, risingwave_meta_model::object},
48            {object_dependencies, risingwave_meta_model::object_dependency},
49            {schemas, risingwave_meta_model::schema},
50            {sinks, risingwave_meta_model::sink},
51            {sources, risingwave_meta_model::source},
52            {streaming_jobs, risingwave_meta_model::streaming_job},
53            {subscriptions, risingwave_meta_model::subscription},
54            {system_parameters, risingwave_meta_model::system_parameter},
55            {tables, risingwave_meta_model::table},
56            {users, risingwave_meta_model::user},
57            {user_privileges, risingwave_meta_model::user_privilege},
58            {views, risingwave_meta_model::view},
59            {workers, risingwave_meta_model::worker},
60            {worker_properties, risingwave_meta_model::worker_property},
61            {hummock_sequences, risingwave_meta_model::hummock_sequence},
62            {session_parameters, risingwave_meta_model::session_parameter},
63            {secrets, risingwave_meta_model::secret},
64            {exactly_once_iceberg_sinks, risingwave_meta_model::exactly_once_iceberg_sink},
65            {iceberg_tables, risingwave_meta_model::iceberg_tables},
66            {iceberg_namespace_properties, risingwave_meta_model::iceberg_namespace_properties},
67            {user_default_privilege, risingwave_meta_model::user_default_privilege},
68            {fragment_splits, risingwave_meta_model::fragment_splits},
69            {pending_sink_state, risingwave_meta_model::pending_sink_state},
70            {refresh_jobs, risingwave_meta_model::refresh_job},
71            {cdc_table_snapshot_splits, risingwave_meta_model::cdc_table_snapshot_split}
72        }
73    };
74}
75
76macro_rules! define_metadata_v2 {
77    ($({ $name:ident, $mod_path:ident::$mod_name:ident }),*) => {
78        #[derive(Default)]
79        pub struct MetadataV2 {
80            pub hummock_version: HummockVersion,
81            $(
82                pub $name: Vec<$mod_path::$mod_name::Model>,
83            )*
84        }
85    };
86}
87
88for_all_metadata_models_v2!(define_metadata_v2);
89
90macro_rules! define_encode_metadata {
91    ($( {$name:ident, $mod_path:ident::$mod_name:ident} ),*) => {
92        fn encode_metadata(
93          metadata: &MetadataV2,
94          buf: &mut Vec<u8>,
95        ) -> BackupResult<()> {
96            let mut _idx = 0;
97            $(
98                if _idx == 1 {
99                    put_1(buf, &metadata.hummock_version.to_protobuf())?;
100                }
101                put_n(buf, &metadata.$name)?;
102                _idx += 1;
103            )*
104            Ok(())
105        }
106    };
107}
108
109for_all_metadata_models_v2!(define_encode_metadata);
110
111macro_rules! define_decode_metadata {
112    ($( {$name:ident, $mod_path:ident::$mod_name:ident} ),*) => {
113        fn decode_metadata(
114          metadata: &mut MetadataV2,
115          mut buf: &[u8],
116        ) -> BackupResult<()> {
117            let mut _idx = 0;
118            $(
119                if _idx == 1 {
120                    metadata.hummock_version = HummockVersion::from_persisted_protobuf(&get_1(&mut buf)?);
121                }
122                metadata.$name = get_n(&mut buf)?;
123                _idx += 1;
124            )*
125            Ok(())
126        }
127    };
128}
129
130for_all_metadata_models_v2!(define_decode_metadata);
131
132impl Display for MetadataV2 {
133    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
134        writeln!(f, "clusters: {:#?}", self.clusters)?;
135        writeln!(
136            f,
137            "Hummock version: id {}, committed_epoch: {:?}",
138            self.hummock_version.id,
139            self.hummock_version.state_table_info.info(),
140        )?;
141        // optionally dump other metadata
142        Ok(())
143    }
144}
145
146impl Metadata for MetadataV2 {
147    fn encode_to(&self, buf: &mut Vec<u8>) -> BackupResult<()> {
148        encode_metadata(self, buf)
149    }
150
151    fn decode(buf: &[u8]) -> BackupResult<Self>
152    where
153        Self: Sized,
154    {
155        let mut metadata = Self::default();
156        decode_metadata(&mut metadata, buf)?;
157        Ok(metadata)
158    }
159
160    fn hummock_version_ref(&self) -> &HummockVersion {
161        &self.hummock_version
162    }
163
164    fn hummock_version(self) -> HummockVersion {
165        self.hummock_version
166    }
167}
168
169fn put_n<T: Serialize>(buf: &mut Vec<u8>, data: &[T]) -> Result<(), serde_json::Error> {
170    buf.put_u32_le(
171        data.len()
172            .try_into()
173            .unwrap_or_else(|_| panic!("cannot convert {} into u32", data.len())),
174    );
175    for d in data {
176        put_with_len_prefix(buf, d)?;
177    }
178    Ok(())
179}
180
181fn put_1<T: Serialize>(buf: &mut Vec<u8>, data: &T) -> Result<(), serde_json::Error> {
182    put_n(buf, &[data])
183}
184
185fn get_n<'a, T: Deserialize<'a>>(buf: &mut &'a [u8]) -> Result<Vec<T>, serde_json::Error> {
186    let n = buf.get_u32_le() as usize;
187    let mut elements = Vec::with_capacity(n);
188    for _ in 0..n {
189        elements.push(get_with_len_prefix(buf)?);
190    }
191    Ok(elements)
192}
193
194fn get_1<'a, T: Deserialize<'a>>(buf: &mut &'a [u8]) -> Result<T, serde_json::Error> {
195    let elements = get_n(buf)?;
196    assert_eq!(elements.len(), 1);
197    Ok(elements.into_iter().next().unwrap())
198}
199
200fn put_with_len_prefix<T: Serialize>(buf: &mut Vec<u8>, data: &T) -> Result<(), serde_json::Error> {
201    let b = serde_json::to_vec(data)?;
202    buf.put_u32_le(
203        b.len()
204            .try_into()
205            .unwrap_or_else(|_| panic!("cannot convert {} into u32", b.len())),
206    );
207    buf.put_slice(&b);
208    Ok(())
209}
210
211fn get_with_len_prefix<'a, T: Deserialize<'a>>(buf: &mut &'a [u8]) -> Result<T, serde_json::Error> {
212    let len = buf.get_u32_le() as usize;
213    let d = serde_json::from_slice(&buf[..len])?;
214    buf.advance(len);
215    Ok(d)
216}
risingwave_backup/meta_snapshot_v2.rs

risingwave_backup/
meta_snapshot_v2.rs
