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()
438 .select_only()
439 .column(streaming_job::Column::JobId)
440 .columns([
441 object::Column::Oid,
442 object::Column::ObjType,
443 object::Column::SchemaId,
444 object::Column::DatabaseId,
445 ])
446 .join(JoinType::InnerJoin, streaming_job::Relation::Object.def())
447 .filter(filter_condition)
448 .into_partial_model()
449 .all(&txn)
450 .await?;
451
452 let updated_table_ids = Self::clean_dirty_sink_downstreams(&txn).await?;
453 let updated_table_objs = if !updated_table_ids.is_empty() {
454 Table::find()
455 .find_also_related(Object)
456 .filter(table::Column::TableId.is_in(updated_table_ids))
457 .all(&txn)
458 .await?
459 } else {
460 vec![]
461 };
462
463 if dirty_job_objs.is_empty() {
464 if !updated_table_objs.is_empty() {
465 txn.commit().await?;
466
467 self.notify_frontend(
469 NotificationOperation::Update,
470 NotificationInfo::ObjectGroup(PbObjectGroup {
471 objects: updated_table_objs
472 .into_iter()
473 .map(|(t, obj)| PbObject {
474 object_info: PbObjectInfo::Table(
475 ObjectModel(t, obj.unwrap()).into(),
476 )
477 .into(),
478 })
479 .collect(),
480 }),
481 )
482 .await;
483 }
484
485 return Ok(vec![]);
486 }
487
488 self.log_cleaned_dirty_jobs(&dirty_job_objs, &txn).await?;
489
490 let dirty_job_ids = dirty_job_objs.iter().map(|obj| obj.oid).collect::<Vec<_>>();
491
492 let all_dirty_table_ids = dirty_job_objs
495 .iter()
496 .filter(|obj| obj.obj_type == ObjectType::Table)
497 .map(|obj| obj.oid)
498 .collect_vec();
499 let dirty_table_type_map: HashMap<ObjectId, TableType> = Table::find()
500 .select_only()
501 .column(table::Column::TableId)
502 .column(table::Column::TableType)
503 .filter(table::Column::TableId.is_in(all_dirty_table_ids))
504 .into_tuple::<(ObjectId, TableType)>()
505 .all(&txn)
506 .await?
507 .into_iter()
508 .collect();
509
510 let dirty_mview_objs = dirty_job_objs
512 .into_iter()
513 .filter(|obj| {
514 matches!(
515 dirty_table_type_map.get(&obj.oid),
516 Some(TableType::MaterializedView)
517 )
518 })
519 .collect_vec();
520
521 let dirty_associated_source_ids: Vec<SourceId> = Table::find()
524 .select_only()
525 .column(table::Column::OptionalAssociatedSourceId)
526 .filter(
527 table::Column::TableId
528 .is_in(dirty_job_ids.clone())
529 .and(table::Column::OptionalAssociatedSourceId.is_not_null()),
530 )
531 .into_tuple()
532 .all(&txn)
533 .await?;
534
535 let dirty_state_table_ids: Vec<TableId> = Table::find()
536 .select_only()
537 .column(table::Column::TableId)
538 .filter(table::Column::BelongsToJobId.is_in(dirty_job_ids.clone()))
539 .into_tuple()
540 .all(&txn)
541 .await?;
542
543 let dirty_mview_internal_table_objs = Object::find()
544 .select_only()
545 .columns([
546 object::Column::Oid,
547 object::Column::ObjType,
548 object::Column::SchemaId,
549 object::Column::DatabaseId,
550 ])
551 .join(JoinType::InnerJoin, object::Relation::Table.def())
552 .filter(table::Column::BelongsToJobId.is_in(dirty_mview_objs.iter().map(|obj| obj.oid)))
553 .into_partial_model()
554 .all(&txn)
555 .await?;
556
557 let to_delete_objs: HashSet<ObjectId> = dirty_job_ids
558 .clone()
559 .into_iter()
560 .chain(dirty_state_table_ids.into_iter())
561 .chain(dirty_associated_source_ids.clone().into_iter())
562 .collect();
563
564 let res = Object::delete_many()
565 .filter(object::Column::Oid.is_in(to_delete_objs))
566 .exec(&txn)
567 .await?;
568 assert!(res.rows_affected > 0);
569
570 txn.commit().await?;
571
572 let object_group = build_object_group_for_delete(
573 dirty_mview_objs
574 .into_iter()
575 .chain(dirty_mview_internal_table_objs.into_iter())
576 .collect_vec(),
577 );
578
579 let _version = self
580 .notify_frontend(NotificationOperation::Delete, object_group)
581 .await;
582
583 if !updated_table_objs.is_empty() {
585 self.notify_frontend(
586 NotificationOperation::Update,
587 NotificationInfo::ObjectGroup(PbObjectGroup {
588 objects: updated_table_objs
589 .into_iter()
590 .map(|(t, obj)| PbObject {
591 object_info: PbObjectInfo::Table(ObjectModel(t, obj.unwrap()).into())
592 .into(),
593 })
594 .collect(),
595 }),
596 )
597 .await;
598 }
599
600 Ok(dirty_associated_source_ids)
601 }
602
603 pub async fn comment_on(&self, comment: PbComment) -> MetaResult<NotificationVersion> {
604 let inner = self.inner.write().await;
605 let txn = inner.db.begin().await?;
606 ensure_object_id(ObjectType::Database, comment.database_id as _, &txn).await?;
607 ensure_object_id(ObjectType::Schema, comment.schema_id as _, &txn).await?;
608 let table_obj = Object::find_by_id(comment.table_id as ObjectId)
609 .one(&txn)
610 .await?
611 .ok_or_else(|| MetaError::catalog_id_not_found("table", comment.table_id))?;
612
613 let table = if let Some(col_idx) = comment.column_index {
614 let columns: ColumnCatalogArray = Table::find_by_id(comment.table_id as TableId)
615 .select_only()
616 .column(table::Column::Columns)
617 .into_tuple()
618 .one(&txn)
619 .await?
620 .ok_or_else(|| MetaError::catalog_id_not_found("table", comment.table_id))?;
621 let mut pb_columns = columns.to_protobuf();
622
623 let column = pb_columns
624 .get_mut(col_idx as usize)
625 .ok_or_else(|| MetaError::catalog_id_not_found("column", col_idx))?;
626 let column_desc = column.column_desc.as_mut().ok_or_else(|| {
627 anyhow!(
628 "column desc at index {} for table id {} not found",
629 col_idx,
630 comment.table_id
631 )
632 })?;
633 column_desc.description = comment.description;
634 table::ActiveModel {
635 table_id: Set(comment.table_id as _),
636 columns: Set(pb_columns.into()),
637 ..Default::default()
638 }
639 .update(&txn)
640 .await?
641 } else {
642 table::ActiveModel {
643 table_id: Set(comment.table_id as _),
644 description: Set(comment.description),
645 ..Default::default()
646 }
647 .update(&txn)
648 .await?
649 };
650 txn.commit().await?;
651
652 let version = self
653 .notify_frontend_relation_info(
654 NotificationOperation::Update,
655 PbObjectInfo::Table(ObjectModel(table, table_obj).into()),
656 )
657 .await;
658
659 Ok(version)
660 }
661
662 pub async fn complete_dropped_tables(
663 &self,
664 table_ids: impl Iterator<Item = TableId>,
665 ) -> Vec<PbTable> {
666 let mut inner = self.inner.write().await;
667 inner.complete_dropped_tables(table_ids)
668 }
669}
670
671pub struct CatalogStats {
673 pub table_num: u64,
674 pub mview_num: u64,
675 pub index_num: u64,
676 pub source_num: u64,
677 pub sink_num: u64,
678 pub function_num: u64,
679 pub streaming_job_num: u64,
680 pub actor_num: u64,
681}
682
683impl CatalogControllerInner {
684 pub async fn snapshot(&self) -> MetaResult<(Catalog, Vec<PbUserInfo>)> {
685 let databases = self.list_databases().await?;
686 let schemas = self.list_schemas().await?;
687 let tables = self.list_tables().await?;
688 let sources = self.list_sources().await?;
689 let sinks = self.list_sinks().await?;
690 let subscriptions = self.list_subscriptions().await?;
691 let indexes = self.list_indexes().await?;
692 let views = self.list_views().await?;
693 let functions = self.list_functions().await?;
694 let connections = self.list_connections().await?;
695 let secrets = self.list_secrets().await?;
696
697 let users = self.list_users().await?;
698
699 Ok((
700 (
701 databases,
702 schemas,
703 tables,
704 sources,
705 sinks,
706 subscriptions,
707 indexes,
708 views,
709 functions,
710 connections,
711 secrets,
712 ),
713 users,
714 ))
715 }
716
717 pub async fn stats(&self) -> MetaResult<CatalogStats> {
718 let mut table_num_map: HashMap<_, _> = Table::find()
719 .select_only()
720 .column(table::Column::TableType)
721 .column_as(table::Column::TableId.count(), "num")
722 .group_by(table::Column::TableType)
723 .having(table::Column::TableType.ne(TableType::Internal))
724 .into_tuple::<(TableType, i64)>()
725 .all(&self.db)
726 .await?
727 .into_iter()
728 .map(|(table_type, num)| (table_type, num as u64))
729 .collect();
730
731 let source_num = Source::find().count(&self.db).await?;
732 let sink_num = Sink::find().count(&self.db).await?;
733 let function_num = Function::find().count(&self.db).await?;
734 let streaming_job_num = StreamingJob::find().count(&self.db).await?;
735 let actor_num = Actor::find().count(&self.db).await?;
736
737 Ok(CatalogStats {
738 table_num: table_num_map.remove(&TableType::Table).unwrap_or(0),
739 mview_num: table_num_map
740 .remove(&TableType::MaterializedView)
741 .unwrap_or(0),
742 index_num: table_num_map.remove(&TableType::Index).unwrap_or(0),
743 source_num,
744 sink_num,
745 function_num,
746 streaming_job_num,
747 actor_num,
748 })
749 }
750
751 async fn list_databases(&self) -> MetaResult<Vec<PbDatabase>> {
752 let db_objs = Database::find()
753 .find_also_related(Object)
754 .all(&self.db)
755 .await?;
756 Ok(db_objs
757 .into_iter()
758 .map(|(db, obj)| ObjectModel(db, obj.unwrap()).into())
759 .collect())
760 }
761
762 async fn list_schemas(&self) -> MetaResult<Vec<PbSchema>> {
763 let schema_objs = Schema::find()
764 .find_also_related(Object)
765 .all(&self.db)
766 .await?;
767
768 Ok(schema_objs
769 .into_iter()
770 .map(|(schema, obj)| ObjectModel(schema, obj.unwrap()).into())
771 .collect())
772 }
773
774 async fn list_users(&self) -> MetaResult<Vec<PbUserInfo>> {
775 let mut user_infos: Vec<PbUserInfo> = User::find()
776 .all(&self.db)
777 .await?
778 .into_iter()
779 .map(Into::into)
780 .collect();
781
782 for user_info in &mut user_infos {
783 user_info.grant_privileges = get_user_privilege(user_info.id as _, &self.db).await?;
784 }
785 Ok(user_infos)
786 }
787
788 pub async fn list_all_state_tables(&self) -> MetaResult<Vec<PbTable>> {
790 let table_objs = Table::find()
791 .find_also_related(Object)
792 .all(&self.db)
793 .await?;
794
795 Ok(table_objs
796 .into_iter()
797 .map(|(table, obj)| ObjectModel(table, obj.unwrap()).into())
798 .collect())
799 }
800
801 async fn list_tables(&self) -> MetaResult<Vec<PbTable>> {
803 let table_objs = Table::find()
804 .find_also_related(Object)
805 .join(JoinType::LeftJoin, object::Relation::StreamingJob.def())
806 .filter(
807 streaming_job::Column::JobStatus
808 .eq(JobStatus::Created)
809 .or(table::Column::TableType.eq(TableType::MaterializedView)),
810 )
811 .all(&self.db)
812 .await?;
813
814 let created_streaming_job_ids: Vec<ObjectId> = StreamingJob::find()
815 .select_only()
816 .column(streaming_job::Column::JobId)
817 .filter(streaming_job::Column::JobStatus.eq(JobStatus::Created))
818 .into_tuple()
819 .all(&self.db)
820 .await?;
821
822 let job_ids: HashSet<ObjectId> = table_objs
823 .iter()
824 .map(|(t, _)| t.table_id)
825 .chain(created_streaming_job_ids.iter().cloned())
826 .collect();
827
828 let internal_table_objs = Table::find()
829 .find_also_related(Object)
830 .filter(
831 table::Column::TableType
832 .eq(TableType::Internal)
833 .and(table::Column::BelongsToJobId.is_in(job_ids)),
834 )
835 .all(&self.db)
836 .await?;
837
838 Ok(table_objs
839 .into_iter()
840 .chain(internal_table_objs.into_iter())
841 .map(|(table, obj)| {
842 let is_created = created_streaming_job_ids.contains(&table.table_id)
844 || (table.table_type == TableType::Internal
845 && created_streaming_job_ids.contains(&table.belongs_to_job_id.unwrap()));
846 let mut pb_table: PbTable = ObjectModel(table, obj.unwrap()).into();
847 pb_table.stream_job_status = if is_created {
848 PbStreamJobStatus::Created.into()
849 } else {
850 PbStreamJobStatus::Creating.into()
851 };
852 pb_table
853 })
854 .collect())
855 }
856
857 async fn list_sources(&self) -> MetaResult<Vec<PbSource>> {
859 let mut source_objs = Source::find()
860 .find_also_related(Object)
861 .join(JoinType::LeftJoin, object::Relation::StreamingJob.def())
862 .filter(
863 streaming_job::Column::JobStatus
864 .is_null()
865 .or(streaming_job::Column::JobStatus.eq(JobStatus::Created)),
866 )
867 .all(&self.db)
868 .await?;
869
870 let created_table_ids: HashSet<TableId> = Table::find()
872 .select_only()
873 .column(table::Column::TableId)
874 .join(JoinType::InnerJoin, table::Relation::Object1.def())
875 .join(JoinType::LeftJoin, object::Relation::StreamingJob.def())
876 .filter(
877 table::Column::OptionalAssociatedSourceId
878 .is_not_null()
879 .and(streaming_job::Column::JobStatus.eq(JobStatus::Created)),
880 )
881 .into_tuple()
882 .all(&self.db)
883 .await?
884 .into_iter()
885 .collect();
886 source_objs.retain_mut(|(source, _)| {
887 source.optional_associated_table_id.is_none()
888 || created_table_ids.contains(&source.optional_associated_table_id.unwrap())
889 });
890
891 Ok(source_objs
892 .into_iter()
893 .map(|(source, obj)| ObjectModel(source, obj.unwrap()).into())
894 .collect())
895 }
896
897 async fn list_sinks(&self) -> MetaResult<Vec<PbSink>> {
899 let sink_objs = Sink::find()
900 .find_also_related(Object)
901 .join(JoinType::LeftJoin, object::Relation::StreamingJob.def())
902 .filter(streaming_job::Column::JobStatus.eq(JobStatus::Created))
903 .all(&self.db)
904 .await?;
905
906 Ok(sink_objs
907 .into_iter()
908 .map(|(sink, obj)| ObjectModel(sink, obj.unwrap()).into())
909 .collect())
910 }
911
912 async fn list_subscriptions(&self) -> MetaResult<Vec<PbSubscription>> {
914 let subscription_objs = Subscription::find()
915 .find_also_related(Object)
916 .filter(subscription::Column::SubscriptionState.eq(SubscriptionState::Created as i32))
917 .all(&self.db)
918 .await?;
919
920 Ok(subscription_objs
921 .into_iter()
922 .map(|(subscription, obj)| ObjectModel(subscription, obj.unwrap()).into())
923 .collect())
924 }
925
926 async fn list_views(&self) -> MetaResult<Vec<PbView>> {
927 let view_objs = View::find().find_also_related(Object).all(&self.db).await?;
928
929 Ok(view_objs
930 .into_iter()
931 .map(|(view, obj)| ObjectModel(view, obj.unwrap()).into())
932 .collect())
933 }
934
935 async fn list_indexes(&self) -> MetaResult<Vec<PbIndex>> {
937 let index_objs = Index::find()
938 .find_also_related(Object)
939 .join(JoinType::LeftJoin, object::Relation::StreamingJob.def())
940 .filter(streaming_job::Column::JobStatus.eq(JobStatus::Created))
941 .all(&self.db)
942 .await?;
943
944 Ok(index_objs
945 .into_iter()
946 .map(|(index, obj)| ObjectModel(index, obj.unwrap()).into())
947 .collect())
948 }
949
950 async fn list_connections(&self) -> MetaResult<Vec<PbConnection>> {
951 let conn_objs = Connection::find()
952 .find_also_related(Object)
953 .all(&self.db)
954 .await?;
955
956 Ok(conn_objs
957 .into_iter()
958 .map(|(conn, obj)| ObjectModel(conn, obj.unwrap()).into())
959 .collect())
960 }
961
962 pub async fn list_secrets(&self) -> MetaResult<Vec<PbSecret>> {
963 let secret_objs = Secret::find()
964 .find_also_related(Object)
965 .all(&self.db)
966 .await?;
967 Ok(secret_objs
968 .into_iter()
969 .map(|(secret, obj)| ObjectModel(secret, obj.unwrap()).into())
970 .collect())
971 }
972
973 async fn list_functions(&self) -> MetaResult<Vec<PbFunction>> {
974 let func_objs = Function::find()
975 .find_also_related(Object)
976 .all(&self.db)
977 .await?;
978
979 Ok(func_objs
980 .into_iter()
981 .map(|(func, obj)| ObjectModel(func, obj.unwrap()).into())
982 .collect())
983 }
984
985 pub(crate) fn register_finish_notifier(
986 &mut self,
987 database_id: DatabaseId,
988 id: ObjectId,
989 sender: Sender<Result<NotificationVersion, String>>,
990 ) {
991 self.creating_table_finish_notifier
992 .entry(database_id)
993 .or_default()
994 .entry(id)
995 .or_default()
996 .push(sender);
997 }
998
999 pub(crate) async fn streaming_job_is_finished(&mut self, id: i32) -> MetaResult<bool> {
1000 let status = StreamingJob::find()
1001 .select_only()
1002 .column(streaming_job::Column::JobStatus)
1003 .filter(streaming_job::Column::JobId.eq(id))
1004 .into_tuple::<JobStatus>()
1005 .one(&self.db)
1006 .await?;
1007
1008 status
1009 .map(|status| status == JobStatus::Created)
1010 .ok_or_else(|| {
1011 MetaError::catalog_id_not_found("streaming job", "may have been cancelled/dropped")
1012 })
1013 }
1014
1015 pub(crate) fn notify_finish_failed(&mut self, database_id: Option<DatabaseId>, err: String) {
1016 if let Some(database_id) = database_id {
1017 if let Some(creating_tables) = self.creating_table_finish_notifier.remove(&database_id)
1018 {
1019 for tx in creating_tables.into_values().flatten() {
1020 let _ = tx.send(Err(err.clone()));
1021 }
1022 }
1023 } else {
1024 for tx in take(&mut self.creating_table_finish_notifier)
1025 .into_values()
1026 .flatten()
1027 .flat_map(|(_, txs)| txs.into_iter())
1028 {
1029 let _ = tx.send(Err(err.clone()));
1030 }
1031 }
1032 }
1033
1034 pub async fn list_time_travel_table_ids(&self) -> MetaResult<Vec<TableId>> {
1035 let table_ids: Vec<TableId> = Table::find()
1036 .select_only()
1037 .filter(table::Column::TableType.is_in(vec![
1038 TableType::Table,
1039 TableType::MaterializedView,
1040 TableType::Index,
1041 ]))
1042 .column(table::Column::TableId)
1043 .into_tuple()
1044 .all(&self.db)
1045 .await?;
1046 Ok(table_ids)
1047 }
1048
1049 pub(crate) fn complete_dropped_tables(
1052 &mut self,
1053 table_ids: impl Iterator<Item = TableId>,
1054 ) -> Vec<PbTable> {
1055 table_ids
1056 .filter_map(|table_id| {
1057 self.dropped_tables.remove(&table_id).map_or_else(
1058 || {
1059 tracing::warn!(table_id, "table not found");
1060 None
1061 },
1062 Some,
1063 )
1064 })
1065 .collect()
1066 }
1067}