risingwave_backup/
meta_snapshot_v1.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::collections::{HashMap, HashSet};
16use std::fmt::{Display, Formatter};
17
18use bytes::{Buf, BufMut, BytesMut};
19use itertools::Itertools;
20use risingwave_common::util::iter_util::ZipEqFast;
21use risingwave_hummock_sdk::HummockRawObjectId;
22use risingwave_hummock_sdk::version::HummockVersion;
23use risingwave_pb::catalog::{
24    Connection, Database, Function, Index, Schema, Secret, Sink, Source, Subscription, Table, View,
25};
26use risingwave_pb::hummock::{CompactionGroup, HummockVersionStats, PbHummockVersion};
27use risingwave_pb::meta::{SystemParams, TableFragments};
28use risingwave_pb::user::UserInfo;
29
30use crate::error::{BackupError, BackupResult};
31use crate::meta_snapshot::{
32    MetaSnapshot, Metadata, SnapshotPayloadReader, SnapshotPayloadWriter, read_u32_le,
33};
34
35// TODO: remove `ClusterMetadata` and even the trait, after applying model v2.
36pub type MetaSnapshotV1 = MetaSnapshot<ClusterMetadata>;
37
38impl Display for ClusterMetadata {
39    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
40        writeln!(f, "default_cf:")?;
41        for (k, v) in &self.default_cf {
42            let key = String::from_utf8(k.clone()).unwrap();
43            writeln!(f, "{} {:x?}", key, v)?;
44        }
45        writeln!(f, "hummock_version:")?;
46        writeln!(f, "{:#?}", self.hummock_version)?;
47        writeln!(f, "version_stats:")?;
48        writeln!(f, "{:#?}", self.version_stats)?;
49        writeln!(f, "compaction_groups:")?;
50        writeln!(f, "{:#?}", self.compaction_groups)?;
51        writeln!(f, "database:")?;
52        writeln!(f, "{:#?}", self.database)?;
53        writeln!(f, "schema:")?;
54        writeln!(f, "{:#?}", self.schema)?;
55        writeln!(f, "table:")?;
56        writeln!(f, "{:#?}", self.table)?;
57        writeln!(f, "index:")?;
58        writeln!(f, "{:#?}", self.index)?;
59        writeln!(f, "sink:")?;
60        writeln!(f, "{:#?}", self.sink)?;
61        writeln!(f, "source:")?;
62        writeln!(f, "{:#?}", self.source)?;
63        writeln!(f, "view:")?;
64        writeln!(f, "{:#?}", self.view)?;
65        writeln!(f, "connection:")?;
66        writeln!(f, "{:#?}", self.connection)?;
67        writeln!(f, "table_fragments:")?;
68        writeln!(f, "{:#?}", self.table_fragments)?;
69        writeln!(f, "user_info:")?;
70        writeln!(f, "{:#?}", self.user_info)?;
71        writeln!(f, "function:")?;
72        writeln!(f, "{:#?}", self.function)?;
73        writeln!(f, "system_param:")?;
74        writeln!(f, "{:#?}", self.system_param)?;
75        writeln!(f, "cluster_id:")?;
76        writeln!(f, "{:#?}", self.cluster_id)?;
77        writeln!(f, "subscription:")?;
78        writeln!(f, "{:#?}", self.subscription)?;
79        writeln!(f, "secret:")?;
80        writeln!(f, "{:#?}", self.secret)?;
81        Ok(())
82    }
83}
84
85impl Metadata for ClusterMetadata {
86    async fn encode_to_writer(&self, mut writer: SnapshotPayloadWriter) -> BackupResult<()> {
87        let mut buf = BytesMut::new();
88        self.encode_to(&mut buf)?;
89        writer.write_snapshot_bytes(buf.freeze()).await?;
90        writer.finish().await?;
91        Ok(())
92    }
93
94    async fn decode_from_reader(reader: SnapshotPayloadReader) -> BackupResult<Self> {
95        let data = reader.read_to_end().await?;
96        ClusterMetadata::decode(&data)
97    }
98
99    fn hummock_version_ref(&self) -> &HummockVersion {
100        &self.hummock_version
101    }
102
103    fn hummock_version(self) -> HummockVersion {
104        self.hummock_version
105    }
106
107    fn storage_url(&self) -> BackupResult<String> {
108        unreachable!("");
109    }
110
111    fn storage_directory(&self) -> BackupResult<String> {
112        unreachable!("");
113    }
114
115    fn table_change_log_object_ids(&self) -> HashSet<HummockRawObjectId> {
116        HashSet::default()
117    }
118}
119
120/// For backward compatibility, never remove fields and only append new field.
121#[derive(Debug, Default, Clone, PartialEq)]
122pub struct ClusterMetadata {
123    /// Unlike other metadata that has implemented `MetadataModel`,
124    /// `DEFAULT_COLUMN_FAMILY` stores various single row metadata, e.g. id offset and epoch offset.
125    /// So we use `default_cf` stores raw KVs for them.
126    pub default_cf: HashMap<Vec<u8>, Vec<u8>>,
127    pub hummock_version: HummockVersion,
128    pub version_stats: HummockVersionStats,
129    pub compaction_groups: Vec<CompactionGroup>,
130    pub database: Vec<Database>,
131    pub schema: Vec<Schema>,
132    pub table: Vec<Table>,
133    pub index: Vec<Index>,
134    pub sink: Vec<Sink>,
135    pub source: Vec<Source>,
136    pub view: Vec<View>,
137    pub table_fragments: Vec<TableFragments>,
138    pub user_info: Vec<UserInfo>,
139    pub function: Vec<Function>,
140    pub connection: Vec<Connection>,
141    pub system_param: SystemParams,
142    pub cluster_id: String,
143    pub subscription: Vec<Subscription>,
144    pub secret: Vec<Secret>,
145}
146
147impl ClusterMetadata {
148    pub fn encode_to(&self, buf: &mut impl BufMut) -> BackupResult<()> {
149        let default_cf_keys = self.default_cf.keys().collect_vec();
150        let default_cf_values = self.default_cf.values().collect_vec();
151        Self::encode_prost_message_list(&default_cf_keys, buf);
152        Self::encode_prost_message_list(&default_cf_values, buf);
153        Self::encode_prost_message(&PbHummockVersion::from(&self.hummock_version), buf);
154        Self::encode_prost_message(&self.version_stats, buf);
155        Self::encode_prost_message_list(&self.compaction_groups.iter().collect_vec(), buf);
156        Self::encode_prost_message_list(&self.table_fragments.iter().collect_vec(), buf);
157        Self::encode_prost_message_list(&self.user_info.iter().collect_vec(), buf);
158        Self::encode_prost_message_list(&self.database.iter().collect_vec(), buf);
159        Self::encode_prost_message_list(&self.schema.iter().collect_vec(), buf);
160        Self::encode_prost_message_list(&self.table.iter().collect_vec(), buf);
161        Self::encode_prost_message_list(&self.index.iter().collect_vec(), buf);
162        Self::encode_prost_message_list(&self.sink.iter().collect_vec(), buf);
163        Self::encode_prost_message_list(&self.source.iter().collect_vec(), buf);
164        Self::encode_prost_message_list(&self.view.iter().collect_vec(), buf);
165        Self::encode_prost_message_list(&self.function.iter().collect_vec(), buf);
166        Self::encode_prost_message_list(&self.connection.iter().collect_vec(), buf);
167        Self::encode_prost_message(&self.system_param, buf);
168        Self::encode_prost_message(&self.cluster_id, buf);
169        Self::encode_prost_message_list(&self.subscription.iter().collect_vec(), buf);
170        Self::encode_prost_message_list(&self.secret.iter().collect_vec(), buf);
171        Ok(())
172    }
173
174    pub fn decode(mut buf: &[u8]) -> BackupResult<Self> {
175        let default_cf_keys: Vec<Vec<u8>> = Self::decode_prost_message_list(&mut buf)?;
176        let default_cf_values: Vec<Vec<u8>> = Self::decode_prost_message_list(&mut buf)?;
177        let default_cf = default_cf_keys
178            .into_iter()
179            .zip_eq_fast(default_cf_values.into_iter())
180            .collect();
181        let hummock_version =
182            HummockVersion::from_persisted_protobuf_owned(Self::decode_prost_message(&mut buf)?);
183        let version_stats = Self::decode_prost_message(&mut buf)?;
184        let compaction_groups: Vec<CompactionGroup> = Self::decode_prost_message_list(&mut buf)?;
185        let table_fragments: Vec<TableFragments> = Self::decode_prost_message_list(&mut buf)?;
186        let user_info: Vec<UserInfo> = Self::decode_prost_message_list(&mut buf)?;
187        let database: Vec<Database> = Self::decode_prost_message_list(&mut buf)?;
188        let schema: Vec<Schema> = Self::decode_prost_message_list(&mut buf)?;
189        let table: Vec<Table> = Self::decode_prost_message_list(&mut buf)?;
190        let index: Vec<Index> = Self::decode_prost_message_list(&mut buf)?;
191        let sink: Vec<Sink> = Self::decode_prost_message_list(&mut buf)?;
192        let source: Vec<Source> = Self::decode_prost_message_list(&mut buf)?;
193        let view: Vec<View> = Self::decode_prost_message_list(&mut buf)?;
194        let function: Vec<Function> = Self::decode_prost_message_list(&mut buf)?;
195        let connection: Vec<Connection> = Self::decode_prost_message_list(&mut buf)?;
196        let system_param: SystemParams = Self::decode_prost_message(&mut buf)?;
197        let cluster_id: String = Self::decode_prost_message(&mut buf)?;
198        let subscription: Vec<Subscription> =
199            Self::try_decode_prost_message_list(&mut buf).unwrap_or_else(|| Ok(vec![]))?;
200        let secret: Vec<Secret> =
201            Self::try_decode_prost_message_list(&mut buf).unwrap_or_else(|| Ok(vec![]))?;
202
203        Ok(Self {
204            default_cf,
205            hummock_version,
206            version_stats,
207            compaction_groups,
208            database,
209            schema,
210            table,
211            index,
212            sink,
213            source,
214            view,
215            table_fragments,
216            user_info,
217            function,
218            connection,
219            system_param,
220            cluster_id,
221            subscription,
222            secret,
223        })
224    }
225
226    fn encode_prost_message(message: &impl prost::Message, buf: &mut impl BufMut) {
227        let encoded_message = message.encode_to_vec();
228        buf.put_u32_le(encoded_message.len() as u32);
229        buf.put_slice(&encoded_message);
230    }
231
232    fn decode_prost_message<T>(buf: &mut &[u8]) -> BackupResult<T>
233    where
234        T: prost::Message + Default,
235    {
236        let len = read_u32_le(buf)? as usize;
237        if buf.remaining() < len {
238            return Err(BackupError::Other(anyhow::anyhow!(
239                "prost payload length {} exceeds remaining buffer {}",
240                len,
241                buf.remaining()
242            )));
243        }
244        let v = buf[..len].to_vec();
245        buf.advance(len);
246        T::decode(v.as_slice()).map_err(|e| BackupError::Decoding(e.into()))
247    }
248
249    fn encode_prost_message_list(messages: &[&impl prost::Message], buf: &mut impl BufMut) {
250        buf.put_u32_le(messages.len() as u32);
251        for message in messages {
252            Self::encode_prost_message(*message, buf);
253        }
254    }
255
256    fn decode_prost_message_list<T>(buf: &mut &[u8]) -> BackupResult<Vec<T>>
257    where
258        T: prost::Message + Default,
259    {
260        let vec_len = read_u32_le(buf)? as usize;
261        let mut result = vec![];
262        for _ in 0..vec_len {
263            let v: T = Self::decode_prost_message(buf)?;
264            result.push(v);
265        }
266        Ok(result)
267    }
268
269    fn try_decode_prost_message_list<T>(buf: &mut &[u8]) -> Option<BackupResult<Vec<T>>>
270    where
271        T: prost::Message + Default,
272    {
273        if buf.is_empty() {
274            return None;
275        }
276        Some(Self::decode_prost_message_list(buf))
277    }
278}
279
280#[cfg(test)]
281mod tests {
282    use bytes::BytesMut;
283    use risingwave_hummock_sdk::HummockVersionId;
284    use risingwave_pb::hummock::{CompactionGroup, TableStats};
285
286    use crate::meta_snapshot_v1::ClusterMetadata;
287
288    #[test]
289    fn test_metadata_encoding_decoding() {
290        let mut buf = BytesMut::new();
291        let mut metadata = ClusterMetadata::default();
292        metadata.hummock_version.id = HummockVersionId::new(321);
293        metadata.default_cf.insert(vec![0, 1, 2], vec![3, 4, 5]);
294        metadata.version_stats.hummock_version_id = 10.into();
295        metadata.version_stats.table_stats.insert(
296            200.into(),
297            TableStats {
298                total_key_count: 1000,
299                ..Default::default()
300            },
301        );
302        metadata.compaction_groups.push(CompactionGroup {
303            id: 3000.into(),
304            ..Default::default()
305        });
306        metadata.encode_to(&mut buf).unwrap();
307        let decoded = ClusterMetadata::decode(&buf).unwrap();
308        assert_eq!(metadata, decoded);
309    }
310}
risingwave_backup/meta_snapshot_v1.rs

risingwave_backup/
meta_snapshot_v1.rs
