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