1mod alter_op;
16mod create_op;
17mod drop_op;
18mod get_op;
19mod list_op;
20mod test;
21mod util;
22
23use std::collections::{BTreeSet, HashMap, HashSet};
24use std::iter;
25use std::mem::take;
26use std::sync::Arc;
27
28use anyhow::anyhow;
29use itertools::Itertools;
30use risingwave_common::catalog::{
31 DEFAULT_SCHEMA_NAME, FragmentTypeFlag, SYSTEM_SCHEMAS, TableOption,
32};
33use risingwave_common::current_cluster_version;
34use risingwave_common::secret::LocalSecretManager;
35use risingwave_common::util::stream_graph_visitor::visit_stream_node_cont_mut;
36use risingwave_connector::source::UPSTREAM_SOURCE_KEY;
37use risingwave_connector::source::cdc::build_cdc_table_id;
38use risingwave_meta_model::object::ObjectType;
39use risingwave_meta_model::prelude::*;
40use risingwave_meta_model::table::TableType;
41use risingwave_meta_model::{
42 ActorId, ColumnCatalogArray, ConnectionId, CreateType, DatabaseId, FragmentId, I32Array,
43 IndexId, JobStatus, ObjectId, Property, SchemaId, SecretId, SinkFormatDesc, SinkId, SourceId,
44 StreamNode, StreamSourceInfo, StreamingParallelism, SubscriptionId, TableId, UserId, ViewId,
45 connection, database, fragment, function, index, object, object_dependency, schema, secret,
46 sink, source, streaming_job, subscription, table, user_privilege, view,
47};
48use risingwave_pb::catalog::connection::Info as ConnectionInfo;
49use risingwave_pb::catalog::subscription::SubscriptionState;
50use risingwave_pb::catalog::table::PbTableType;
51use risingwave_pb::catalog::{
52 PbComment, PbConnection, PbDatabase, PbFunction, PbIndex, PbSchema, PbSecret, PbSink, PbSource,
53 PbStreamJobStatus, PbSubscription, PbTable, PbView,
54};
55use risingwave_pb::meta::cancel_creating_jobs_request::PbCreatingJobInfo;
56use risingwave_pb::meta::list_object_dependencies_response::PbObjectDependencies;
57use risingwave_pb::meta::object::PbObjectInfo;
58use risingwave_pb::meta::subscribe_response::{
59 Info as NotificationInfo, Info, Operation as NotificationOperation, Operation,
60};
61use risingwave_pb::meta::{PbFragmentWorkerSlotMapping, PbObject, PbObjectGroup};
62use risingwave_pb::stream_plan::stream_node::NodeBody;
63use risingwave_pb::telemetry::PbTelemetryEventStage;
64use risingwave_pb::user::PbUserInfo;
65use sea_orm::ActiveValue::Set;
66use sea_orm::sea_query::{Expr, Query, SimpleExpr};
67use sea_orm::{
68 ActiveModelTrait, ColumnTrait, DatabaseConnection, DatabaseTransaction, EntityTrait,
69 IntoActiveModel, JoinType, PaginatorTrait, QueryFilter, QuerySelect, RelationTrait,
70 SelectColumns, TransactionTrait, Value,
71};
72use tokio::sync::oneshot::Sender;
73use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
74use tracing::info;
75
76use super::utils::{
77 check_subscription_name_duplicate, get_internal_tables_by_id, rename_relation,
78 rename_relation_refer,
79};
80use crate::controller::ObjectModel;
81use crate::controller::catalog::util::update_internal_tables;
82use crate::controller::utils::*;
83use crate::manager::{
84 IGNORED_NOTIFICATION_VERSION, MetaSrvEnv, NotificationVersion,
85 get_referred_connection_ids_from_source, get_referred_secret_ids_from_source,
86};
87use crate::rpc::ddl_controller::DropMode;
88use crate::telemetry::{MetaTelemetryJobDesc, report_event};
89use crate::{MetaError, MetaResult};
90
91pub type Catalog = (
92 Vec<PbDatabase>,
93 Vec<PbSchema>,
94 Vec<PbTable>,
95 Vec<PbSource>,
96 Vec<PbSink>,
97 Vec<PbSubscription>,
98 Vec<PbIndex>,
99 Vec<PbView>,
100 Vec<PbFunction>,
101 Vec<PbConnection>,
102 Vec<PbSecret>,
103);
104
105pub type CatalogControllerRef = Arc<CatalogController>;
106
107pub struct CatalogController {
109 pub(crate) env: MetaSrvEnv,
110 pub(crate) inner: RwLock<CatalogControllerInner>,
111}
112
113#[derive(Clone, Default, Debug)]
114pub struct DropTableConnectorContext {
115 pub(crate) to_change_streaming_job_id: ObjectId,
117 pub(crate) to_remove_state_table_id: TableId,
118 pub(crate) to_remove_source_id: SourceId,
119}
120
121#[derive(Clone, Default, Debug)]
122pub struct ReleaseContext {
123 pub(crate) database_id: DatabaseId,
124 pub(crate) removed_streaming_job_ids: Vec<ObjectId>,
125 pub(crate) removed_state_table_ids: Vec<TableId>,
127
128 pub(crate) removed_secret_ids: Vec<SecretId>,
130 pub(crate) removed_source_ids: Vec<SourceId>,
132 pub(crate) removed_source_fragments: HashMap<SourceId, BTreeSet<FragmentId>>,
135
136 pub(crate) removed_actors: HashSet<ActorId>,
137 pub(crate) removed_fragments: HashSet<FragmentId>,
138}
139
140impl CatalogController {
141 pub async fn new(env: MetaSrvEnv) -> MetaResult<Self> {
142 let meta_store = env.meta_store();
143 let catalog_controller = Self {
144 env,
145 inner: RwLock::new(CatalogControllerInner {
146 db: meta_store.conn,
147 creating_table_finish_notifier: HashMap::new(),
148 dropped_tables: HashMap::new(),
149 }),
150 };
151
152 catalog_controller.init().await?;
153 Ok(catalog_controller)
154 }
155
156 pub async fn get_inner_read_guard(&self) -> RwLockReadGuard<'_, CatalogControllerInner> {
159 self.inner.read().await
160 }
161
162 pub async fn get_inner_write_guard(&self) -> RwLockWriteGuard<'_, CatalogControllerInner> {
163 self.inner.write().await
164 }
165}
166
167pub struct CatalogControllerInner {
168 pub(crate) db: DatabaseConnection,
169 #[expect(clippy::type_complexity)]
174 pub creating_table_finish_notifier:
175 HashMap<DatabaseId, HashMap<ObjectId, Vec<Sender<Result<NotificationVersion, String>>>>>,
176 pub dropped_tables: HashMap<TableId, PbTable>,
178}
179
180impl CatalogController {
181 pub(crate) async fn notify_frontend(
182 &self,
183 operation: NotificationOperation,
184 info: NotificationInfo,
185 ) -> NotificationVersion {
186 self.env
187 .notification_manager()
188 .notify_frontend(operation, info)
189 .await
190 }
191
192 pub(crate) async fn notify_frontend_relation_info(
193 &self,
194 operation: NotificationOperation,
195 relation_info: PbObjectInfo,
196 ) -> NotificationVersion {
197 self.env
198 .notification_manager()
199 .notify_frontend_object_info(operation, relation_info)
200 .await
201 }
202
203 pub(crate) async fn current_notification_version(&self) -> NotificationVersion {
204 self.env.notification_manager().current_version().await
205 }
206}
207
208impl CatalogController {
209 pub async fn finish_create_subscription_catalog(&self, subscription_id: u32) -> MetaResult<()> {
210 let inner = self.inner.write().await;
211 let txn = inner.db.begin().await?;
212 let job_id = subscription_id as i32;
213
214 let res = Object::update_many()
216 .col_expr(object::Column::CreatedAt, Expr::current_timestamp().into())
217 .col_expr(
218 object::Column::CreatedAtClusterVersion,
219 current_cluster_version().into(),
220 )
221 .filter(object::Column::Oid.eq(job_id))
222 .exec(&txn)
223 .await?;
224 if res.rows_affected == 0 {
225 return Err(MetaError::catalog_id_not_found("subscription", job_id));
226 }
227
228 let job = subscription::ActiveModel {
230 subscription_id: Set(job_id),
231 subscription_state: Set(SubscriptionState::Created.into()),
232 ..Default::default()
233 };
234 job.update(&txn).await?;
235
236 let _ = grant_default_privileges_automatically(&txn, job_id).await?;
237
238 txn.commit().await?;
239
240 Ok(())
241 }
242
243 pub async fn notify_create_subscription(
244 &self,
245 subscription_id: u32,
246 ) -> MetaResult<NotificationVersion> {
247 let inner = self.inner.read().await;
248 let job_id = subscription_id as i32;
249 let (subscription, obj) = Subscription::find_by_id(job_id)
250 .find_also_related(Object)
251 .filter(subscription::Column::SubscriptionState.eq(SubscriptionState::Created as i32))
252 .one(&inner.db)
253 .await?
254 .ok_or_else(|| MetaError::catalog_id_not_found("subscription", job_id))?;
255
256 let mut version = self
257 .notify_frontend(
258 NotificationOperation::Add,
259 NotificationInfo::ObjectGroup(PbObjectGroup {
260 objects: vec![PbObject {
261 object_info: PbObjectInfo::Subscription(
262 ObjectModel(subscription, obj.unwrap()).into(),
263 )
264 .into(),
265 }],
266 }),
267 )
268 .await;
269
270 let updated_user_ids: Vec<UserId> = UserPrivilege::find()
272 .select_only()
273 .distinct()
274 .column(user_privilege::Column::UserId)
275 .filter(user_privilege::Column::Oid.eq(subscription_id as ObjectId))
276 .into_tuple()
277 .all(&inner.db)
278 .await?;
279
280 if !updated_user_ids.is_empty() {
281 let updated_user_infos = list_user_info_by_ids(updated_user_ids, &inner.db).await?;
282 version = self.notify_users_update(updated_user_infos).await;
283 }
284
285 Ok(version)
286 }
287
288 pub async fn get_connector_usage(&self) -> MetaResult<jsonbb::Value> {
290 let inner = self.inner.read().await;
310 let source_props_and_info: Vec<(i32, Property, Option<StreamSourceInfo>)> = Source::find()
311 .select_only()
312 .column(source::Column::SourceId)
313 .column(source::Column::WithProperties)
314 .column(source::Column::SourceInfo)
315 .into_tuple()
316 .all(&inner.db)
317 .await?;
318 let sink_props_and_info: Vec<(i32, Property, Option<SinkFormatDesc>)> = Sink::find()
319 .select_only()
320 .column(sink::Column::SinkId)
321 .column(sink::Column::Properties)
322 .column(sink::Column::SinkFormatDesc)
323 .into_tuple()
324 .all(&inner.db)
325 .await?;
326 drop(inner);
327
328 let get_connector_from_property = |property: &Property| -> String {
329 property
330 .0
331 .get(UPSTREAM_SOURCE_KEY)
332 .cloned()
333 .unwrap_or_default()
334 };
335
336 let source_report: Vec<jsonbb::Value> = source_props_and_info
337 .iter()
338 .map(|(oid, property, info)| {
339 let connector_name = get_connector_from_property(property);
340 let mut format = None;
341 let mut encode = None;
342 if let Some(info) = info {
343 let pb_info = info.to_protobuf();
344 format = Some(pb_info.format().as_str_name());
345 encode = Some(pb_info.row_encode().as_str_name());
346 }
347 jsonbb::json!({
348 oid.to_string(): {
349 "connector": connector_name,
350 "format": format,
351 "encode": encode,
352 },
353 })
354 })
355 .collect_vec();
356
357 let sink_report: Vec<jsonbb::Value> = sink_props_and_info
358 .iter()
359 .map(|(oid, property, info)| {
360 let connector_name = get_connector_from_property(property);
361 let mut format = None;
362 let mut encode = None;
363 if let Some(info) = info {
364 let pb_info = info.to_protobuf();
365 format = Some(pb_info.format().as_str_name());
366 encode = Some(pb_info.encode().as_str_name());
367 }
368 jsonbb::json!({
369 oid.to_string(): {
370 "connector": connector_name,
371 "format": format,
372 "encode": encode,
373 },
374 })
375 })
376 .collect_vec();
377
378 Ok(jsonbb::json!({
379 "source": source_report,
380 "sink": sink_report,
381 }))
382 }
383
384 pub async fn clean_dirty_subscription(
385 &self,
386 database_id: Option<DatabaseId>,
387 ) -> MetaResult<()> {
388 let inner = self.inner.write().await;
389 let txn = inner.db.begin().await?;
390 let filter_condition = object::Column::ObjType.eq(ObjectType::Subscription).and(
391 object::Column::Oid.not_in_subquery(
392 Query::select()
393 .column(subscription::Column::SubscriptionId)
394 .from(Subscription)
395 .and_where(
396 subscription::Column::SubscriptionState
397 .eq(SubscriptionState::Created as i32),
398 )
399 .take(),
400 ),
401 );
402
403 let filter_condition = if let Some(database_id) = database_id {
404 filter_condition.and(object::Column::DatabaseId.eq(database_id))
405 } else {
406 filter_condition
407 };
408 Object::delete_many()
409 .filter(filter_condition)
410 .exec(&txn)
411 .await?;
412 txn.commit().await?;
413 Ok(())
415 }
416
417 pub async fn clean_dirty_creating_jobs(
419 &self,
420 database_id: Option<DatabaseId>,
421 ) -> MetaResult<Vec<SourceId>> {
422 let inner = self.inner.write().await;
423 let txn = inner.db.begin().await?;
424
425 let filter_condition = streaming_job::Column::JobStatus.eq(JobStatus::Initial).or(
426 streaming_job::Column::JobStatus
427 .eq(JobStatus::Creating)
428 .and(streaming_job::Column::CreateType.eq(CreateType::Foreground)),
429 );
430
431 let filter_condition = if let Some(database_id) = database_id {
432 filter_condition.and(object::Column::DatabaseId.eq(database_id))
433 } else {
434 filter_condition
435 };
436
437 let dirty_job_objs: Vec<PartialObject> = streaming_job::Entity::find()
441 .select_only()
442 .column(streaming_job::Column::JobId)
443 .columns([
444 object::Column::Oid,
445 object::Column::ObjType,
446 object::Column::SchemaId,
447 object::Column::DatabaseId,
448 ])
449 .join(JoinType::InnerJoin, streaming_job::Relation::Object.def())
450 .filter(filter_condition)
451 .into_partial_model()
452 .all(&txn)
453 .await?;
454
455 let updated_table_ids = Self::clean_dirty_sink_downstreams(&txn).await?;
456 let updated_table_objs = if !updated_table_ids.is_empty() {
457 Table::find()
458 .find_also_related(Object)
459 .filter(table::Column::TableId.is_in(updated_table_ids))
460 .all(&txn)
461 .await?
462 } else {
463 vec![]
464 };
465
466 if dirty_job_objs.is_empty() {
467 if !updated_table_objs.is_empty() {
468 txn.commit().await?;
469
470 self.notify_frontend(
472 NotificationOperation::Update,
473 NotificationInfo::ObjectGroup(PbObjectGroup {
474 objects: updated_table_objs
475 .into_iter()
476 .map(|(t, obj)| PbObject {
477 object_info: PbObjectInfo::Table(
478 ObjectModel(t, obj.unwrap()).into(),
479 )
480 .into(),
481 })
482 .collect(),
483 }),
484 )
485 .await;
486 }
487
488 return Ok(vec![]);
489 }
490
491 self.log_cleaned_dirty_jobs(&dirty_job_objs, &txn).await?;
492
493 let dirty_job_ids = dirty_job_objs.iter().map(|obj| obj.oid).collect::<Vec<_>>();
494
495 let all_dirty_table_ids = dirty_job_objs
498 .iter()
499 .filter(|obj| obj.obj_type == ObjectType::Table)
500 .map(|obj| obj.oid)
501 .collect_vec();
502 let dirty_table_type_map: HashMap<ObjectId, TableType> = Table::find()
503 .select_only()
504 .column(table::Column::TableId)
505 .column(table::Column::TableType)
506 .filter(table::Column::TableId.is_in(all_dirty_table_ids))
507 .into_tuple::<(ObjectId, TableType)>()
508 .all(&txn)
509 .await?
510 .into_iter()
511 .collect();
512
513 let dirty_background_jobs: HashSet<ObjectId> = streaming_job::Entity::find()
514 .select_only()
515 .column(streaming_job::Column::JobId)
516 .filter(
517 streaming_job::Column::JobId
518 .is_in(dirty_job_ids.clone())
519 .and(streaming_job::Column::CreateType.eq(CreateType::Background)),
520 )
521 .into_tuple()
522 .all(&txn)
523 .await?
524 .into_iter()
525 .collect();
526
527 let to_notify_objs = dirty_job_objs
529 .into_iter()
530 .filter(|obj| {
531 matches!(
532 dirty_table_type_map.get(&obj.oid),
533 Some(TableType::MaterializedView)
534 ) || dirty_background_jobs.contains(&obj.oid)
535 })
536 .collect_vec();
537
538 let dirty_associated_source_ids: Vec<SourceId> = Table::find()
541 .select_only()
542 .column(table::Column::OptionalAssociatedSourceId)
543 .filter(
544 table::Column::TableId
545 .is_in(dirty_job_ids.clone())
546 .and(table::Column::OptionalAssociatedSourceId.is_not_null()),
547 )
548 .into_tuple()
549 .all(&txn)
550 .await?;
551
552 let dirty_state_table_ids: Vec<TableId> = Table::find()
553 .select_only()
554 .column(table::Column::TableId)
555 .filter(table::Column::BelongsToJobId.is_in(dirty_job_ids.clone()))
556 .into_tuple()
557 .all(&txn)
558 .await?;
559
560 let dirty_internal_table_objs = Object::find()
561 .select_only()
562 .columns([
563 object::Column::Oid,
564 object::Column::ObjType,
565 object::Column::SchemaId,
566 object::Column::DatabaseId,
567 ])
568 .join(JoinType::InnerJoin, object::Relation::Table.def())
569 .filter(table::Column::BelongsToJobId.is_in(to_notify_objs.iter().map(|obj| obj.oid)))
570 .into_partial_model()
571 .all(&txn)
572 .await?;
573
574 let to_delete_objs: HashSet<ObjectId> = dirty_job_ids
575 .clone()
576 .into_iter()
577 .chain(dirty_state_table_ids.into_iter())
578 .chain(dirty_associated_source_ids.clone().into_iter())
579 .collect();
580
581 let res = Object::delete_many()
582 .filter(object::Column::Oid.is_in(to_delete_objs))
583 .exec(&txn)
584 .await?;
585 assert!(res.rows_affected > 0);
586
587 txn.commit().await?;
588
589 let object_group = build_object_group_for_delete(
590 to_notify_objs
591 .into_iter()
592 .chain(dirty_internal_table_objs.into_iter())
593 .collect_vec(),
594 );
595
596 let _version = self
597 .notify_frontend(NotificationOperation::Delete, object_group)
598 .await;
599
600 if !updated_table_objs.is_empty() {
602 self.notify_frontend(
603 NotificationOperation::Update,
604 NotificationInfo::ObjectGroup(PbObjectGroup {
605 objects: updated_table_objs
606 .into_iter()
607 .map(|(t, obj)| PbObject {
608 object_info: PbObjectInfo::Table(ObjectModel(t, obj.unwrap()).into())
609 .into(),
610 })
611 .collect(),
612 }),
613 )
614 .await;
615 }
616
617 Ok(dirty_associated_source_ids)
618 }
619
620 pub async fn comment_on(&self, comment: PbComment) -> MetaResult<NotificationVersion> {
621 let inner = self.inner.write().await;
622 let txn = inner.db.begin().await?;
623 ensure_object_id(ObjectType::Database, comment.database_id as _, &txn).await?;
624 ensure_object_id(ObjectType::Schema, comment.schema_id as _, &txn).await?;
625 let table_obj = Object::find_by_id(comment.table_id as ObjectId)
626 .one(&txn)
627 .await?
628 .ok_or_else(|| MetaError::catalog_id_not_found("table", comment.table_id))?;
629
630 let table = if let Some(col_idx) = comment.column_index {
631 let columns: ColumnCatalogArray = Table::find_by_id(comment.table_id as TableId)
632 .select_only()
633 .column(table::Column::Columns)
634 .into_tuple()
635 .one(&txn)
636 .await?
637 .ok_or_else(|| MetaError::catalog_id_not_found("table", comment.table_id))?;
638 let mut pb_columns = columns.to_protobuf();
639
640 let column = pb_columns
641 .get_mut(col_idx as usize)
642 .ok_or_else(|| MetaError::catalog_id_not_found("column", col_idx))?;
643 let column_desc = column.column_desc.as_mut().ok_or_else(|| {
644 anyhow!(
645 "column desc at index {} for table id {} not found",
646 col_idx,
647 comment.table_id
648 )
649 })?;
650 column_desc.description = comment.description;
651 table::ActiveModel {
652 table_id: Set(comment.table_id as _),
653 columns: Set(pb_columns.into()),
654 ..Default::default()
655 }
656 .update(&txn)
657 .await?
658 } else {
659 table::ActiveModel {
660 table_id: Set(comment.table_id as _),
661 description: Set(comment.description),
662 ..Default::default()
663 }
664 .update(&txn)
665 .await?
666 };
667 txn.commit().await?;
668
669 let version = self
670 .notify_frontend_relation_info(
671 NotificationOperation::Update,
672 PbObjectInfo::Table(ObjectModel(table, table_obj).into()),
673 )
674 .await;
675
676 Ok(version)
677 }
678
679 pub async fn complete_dropped_tables(
680 &self,
681 table_ids: impl Iterator<Item = TableId>,
682 ) -> Vec<PbTable> {
683 let mut inner = self.inner.write().await;
684 inner.complete_dropped_tables(table_ids)
685 }
686}
687
688pub struct CatalogStats {
690 pub table_num: u64,
691 pub mview_num: u64,
692 pub index_num: u64,
693 pub source_num: u64,
694 pub sink_num: u64,
695 pub function_num: u64,
696 pub streaming_job_num: u64,
697 pub actor_num: u64,
698}
699
700impl CatalogControllerInner {
701 pub async fn snapshot(&self) -> MetaResult<(Catalog, Vec<PbUserInfo>)> {
702 let databases = self.list_databases().await?;
703 let schemas = self.list_schemas().await?;
704 let tables = self.list_tables().await?;
705 let sources = self.list_sources().await?;
706 let sinks = self.list_sinks().await?;
707 let subscriptions = self.list_subscriptions().await?;
708 let indexes = self.list_indexes().await?;
709 let views = self.list_views().await?;
710 let functions = self.list_functions().await?;
711 let connections = self.list_connections().await?;
712 let secrets = self.list_secrets().await?;
713
714 let users = self.list_users().await?;
715
716 Ok((
717 (
718 databases,
719 schemas,
720 tables,
721 sources,
722 sinks,
723 subscriptions,
724 indexes,
725 views,
726 functions,
727 connections,
728 secrets,
729 ),
730 users,
731 ))
732 }
733
734 pub async fn stats(&self) -> MetaResult<CatalogStats> {
735 let mut table_num_map: HashMap<_, _> = Table::find()
736 .select_only()
737 .column(table::Column::TableType)
738 .column_as(table::Column::TableId.count(), "num")
739 .group_by(table::Column::TableType)
740 .having(table::Column::TableType.ne(TableType::Internal))
741 .into_tuple::<(TableType, i64)>()
742 .all(&self.db)
743 .await?
744 .into_iter()
745 .map(|(table_type, num)| (table_type, num as u64))
746 .collect();
747
748 let source_num = Source::find().count(&self.db).await?;
749 let sink_num = Sink::find().count(&self.db).await?;
750 let function_num = Function::find().count(&self.db).await?;
751 let streaming_job_num = StreamingJob::find().count(&self.db).await?;
752 let actor_num = Actor::find().count(&self.db).await?;
753
754 Ok(CatalogStats {
755 table_num: table_num_map.remove(&TableType::Table).unwrap_or(0),
756 mview_num: table_num_map
757 .remove(&TableType::MaterializedView)
758 .unwrap_or(0),
759 index_num: table_num_map.remove(&TableType::Index).unwrap_or(0),
760 source_num,
761 sink_num,
762 function_num,
763 streaming_job_num,
764 actor_num,
765 })
766 }
767
768 async fn list_databases(&self) -> MetaResult<Vec<PbDatabase>> {
769 let db_objs = Database::find()
770 .find_also_related(Object)
771 .all(&self.db)
772 .await?;
773 Ok(db_objs
774 .into_iter()
775 .map(|(db, obj)| ObjectModel(db, obj.unwrap()).into())
776 .collect())
777 }
778
779 async fn list_schemas(&self) -> MetaResult<Vec<PbSchema>> {
780 let schema_objs = Schema::find()
781 .find_also_related(Object)
782 .all(&self.db)
783 .await?;
784
785 Ok(schema_objs
786 .into_iter()
787 .map(|(schema, obj)| ObjectModel(schema, obj.unwrap()).into())
788 .collect())
789 }
790
791 async fn list_users(&self) -> MetaResult<Vec<PbUserInfo>> {
792 let mut user_infos: Vec<PbUserInfo> = User::find()
793 .all(&self.db)
794 .await?
795 .into_iter()
796 .map(Into::into)
797 .collect();
798
799 for user_info in &mut user_infos {
800 user_info.grant_privileges = get_user_privilege(user_info.id as _, &self.db).await?;
801 }
802 Ok(user_infos)
803 }
804
805 pub async fn list_all_state_tables(&self) -> MetaResult<Vec<PbTable>> {
807 let table_objs = Table::find()
808 .find_also_related(Object)
809 .all(&self.db)
810 .await?;
811
812 Ok(table_objs
813 .into_iter()
814 .map(|(table, obj)| ObjectModel(table, obj.unwrap()).into())
815 .collect())
816 }
817
818 async fn list_tables(&self) -> MetaResult<Vec<PbTable>> {
820 let table_objs = Table::find()
821 .find_also_related(Object)
822 .join(JoinType::LeftJoin, object::Relation::StreamingJob.def())
823 .filter(
824 streaming_job::Column::JobStatus.eq(JobStatus::Created).or(
825 table::Column::TableType
826 .eq(TableType::MaterializedView)
827 .or(streaming_job::Column::CreateType.eq(CreateType::Background)),
828 ),
829 )
830 .all(&self.db)
831 .await?;
832
833 let job_statuses: HashMap<ObjectId, JobStatus> = StreamingJob::find()
834 .select_only()
835 .column(streaming_job::Column::JobId)
836 .column(streaming_job::Column::JobStatus)
837 .filter(
838 streaming_job::Column::JobStatus
839 .eq(JobStatus::Created)
840 .or(streaming_job::Column::CreateType.eq(CreateType::Background)),
841 )
842 .into_tuple::<(ObjectId, JobStatus)>()
843 .all(&self.db)
844 .await?
845 .into_iter()
846 .collect();
847
848 let job_ids: HashSet<ObjectId> = table_objs
849 .iter()
850 .map(|(t, _)| t.table_id)
851 .chain(job_statuses.keys().cloned())
852 .collect();
853
854 let internal_table_objs = Table::find()
855 .find_also_related(Object)
856 .filter(
857 table::Column::TableType
858 .eq(TableType::Internal)
859 .and(table::Column::BelongsToJobId.is_in(job_ids)),
860 )
861 .all(&self.db)
862 .await?;
863
864 Ok(table_objs
865 .into_iter()
866 .chain(internal_table_objs.into_iter())
867 .map(|(table, obj)| {
868 let status: PbStreamJobStatus = if table.table_type == TableType::Internal {
871 (*job_statuses
872 .get(&table.belongs_to_job_id.unwrap())
873 .unwrap_or(&JobStatus::Creating))
874 .into()
875 } else {
876 (*job_statuses
877 .get(&table.table_id)
878 .unwrap_or(&JobStatus::Creating))
879 .into()
880 };
881 let mut pb_table: PbTable = ObjectModel(table, obj.unwrap()).into();
882 pb_table.stream_job_status = status.into();
883 pb_table
884 })
885 .collect())
886 }
887
888 async fn list_sources(&self) -> MetaResult<Vec<PbSource>> {
890 let mut source_objs = Source::find()
891 .find_also_related(Object)
892 .join(JoinType::LeftJoin, object::Relation::StreamingJob.def())
893 .filter(
894 streaming_job::Column::JobStatus
895 .is_null()
896 .or(streaming_job::Column::JobStatus.eq(JobStatus::Created)),
897 )
898 .all(&self.db)
899 .await?;
900
901 let created_table_ids: HashSet<TableId> = Table::find()
903 .select_only()
904 .column(table::Column::TableId)
905 .join(JoinType::InnerJoin, table::Relation::Object1.def())
906 .join(JoinType::LeftJoin, object::Relation::StreamingJob.def())
907 .filter(
908 table::Column::OptionalAssociatedSourceId
909 .is_not_null()
910 .and(streaming_job::Column::JobStatus.eq(JobStatus::Created)),
911 )
912 .into_tuple()
913 .all(&self.db)
914 .await?
915 .into_iter()
916 .collect();
917 source_objs.retain_mut(|(source, _)| {
918 source.optional_associated_table_id.is_none()
919 || created_table_ids.contains(&source.optional_associated_table_id.unwrap())
920 });
921
922 Ok(source_objs
923 .into_iter()
924 .map(|(source, obj)| ObjectModel(source, obj.unwrap()).into())
925 .collect())
926 }
927
928 async fn list_sinks(&self) -> MetaResult<Vec<PbSink>> {
930 let sink_objs = Sink::find()
931 .find_also_related(Object)
932 .join(JoinType::LeftJoin, object::Relation::StreamingJob.def())
933 .filter(
934 streaming_job::Column::JobStatus
935 .eq(JobStatus::Created)
936 .or(streaming_job::Column::CreateType.eq(CreateType::Background)),
937 )
938 .all(&self.db)
939 .await?;
940
941 let creating_sinks: HashSet<_> = StreamingJob::find()
942 .select_only()
943 .column(streaming_job::Column::JobId)
944 .filter(
945 streaming_job::Column::JobStatus
946 .eq(JobStatus::Creating)
947 .and(
948 streaming_job::Column::JobId
949 .is_in(sink_objs.iter().map(|(sink, _)| sink.sink_id)),
950 ),
951 )
952 .into_tuple::<SinkId>()
953 .all(&self.db)
954 .await?
955 .into_iter()
956 .collect();
957
958 Ok(sink_objs
959 .into_iter()
960 .map(|(sink, obj)| {
961 let is_creating = creating_sinks.contains(&sink.sink_id);
962 let mut pb_sink: PbSink = ObjectModel(sink, obj.unwrap()).into();
963 pb_sink.stream_job_status = if is_creating {
964 PbStreamJobStatus::Creating.into()
965 } else {
966 PbStreamJobStatus::Created.into()
967 };
968 pb_sink
969 })
970 .collect())
971 }
972
973 async fn list_subscriptions(&self) -> MetaResult<Vec<PbSubscription>> {
975 let subscription_objs = Subscription::find()
976 .find_also_related(Object)
977 .filter(subscription::Column::SubscriptionState.eq(SubscriptionState::Created as i32))
978 .all(&self.db)
979 .await?;
980
981 Ok(subscription_objs
982 .into_iter()
983 .map(|(subscription, obj)| ObjectModel(subscription, obj.unwrap()).into())
984 .collect())
985 }
986
987 async fn list_views(&self) -> MetaResult<Vec<PbView>> {
988 let view_objs = View::find().find_also_related(Object).all(&self.db).await?;
989
990 Ok(view_objs
991 .into_iter()
992 .map(|(view, obj)| ObjectModel(view, obj.unwrap()).into())
993 .collect())
994 }
995
996 async fn list_indexes(&self) -> MetaResult<Vec<PbIndex>> {
998 let index_objs = Index::find()
999 .find_also_related(Object)
1000 .join(JoinType::LeftJoin, object::Relation::StreamingJob.def())
1001 .filter(streaming_job::Column::JobStatus.eq(JobStatus::Created))
1002 .all(&self.db)
1003 .await?;
1004
1005 Ok(index_objs
1006 .into_iter()
1007 .map(|(index, obj)| ObjectModel(index, obj.unwrap()).into())
1008 .collect())
1009 }
1010
1011 async fn list_connections(&self) -> MetaResult<Vec<PbConnection>> {
1012 let conn_objs = Connection::find()
1013 .find_also_related(Object)
1014 .all(&self.db)
1015 .await?;
1016
1017 Ok(conn_objs
1018 .into_iter()
1019 .map(|(conn, obj)| ObjectModel(conn, obj.unwrap()).into())
1020 .collect())
1021 }
1022
1023 pub async fn list_secrets(&self) -> MetaResult<Vec<PbSecret>> {
1024 let secret_objs = Secret::find()
1025 .find_also_related(Object)
1026 .all(&self.db)
1027 .await?;
1028 Ok(secret_objs
1029 .into_iter()
1030 .map(|(secret, obj)| ObjectModel(secret, obj.unwrap()).into())
1031 .collect())
1032 }
1033
1034 async fn list_functions(&self) -> MetaResult<Vec<PbFunction>> {
1035 let func_objs = Function::find()
1036 .find_also_related(Object)
1037 .all(&self.db)
1038 .await?;
1039
1040 Ok(func_objs
1041 .into_iter()
1042 .map(|(func, obj)| ObjectModel(func, obj.unwrap()).into())
1043 .collect())
1044 }
1045
1046 pub(crate) fn register_finish_notifier(
1047 &mut self,
1048 database_id: DatabaseId,
1049 id: ObjectId,
1050 sender: Sender<Result<NotificationVersion, String>>,
1051 ) {
1052 self.creating_table_finish_notifier
1053 .entry(database_id)
1054 .or_default()
1055 .entry(id)
1056 .or_default()
1057 .push(sender);
1058 }
1059
1060 pub(crate) async fn streaming_job_is_finished(&mut self, id: i32) -> MetaResult<bool> {
1061 let status = StreamingJob::find()
1062 .select_only()
1063 .column(streaming_job::Column::JobStatus)
1064 .filter(streaming_job::Column::JobId.eq(id))
1065 .into_tuple::<JobStatus>()
1066 .one(&self.db)
1067 .await?;
1068
1069 status
1070 .map(|status| status == JobStatus::Created)
1071 .ok_or_else(|| {
1072 MetaError::catalog_id_not_found("streaming job", "may have been cancelled/dropped")
1073 })
1074 }
1075
1076 pub(crate) fn notify_finish_failed(&mut self, database_id: Option<DatabaseId>, err: String) {
1077 if let Some(database_id) = database_id {
1078 if let Some(creating_tables) = self.creating_table_finish_notifier.remove(&database_id)
1079 {
1080 for tx in creating_tables.into_values().flatten() {
1081 let _ = tx.send(Err(err.clone()));
1082 }
1083 }
1084 } else {
1085 for tx in take(&mut self.creating_table_finish_notifier)
1086 .into_values()
1087 .flatten()
1088 .flat_map(|(_, txs)| txs.into_iter())
1089 {
1090 let _ = tx.send(Err(err.clone()));
1091 }
1092 }
1093 }
1094
1095 pub async fn list_time_travel_table_ids(&self) -> MetaResult<Vec<TableId>> {
1096 let table_ids: Vec<TableId> = Table::find()
1097 .select_only()
1098 .filter(table::Column::TableType.is_in(vec![
1099 TableType::Table,
1100 TableType::MaterializedView,
1101 TableType::Index,
1102 ]))
1103 .column(table::Column::TableId)
1104 .into_tuple()
1105 .all(&self.db)
1106 .await?;
1107 Ok(table_ids)
1108 }
1109
1110 pub(crate) fn complete_dropped_tables(
1113 &mut self,
1114 table_ids: impl Iterator<Item = TableId>,
1115 ) -> Vec<PbTable> {
1116 table_ids
1117 .filter_map(|table_id| {
1118 self.dropped_tables.remove(&table_id).map_or_else(
1119 || {
1120 tracing::warn!(table_id, "table not found");
1121 None
1122 },
1123 Some,
1124 )
1125 })
1126 .collect()
1127 }
1128}