1use std::collections::BTreeMap;
16use std::time::Duration;
17
18use anyhow::{Context, anyhow};
19use risingwave_common::bail;
20use risingwave_common::hash::VnodeCount;
21use risingwave_common::util::epoch::Epoch;
22use risingwave_meta_model::{
23 PrivateLinkService, connection, database, function, index, object, schema, secret, sink,
24 source, subscription, table, view,
25};
26use risingwave_meta_model_migration::{MigrationStatus, Migrator, MigratorTrait};
27use risingwave_pb::catalog::connection::PbInfo as PbConnectionInfo;
28use risingwave_pb::catalog::source::PbOptionalAssociatedTableId;
29use risingwave_pb::catalog::table::{PbEngine, PbOptionalAssociatedSourceId, PbTableType};
30use risingwave_pb::catalog::{
31 PbConnection, PbCreateType, PbDatabase, PbFunction, PbHandleConflictBehavior, PbIndex,
32 PbSchema, PbSecret, PbSink, PbSinkType, PbSource, PbStreamJobStatus, PbSubscription, PbTable,
33 PbView,
34};
35use sea_orm::{ConnectOptions, DatabaseConnection, DbBackend, ModelTrait};
36
37use crate::{MetaError, MetaResult, MetaStoreBackend};
38
39pub mod catalog;
40pub mod cluster;
41pub mod fragment;
42pub mod id;
43pub mod rename;
44pub mod scale;
45pub mod session_params;
46pub mod streaming_job;
47pub mod system_param;
48pub mod user;
49pub mod utils;
50
51impl From<sea_orm::DbErr> for MetaError {
53 fn from(err: sea_orm::DbErr) -> Self {
54 if let Some(err) = err.sql_err() {
55 return anyhow!(err).into();
56 }
57 anyhow!(err).into()
58 }
59}
60
61#[derive(Clone)]
62pub struct SqlMetaStore {
63 pub conn: DatabaseConnection,
64 pub endpoint: String,
65}
66
67impl SqlMetaStore {
68 pub async fn connect(backend: MetaStoreBackend) -> MetaResult<Self> {
70 const MAX_DURATION: Duration = Duration::new(u64::MAX / 4, 0);
71
72 #[easy_ext::ext]
73 impl ConnectOptions {
74 fn sqlite_common(&mut self) -> &mut Self {
76 self
77 .min_connections(1)
80 .max_connections(1)
81 .acquire_timeout(MAX_DURATION)
85 .connect_timeout(MAX_DURATION)
86 }
87 }
88
89 Ok(match backend {
90 MetaStoreBackend::Mem => {
91 const IN_MEMORY_STORE: &str = "sqlite::memory:";
92
93 let mut options = ConnectOptions::new(IN_MEMORY_STORE);
94
95 options
96 .sqlite_common()
97 .idle_timeout(MAX_DURATION)
102 .max_lifetime(MAX_DURATION);
103
104 let conn = sea_orm::Database::connect(options).await?;
105 Self {
106 conn,
107 endpoint: IN_MEMORY_STORE.to_owned(),
108 }
109 }
110 MetaStoreBackend::Sql { endpoint, config } => {
111 let mut options = ConnectOptions::new(endpoint.clone());
112 options
113 .max_connections(config.max_connections)
114 .min_connections(config.min_connections)
115 .connect_timeout(Duration::from_secs(config.connection_timeout_sec))
116 .idle_timeout(Duration::from_secs(config.idle_timeout_sec))
117 .acquire_timeout(Duration::from_secs(config.acquire_timeout_sec));
118
119 if DbBackend::Sqlite.is_prefix_of(&endpoint) {
120 if endpoint.contains(":memory:") || endpoint.contains("mode=memory") {
121 bail!(
122 "use the `mem` backend instead of specifying a URL of in-memory SQLite"
123 );
124 }
125 options.sqlite_common();
126 }
127
128 let conn = sea_orm::Database::connect(options).await?;
129 Self { conn, endpoint }
130 }
131 })
132 }
133
134 #[cfg(any(test, feature = "test"))]
135 pub async fn for_test() -> Self {
136 let this = Self::connect(MetaStoreBackend::Mem).await.unwrap();
137 Migrator::up(&this.conn, None).await.unwrap();
138 this
139 }
140
141 async fn is_first_launch(&self) -> MetaResult<bool> {
147 let migrations = Migrator::get_applied_migrations(&self.conn)
148 .await
149 .context("failed to get applied migrations")?;
150 for migration in migrations {
151 if migration.name() == "m20230908_072257_init"
152 && migration.status() == MigrationStatus::Applied
153 {
154 return Ok(false);
155 }
156 }
157 Ok(true)
158 }
159
160 pub async fn up(&self) -> MetaResult<bool> {
164 let cluster_first_launch = self.is_first_launch().await?;
165 Migrator::up(&self.conn, None)
167 .await
168 .context("failed to upgrade models in meta store")?;
169
170 Ok(cluster_first_launch)
171 }
172}
173
174pub struct ObjectModel<M: ModelTrait>(M, object::Model);
175
176impl From<ObjectModel<database::Model>> for PbDatabase {
177 fn from(value: ObjectModel<database::Model>) -> Self {
178 Self {
179 id: value.0.database_id as _,
180 name: value.0.name,
181 owner: value.1.owner_id as _,
182 resource_group: value.0.resource_group.clone(),
183 }
184 }
185}
186
187impl From<ObjectModel<secret::Model>> for PbSecret {
188 fn from(value: ObjectModel<secret::Model>) -> Self {
189 Self {
190 id: value.0.secret_id as _,
191 name: value.0.name,
192 database_id: value.1.database_id.unwrap() as _,
193 value: value.0.value,
194 owner: value.1.owner_id as _,
195 schema_id: value.1.schema_id.unwrap() as _,
196 }
197 }
198}
199
200impl From<ObjectModel<schema::Model>> for PbSchema {
201 fn from(value: ObjectModel<schema::Model>) -> Self {
202 Self {
203 id: value.0.schema_id as _,
204 name: value.0.name,
205 database_id: value.1.database_id.unwrap() as _,
206 owner: value.1.owner_id as _,
207 }
208 }
209}
210
211impl From<ObjectModel<table::Model>> for PbTable {
212 fn from(value: ObjectModel<table::Model>) -> Self {
213 Self {
214 id: value.0.table_id as _,
215 schema_id: value.1.schema_id.unwrap() as _,
216 database_id: value.1.database_id.unwrap() as _,
217 name: value.0.name,
218 columns: value.0.columns.to_protobuf(),
219 pk: value.0.pk.to_protobuf(),
220 dependent_relations: vec![], table_type: PbTableType::from(value.0.table_type) as _,
222 distribution_key: value.0.distribution_key.0,
223 stream_key: value.0.stream_key.0,
224 append_only: value.0.append_only,
225 owner: value.1.owner_id as _,
226 fragment_id: value.0.fragment_id.unwrap_or_default() as u32,
227 vnode_col_index: value.0.vnode_col_index.map(|index| index as _),
228 row_id_index: value.0.row_id_index.map(|index| index as _),
229 value_indices: value.0.value_indices.0,
230 definition: value.0.definition,
231 handle_pk_conflict_behavior: PbHandleConflictBehavior::from(
232 value.0.handle_pk_conflict_behavior,
233 ) as _,
234 version_column_index: value.0.version_column_index.map(|x| x as u32),
235 read_prefix_len_hint: value.0.read_prefix_len_hint as _,
236 watermark_indices: value.0.watermark_indices.0,
237 dist_key_in_pk: value.0.dist_key_in_pk.0,
238 dml_fragment_id: value.0.dml_fragment_id.map(|id| id as u32),
239 cardinality: value
240 .0
241 .cardinality
242 .map(|cardinality| cardinality.to_protobuf()),
243 initialized_at_epoch: Some(
244 Epoch::from_unix_millis(value.1.initialized_at.and_utc().timestamp_millis() as _).0,
245 ),
246 created_at_epoch: Some(
247 Epoch::from_unix_millis(value.1.created_at.and_utc().timestamp_millis() as _).0,
248 ),
249 cleaned_by_watermark: value.0.cleaned_by_watermark,
250 stream_job_status: PbStreamJobStatus::Created as _,
251 create_type: PbCreateType::Foreground as _,
252 version: value.0.version.map(|v| v.to_protobuf()),
253 optional_associated_source_id: value
254 .0
255 .optional_associated_source_id
256 .map(|id| PbOptionalAssociatedSourceId::AssociatedSourceId(id as _)),
257 description: value.0.description,
258 incoming_sinks: value.0.incoming_sinks.into_u32_array(),
259 initialized_at_cluster_version: value.1.initialized_at_cluster_version,
260 created_at_cluster_version: value.1.created_at_cluster_version,
261 retention_seconds: value.0.retention_seconds.map(|id| id as u32),
262 cdc_table_id: value.0.cdc_table_id,
263 maybe_vnode_count: VnodeCount::set(value.0.vnode_count).to_protobuf(),
264 webhook_info: value.0.webhook_info.map(|info| info.to_protobuf()),
265 job_id: value.0.belongs_to_job_id.map(|id| id as _),
266 engine: value.0.engine.map(|engine| PbEngine::from(engine) as i32),
267 clean_watermark_index_in_pk: value.0.clean_watermark_index_in_pk,
268 }
269 }
270}
271
272impl From<ObjectModel<source::Model>> for PbSource {
273 fn from(value: ObjectModel<source::Model>) -> Self {
274 let mut secret_ref_map = BTreeMap::new();
275 if let Some(secret_ref) = value.0.secret_ref {
276 secret_ref_map = secret_ref.to_protobuf();
277 }
278 Self {
279 id: value.0.source_id as _,
280 schema_id: value.1.schema_id.unwrap() as _,
281 database_id: value.1.database_id.unwrap() as _,
282 name: value.0.name,
283 row_id_index: value.0.row_id_index.map(|id| id as _),
284 columns: value.0.columns.to_protobuf(),
285 pk_column_ids: value.0.pk_column_ids.0,
286 with_properties: value.0.with_properties.0,
287 owner: value.1.owner_id as _,
288 info: value.0.source_info.map(|info| info.to_protobuf()),
289 watermark_descs: value.0.watermark_descs.to_protobuf(),
290 definition: value.0.definition,
291 connection_id: value.0.connection_id.map(|id| id as _),
292 initialized_at_epoch: Some(
294 Epoch::from_unix_millis(value.1.initialized_at.and_utc().timestamp_millis() as _).0,
295 ),
296 created_at_epoch: Some(
297 Epoch::from_unix_millis(value.1.created_at.and_utc().timestamp_millis() as _).0,
298 ),
299 version: value.0.version as _,
300 optional_associated_table_id: value
301 .0
302 .optional_associated_table_id
303 .map(|id| PbOptionalAssociatedTableId::AssociatedTableId(id as _)),
304 initialized_at_cluster_version: value.1.initialized_at_cluster_version,
305 created_at_cluster_version: value.1.created_at_cluster_version,
306 secret_refs: secret_ref_map,
307 rate_limit: value.0.rate_limit.map(|v| v as _),
308 }
309 }
310}
311
312impl From<ObjectModel<sink::Model>> for PbSink {
313 fn from(value: ObjectModel<sink::Model>) -> Self {
314 let mut secret_ref_map = BTreeMap::new();
315 if let Some(secret_ref) = value.0.secret_ref {
316 secret_ref_map = secret_ref.to_protobuf();
317 }
318 #[allow(deprecated)] Self {
320 id: value.0.sink_id as _,
321 schema_id: value.1.schema_id.unwrap() as _,
322 database_id: value.1.database_id.unwrap() as _,
323 name: value.0.name,
324 columns: value.0.columns.to_protobuf(),
325 plan_pk: value.0.plan_pk.to_protobuf(),
326 dependent_relations: vec![],
327 distribution_key: value.0.distribution_key.0,
328 downstream_pk: value.0.downstream_pk.0,
329 sink_type: PbSinkType::from(value.0.sink_type) as _,
330 owner: value.1.owner_id as _,
331 properties: value.0.properties.0,
332 definition: value.0.definition,
333 connection_id: value.0.connection_id.map(|id| id as _),
334 initialized_at_epoch: Some(
335 Epoch::from_unix_millis(value.1.initialized_at.and_utc().timestamp_millis() as _).0,
336 ),
337 created_at_epoch: Some(
338 Epoch::from_unix_millis(value.1.created_at.and_utc().timestamp_millis() as _).0,
339 ),
340 db_name: value.0.db_name,
341 sink_from_name: value.0.sink_from_name,
342 stream_job_status: PbStreamJobStatus::Created as _,
343 format_desc: value.0.sink_format_desc.map(|desc| desc.to_protobuf()),
344 target_table: value.0.target_table.map(|id| id as _),
345 initialized_at_cluster_version: value.1.initialized_at_cluster_version,
346 created_at_cluster_version: value.1.created_at_cluster_version,
347 create_type: PbCreateType::Foreground as _,
348 secret_refs: secret_ref_map,
349 original_target_columns: value
350 .0
351 .original_target_columns
352 .map(|cols| cols.to_protobuf())
353 .unwrap_or_default(),
354 }
355 }
356}
357
358impl From<ObjectModel<subscription::Model>> for PbSubscription {
359 fn from(value: ObjectModel<subscription::Model>) -> Self {
360 Self {
361 id: value.0.subscription_id as _,
362 schema_id: value.1.schema_id.unwrap() as _,
363 database_id: value.1.database_id.unwrap() as _,
364 name: value.0.name,
365 owner: value.1.owner_id as _,
366 retention_seconds: value.0.retention_seconds as _,
367 definition: value.0.definition,
368 initialized_at_epoch: Some(
369 Epoch::from_unix_millis(value.1.initialized_at.and_utc().timestamp_millis() as _).0,
370 ),
371 created_at_epoch: Some(
372 Epoch::from_unix_millis(value.1.created_at.and_utc().timestamp_millis() as _).0,
373 ),
374 initialized_at_cluster_version: value.1.initialized_at_cluster_version,
375 created_at_cluster_version: value.1.created_at_cluster_version,
376 dependent_table_id: value.0.dependent_table_id as _,
377 subscription_state: value.0.subscription_state as _,
378 }
379 }
380}
381
382impl From<ObjectModel<index::Model>> for PbIndex {
383 fn from(value: ObjectModel<index::Model>) -> Self {
384 Self {
385 id: value.0.index_id as _,
386 schema_id: value.1.schema_id.unwrap() as _,
387 database_id: value.1.database_id.unwrap() as _,
388 name: value.0.name,
389 owner: value.1.owner_id as _,
390 index_table_id: value.0.index_table_id as _,
391 primary_table_id: value.0.primary_table_id as _,
392 index_item: value.0.index_items.to_protobuf(),
393 index_column_properties: value
394 .0
395 .index_column_properties
396 .map(|p| p.to_protobuf())
397 .unwrap_or_default(),
398 index_columns_len: value.0.index_columns_len as _,
399 initialized_at_epoch: Some(
400 Epoch::from_unix_millis(value.1.initialized_at.and_utc().timestamp_millis() as _).0,
401 ),
402 created_at_epoch: Some(
403 Epoch::from_unix_millis(value.1.created_at.and_utc().timestamp_millis() as _).0,
404 ),
405 stream_job_status: PbStreamJobStatus::Created as _,
406 initialized_at_cluster_version: value.1.initialized_at_cluster_version,
407 created_at_cluster_version: value.1.created_at_cluster_version,
408 }
409 }
410}
411
412impl From<ObjectModel<view::Model>> for PbView {
413 fn from(value: ObjectModel<view::Model>) -> Self {
414 Self {
415 id: value.0.view_id as _,
416 schema_id: value.1.schema_id.unwrap() as _,
417 database_id: value.1.database_id.unwrap() as _,
418 name: value.0.name,
419 owner: value.1.owner_id as _,
420 properties: value.0.properties.0,
421 sql: value.0.definition,
422 dependent_relations: vec![], columns: value.0.columns.to_protobuf(),
424 }
425 }
426}
427
428impl From<ObjectModel<connection::Model>> for PbConnection {
429 fn from(value: ObjectModel<connection::Model>) -> Self {
430 let info: PbConnectionInfo = if value.0.info == PrivateLinkService::default() {
431 PbConnectionInfo::ConnectionParams(value.0.params.to_protobuf())
432 } else {
433 PbConnectionInfo::PrivateLinkService(value.0.info.to_protobuf())
434 };
435 Self {
436 id: value.1.oid as _,
437 schema_id: value.1.schema_id.unwrap() as _,
438 database_id: value.1.database_id.unwrap() as _,
439 name: value.0.name,
440 owner: value.1.owner_id as _,
441 info: Some(info),
442 }
443 }
444}
445
446impl From<ObjectModel<function::Model>> for PbFunction {
447 fn from(value: ObjectModel<function::Model>) -> Self {
448 Self {
449 id: value.0.function_id as _,
450 schema_id: value.1.schema_id.unwrap() as _,
451 database_id: value.1.database_id.unwrap() as _,
452 name: value.0.name,
453 owner: value.1.owner_id as _,
454 arg_names: value.0.arg_names.split(',').map(|s| s.to_owned()).collect(),
455 arg_types: value.0.arg_types.to_protobuf(),
456 return_type: Some(value.0.return_type.to_protobuf()),
457 language: value.0.language,
458 runtime: value.0.runtime,
459 link: value.0.link,
460 name_in_runtime: value.0.name_in_runtime,
461 body: value.0.body,
462 compressed_binary: value.0.compressed_binary,
463 kind: Some(value.0.kind.into()),
464 always_retry_on_network_error: value.0.always_retry_on_network_error,
465 is_async: value
466 .0
467 .options
468 .as_ref()
469 .and_then(|o| o.0.get("async").map(|v| v == "true")),
470 is_batched: value
471 .0
472 .options
473 .as_ref()
474 .and_then(|o| o.0.get("batch").map(|v| v == "true")),
475 }
476 }
477}