risingwave_meta/controller/catalog/
mod.rs

1// Copyright 2025 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15mod 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
107/// `CatalogController` is the controller for catalog related operations, including database, schema, table, view, etc.
108pub struct CatalogController {
109    pub(crate) env: MetaSrvEnv,
110    pub(crate) inner: RwLock<CatalogControllerInner>,
111}
112
113#[derive(Clone, Default, Debug)]
114pub struct DropTableConnectorContext {
115    // we only apply one drop connector action for one table each time, so no need to vector here
116    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    /// Dropped state table list, need to unregister from hummock.
126    pub(crate) removed_state_table_ids: Vec<TableId>,
127
128    /// Dropped secrets, need to remove from secret manager.
129    pub(crate) removed_secret_ids: Vec<SecretId>,
130    /// Dropped sources (when `DROP SOURCE`), need to unregister from source manager.
131    pub(crate) removed_source_ids: Vec<SourceId>,
132    /// Dropped Source fragments (when `DROP MATERIALIZED VIEW` referencing sources),
133    /// need to unregister from source manager.
134    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    /// Used in `NotificationService::subscribe`.
157    /// Need to pay attention to the order of acquiring locks to prevent deadlock problems.
158    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    /// Registered finish notifiers for creating tables.
170    ///
171    /// `DdlController` will update this map, and pass the `tx` side to `CatalogController`.
172    /// On notifying, we can remove the entry from this map.
173    #[expect(clippy::type_complexity)]
174    pub creating_table_finish_notifier:
175        HashMap<DatabaseId, HashMap<ObjectId, Vec<Sender<Result<NotificationVersion, String>>>>>,
176    /// Tables have been dropped from the meta store, but the corresponding barrier remains unfinished.
177    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        // update `created_at` as now() and `created_at_cluster_version` as current cluster version.
215        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        // mark the target subscription as `Create`.
229        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        // notify default privileges about the new subscription
271        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    // for telemetry
289    pub async fn get_connector_usage(&self) -> MetaResult<jsonbb::Value> {
290        // get connector usage by source/sink
291        // the expect format is like:
292        // {
293        //     "source": [{
294        //         "$source_id": {
295        //             "connector": "kafka",
296        //             "format": "plain",
297        //             "encode": "json"
298        //         },
299        //     }],
300        //     "sink": [{
301        //         "$sink_id": {
302        //             "connector": "pulsar",
303        //             "format": "upsert",
304        //             "encode": "avro"
305        //         },
306        //     }],
307        // }
308
309        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        // We don't need to notify the frontend, because the Init subscription is not send to frontend.
414        Ok(())
415    }
416
417    /// `clean_dirty_creating_jobs` cleans up creating jobs that are creating in Foreground mode or in Initial status.
418    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                // Notify the frontend to update the table info.
468                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        // Filter out dummy objs for replacement.
493        // todo: we'd better introduce a new dummy object type for replacement.
494        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        // Only notify delete for failed materialized views.
511        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        // The source ids for dirty tables with connector.
522        // FIXME: we should also clean dirty sources.
523        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        // Notify the frontend to update the table info.
584        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
671/// `CatalogStats` is a struct to store the statistics of all catalogs.
672pub 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    /// `list_all_tables` return all tables and internal tables.
789    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    /// `list_tables` return all `CREATED` tables, `CREATING` materialized views and internal tables that belong to them.
802    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                // Correctly set the stream job status for creating materialized views and internal tables.
843                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    /// `list_sources` return all sources and `CREATED` ones if contains any streaming jobs.
858    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        // filter out inner connector sources that are still under creating.
871        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    /// `list_sinks` return all `CREATED` sinks.
898    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    /// `list_subscriptions` return all `CREATED` subscriptions.
913    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    /// `list_indexes` return all `CREATED` indexes.
936    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    /// Since the tables have been dropped from both meta store and streaming jobs, this method removes those table copies.
1050    /// Returns the removed table copies.
1051    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}