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::{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 pub(crate) removed_sink_fragment_by_targets: HashMap<FragmentId, Vec<FragmentId>>,
141}
142
143impl CatalogController {
144 pub async fn new(env: MetaSrvEnv) -> MetaResult<Self> {
145 let meta_store = env.meta_store();
146 let catalog_controller = Self {
147 env,
148 inner: RwLock::new(CatalogControllerInner {
149 db: meta_store.conn,
150 creating_table_finish_notifier: HashMap::new(),
151 dropped_tables: HashMap::new(),
152 }),
153 };
154
155 catalog_controller.init().await?;
156 Ok(catalog_controller)
157 }
158
159 pub async fn get_inner_read_guard(&self) -> RwLockReadGuard<'_, CatalogControllerInner> {
162 self.inner.read().await
163 }
164
165 pub async fn get_inner_write_guard(&self) -> RwLockWriteGuard<'_, CatalogControllerInner> {
166 self.inner.write().await
167 }
168}
169
170pub struct CatalogControllerInner {
171 pub(crate) db: DatabaseConnection,
172 #[expect(clippy::type_complexity)]
177 pub creating_table_finish_notifier:
178 HashMap<DatabaseId, HashMap<ObjectId, Vec<Sender<Result<NotificationVersion, String>>>>>,
179 pub dropped_tables: HashMap<TableId, PbTable>,
181}
182
183impl CatalogController {
184 pub(crate) async fn notify_frontend(
185 &self,
186 operation: NotificationOperation,
187 info: NotificationInfo,
188 ) -> NotificationVersion {
189 self.env
190 .notification_manager()
191 .notify_frontend(operation, info)
192 .await
193 }
194
195 pub(crate) async fn notify_frontend_relation_info(
196 &self,
197 operation: NotificationOperation,
198 relation_info: PbObjectInfo,
199 ) -> NotificationVersion {
200 self.env
201 .notification_manager()
202 .notify_frontend_object_info(operation, relation_info)
203 .await
204 }
205
206 pub(crate) async fn current_notification_version(&self) -> NotificationVersion {
207 self.env.notification_manager().current_version().await
208 }
209}
210
211impl CatalogController {
212 pub async fn finish_create_subscription_catalog(&self, subscription_id: u32) -> MetaResult<()> {
213 let inner = self.inner.write().await;
214 let txn = inner.db.begin().await?;
215 let job_id = subscription_id as i32;
216
217 let res = Object::update_many()
219 .col_expr(object::Column::CreatedAt, Expr::current_timestamp().into())
220 .col_expr(
221 object::Column::CreatedAtClusterVersion,
222 current_cluster_version().into(),
223 )
224 .filter(object::Column::Oid.eq(job_id))
225 .exec(&txn)
226 .await?;
227 if res.rows_affected == 0 {
228 return Err(MetaError::catalog_id_not_found("subscription", job_id));
229 }
230
231 let job = subscription::ActiveModel {
233 subscription_id: Set(job_id),
234 subscription_state: Set(SubscriptionState::Created.into()),
235 ..Default::default()
236 };
237 job.update(&txn).await?;
238
239 let _ = grant_default_privileges_automatically(&txn, job_id).await?;
240
241 txn.commit().await?;
242
243 Ok(())
244 }
245
246 pub async fn notify_create_subscription(
247 &self,
248 subscription_id: u32,
249 ) -> MetaResult<NotificationVersion> {
250 let inner = self.inner.read().await;
251 let job_id = subscription_id as i32;
252 let (subscription, obj) = Subscription::find_by_id(job_id)
253 .find_also_related(Object)
254 .filter(subscription::Column::SubscriptionState.eq(SubscriptionState::Created as i32))
255 .one(&inner.db)
256 .await?
257 .ok_or_else(|| MetaError::catalog_id_not_found("subscription", job_id))?;
258
259 let mut version = self
260 .notify_frontend(
261 NotificationOperation::Add,
262 NotificationInfo::ObjectGroup(PbObjectGroup {
263 objects: vec![PbObject {
264 object_info: PbObjectInfo::Subscription(
265 ObjectModel(subscription, obj.unwrap()).into(),
266 )
267 .into(),
268 }],
269 }),
270 )
271 .await;
272
273 let updated_user_ids: Vec<UserId> = UserPrivilege::find()
275 .select_only()
276 .distinct()
277 .column(user_privilege::Column::UserId)
278 .filter(user_privilege::Column::Oid.eq(subscription_id as ObjectId))
279 .into_tuple()
280 .all(&inner.db)
281 .await?;
282
283 if !updated_user_ids.is_empty() {
284 let updated_user_infos = list_user_info_by_ids(updated_user_ids, &inner.db).await?;
285 version = self.notify_users_update(updated_user_infos).await;
286 }
287
288 Ok(version)
289 }
290
291 pub async fn get_connector_usage(&self) -> MetaResult<jsonbb::Value> {
293 let inner = self.inner.read().await;
313 let source_props_and_info: Vec<(i32, Property, Option<StreamSourceInfo>)> = Source::find()
314 .select_only()
315 .column(source::Column::SourceId)
316 .column(source::Column::WithProperties)
317 .column(source::Column::SourceInfo)
318 .into_tuple()
319 .all(&inner.db)
320 .await?;
321 let sink_props_and_info: Vec<(i32, Property, Option<SinkFormatDesc>)> = Sink::find()
322 .select_only()
323 .column(sink::Column::SinkId)
324 .column(sink::Column::Properties)
325 .column(sink::Column::SinkFormatDesc)
326 .into_tuple()
327 .all(&inner.db)
328 .await?;
329 drop(inner);
330
331 let get_connector_from_property = |property: &Property| -> String {
332 property
333 .0
334 .get(UPSTREAM_SOURCE_KEY)
335 .cloned()
336 .unwrap_or_default()
337 };
338
339 let source_report: Vec<jsonbb::Value> = source_props_and_info
340 .iter()
341 .map(|(oid, property, info)| {
342 let connector_name = get_connector_from_property(property);
343 let mut format = None;
344 let mut encode = None;
345 if let Some(info) = info {
346 let pb_info = info.to_protobuf();
347 format = Some(pb_info.format().as_str_name());
348 encode = Some(pb_info.row_encode().as_str_name());
349 }
350 jsonbb::json!({
351 oid.to_string(): {
352 "connector": connector_name,
353 "format": format,
354 "encode": encode,
355 },
356 })
357 })
358 .collect_vec();
359
360 let sink_report: Vec<jsonbb::Value> = sink_props_and_info
361 .iter()
362 .map(|(oid, property, info)| {
363 let connector_name = get_connector_from_property(property);
364 let mut format = None;
365 let mut encode = None;
366 if let Some(info) = info {
367 let pb_info = info.to_protobuf();
368 format = Some(pb_info.format().as_str_name());
369 encode = Some(pb_info.encode().as_str_name());
370 }
371 jsonbb::json!({
372 oid.to_string(): {
373 "connector": connector_name,
374 "format": format,
375 "encode": encode,
376 },
377 })
378 })
379 .collect_vec();
380
381 Ok(jsonbb::json!({
382 "source": source_report,
383 "sink": sink_report,
384 }))
385 }
386
387 pub async fn clean_dirty_subscription(
388 &self,
389 database_id: Option<DatabaseId>,
390 ) -> MetaResult<()> {
391 let inner = self.inner.write().await;
392 let txn = inner.db.begin().await?;
393 let filter_condition = object::Column::ObjType.eq(ObjectType::Subscription).and(
394 object::Column::Oid.not_in_subquery(
395 Query::select()
396 .column(subscription::Column::SubscriptionId)
397 .from(Subscription)
398 .and_where(
399 subscription::Column::SubscriptionState
400 .eq(SubscriptionState::Created as i32),
401 )
402 .take(),
403 ),
404 );
405
406 let filter_condition = if let Some(database_id) = database_id {
407 filter_condition.and(object::Column::DatabaseId.eq(database_id))
408 } else {
409 filter_condition
410 };
411 Object::delete_many()
412 .filter(filter_condition)
413 .exec(&txn)
414 .await?;
415 txn.commit().await?;
416 Ok(())
418 }
419
420 pub async fn clean_dirty_creating_jobs(
422 &self,
423 database_id: Option<DatabaseId>,
424 ) -> MetaResult<Vec<SourceId>> {
425 let inner = self.inner.write().await;
426 let txn = inner.db.begin().await?;
427
428 let filter_condition = streaming_job::Column::JobStatus.eq(JobStatus::Initial).or(
429 streaming_job::Column::JobStatus
430 .eq(JobStatus::Creating)
431 .and(streaming_job::Column::CreateType.eq(CreateType::Foreground)),
432 );
433
434 let filter_condition = if let Some(database_id) = database_id {
435 filter_condition.and(object::Column::DatabaseId.eq(database_id))
436 } else {
437 filter_condition
438 };
439
440 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(
1002 streaming_job::Column::JobStatus
1003 .eq(JobStatus::Created)
1004 .or(streaming_job::Column::CreateType.eq(CreateType::Background)),
1005 )
1006 .all(&self.db)
1007 .await?;
1008
1009 let creating_indexes: HashSet<_> = StreamingJob::find()
1010 .select_only()
1011 .column(streaming_job::Column::JobId)
1012 .filter(
1013 streaming_job::Column::JobStatus
1014 .eq(JobStatus::Creating)
1015 .and(
1016 streaming_job::Column::JobId
1017 .is_in(index_objs.iter().map(|(index, _)| index.index_id)),
1018 ),
1019 )
1020 .into_tuple::<IndexId>()
1021 .all(&self.db)
1022 .await?
1023 .into_iter()
1024 .collect();
1025
1026 Ok(index_objs
1027 .into_iter()
1028 .map(|(index, obj)| {
1029 let is_creating = creating_indexes.contains(&index.index_id);
1030 let mut pb_index: PbIndex = ObjectModel(index, obj.unwrap()).into();
1031 pb_index.stream_job_status = if is_creating {
1032 PbStreamJobStatus::Creating.into()
1033 } else {
1034 PbStreamJobStatus::Created.into()
1035 };
1036 pb_index
1037 })
1038 .collect())
1039 }
1040
1041 async fn list_connections(&self) -> MetaResult<Vec<PbConnection>> {
1042 let conn_objs = Connection::find()
1043 .find_also_related(Object)
1044 .all(&self.db)
1045 .await?;
1046
1047 Ok(conn_objs
1048 .into_iter()
1049 .map(|(conn, obj)| ObjectModel(conn, obj.unwrap()).into())
1050 .collect())
1051 }
1052
1053 pub async fn list_secrets(&self) -> MetaResult<Vec<PbSecret>> {
1054 let secret_objs = Secret::find()
1055 .find_also_related(Object)
1056 .all(&self.db)
1057 .await?;
1058 Ok(secret_objs
1059 .into_iter()
1060 .map(|(secret, obj)| ObjectModel(secret, obj.unwrap()).into())
1061 .collect())
1062 }
1063
1064 async fn list_functions(&self) -> MetaResult<Vec<PbFunction>> {
1065 let func_objs = Function::find()
1066 .find_also_related(Object)
1067 .all(&self.db)
1068 .await?;
1069
1070 Ok(func_objs
1071 .into_iter()
1072 .map(|(func, obj)| ObjectModel(func, obj.unwrap()).into())
1073 .collect())
1074 }
1075
1076 pub(crate) fn register_finish_notifier(
1077 &mut self,
1078 database_id: DatabaseId,
1079 id: ObjectId,
1080 sender: Sender<Result<NotificationVersion, String>>,
1081 ) {
1082 self.creating_table_finish_notifier
1083 .entry(database_id)
1084 .or_default()
1085 .entry(id)
1086 .or_default()
1087 .push(sender);
1088 }
1089
1090 pub(crate) async fn streaming_job_is_finished(&mut self, id: i32) -> MetaResult<bool> {
1091 let status = StreamingJob::find()
1092 .select_only()
1093 .column(streaming_job::Column::JobStatus)
1094 .filter(streaming_job::Column::JobId.eq(id))
1095 .into_tuple::<JobStatus>()
1096 .one(&self.db)
1097 .await?;
1098
1099 status
1100 .map(|status| status == JobStatus::Created)
1101 .ok_or_else(|| {
1102 MetaError::catalog_id_not_found("streaming job", "may have been cancelled/dropped")
1103 })
1104 }
1105
1106 pub(crate) fn notify_finish_failed(&mut self, database_id: Option<DatabaseId>, err: String) {
1107 if let Some(database_id) = database_id {
1108 if let Some(creating_tables) = self.creating_table_finish_notifier.remove(&database_id)
1109 {
1110 for tx in creating_tables.into_values().flatten() {
1111 let _ = tx.send(Err(err.clone()));
1112 }
1113 }
1114 } else {
1115 for tx in take(&mut self.creating_table_finish_notifier)
1116 .into_values()
1117 .flatten()
1118 .flat_map(|(_, txs)| txs.into_iter())
1119 {
1120 let _ = tx.send(Err(err.clone()));
1121 }
1122 }
1123 }
1124
1125 pub(crate) fn notify_cancelled(&mut self, database_id: DatabaseId, id: ObjectId) {
1126 if let Some(creating_tables) = self.creating_table_finish_notifier.get_mut(&database_id)
1127 && let Some(tx_list) = creating_tables.remove(&id)
1128 {
1129 for tx in tx_list {
1130 let _ = tx.send(Err("Cancelled".to_owned()));
1131 }
1132 }
1133 }
1134
1135 pub async fn list_time_travel_table_ids(&self) -> MetaResult<Vec<TableId>> {
1136 let table_ids: Vec<TableId> = Table::find()
1137 .select_only()
1138 .filter(table::Column::TableType.is_in(vec![
1139 TableType::Table,
1140 TableType::MaterializedView,
1141 TableType::Index,
1142 ]))
1143 .column(table::Column::TableId)
1144 .into_tuple()
1145 .all(&self.db)
1146 .await?;
1147 Ok(table_ids)
1148 }
1149
1150 pub(crate) fn complete_dropped_tables(
1153 &mut self,
1154 table_ids: impl Iterator<Item = TableId>,
1155 ) -> Vec<PbTable> {
1156 table_ids
1157 .filter_map(|table_id| {
1158 self.dropped_tables.remove(&table_id).map_or_else(
1159 || {
1160 tracing::warn!(table_id, "table not found");
1161 None
1162 },
1163 Some,
1164 )
1165 })
1166 .collect()
1167 }
1168}