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