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::{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
106/// `CatalogController` is the controller for catalog related operations, including database, schema, table, view, etc.
107pub struct CatalogController {
108    pub(crate) env: MetaSrvEnv,
109    pub(crate) inner: RwLock<CatalogControllerInner>,
110}
111
112#[derive(Clone, Default, Debug)]
113pub struct DropTableConnectorContext {
114    // we only apply one drop connector action for one table each time, so no need to vector here
115    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    /// Dropped state table list, need to unregister from hummock.
125    pub(crate) removed_state_table_ids: Vec<TableId>,
126
127    /// Dropped secrets, need to remove from secret manager.
128    pub(crate) removed_secret_ids: Vec<SecretId>,
129    /// Dropped sources (when `DROP SOURCE`), need to unregister from source manager.
130    pub(crate) removed_source_ids: Vec<SourceId>,
131    /// Dropped Source fragments (when `DROP MATERIALIZED VIEW` referencing sources),
132    /// need to unregister from source manager.
133    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    /// Used in `NotificationService::subscribe`.
156    /// Need to pay attention to the order of acquiring locks to prevent deadlock problems.
157    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    /// Registered finish notifiers for creating tables.
169    ///
170    /// `DdlController` will update this map, and pass the `tx` side to `CatalogController`.
171    /// On notifying, we can remove the entry from this map.
172    #[expect(clippy::type_complexity)]
173    pub creating_table_finish_notifier:
174        HashMap<DatabaseId, HashMap<ObjectId, Vec<Sender<Result<NotificationVersion, String>>>>>,
175    /// Tables have been dropped from the meta store, but the corresponding barrier remains unfinished.
176    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        // update `created_at` as now() and `created_at_cluster_version` as current cluster version.
214        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        // mark the target subscription as `Create`.
228        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    // for telemetry
269    pub async fn get_connector_usage(&self) -> MetaResult<jsonbb::Value> {
270        // get connector usage by source/sink
271        // the expect format is like:
272        // {
273        //     "source": [{
274        //         "$source_id": {
275        //             "connector": "kafka",
276        //             "format": "plain",
277        //             "encode": "json"
278        //         },
279        //     }],
280        //     "sink": [{
281        //         "$sink_id": {
282        //             "connector": "pulsar",
283        //             "format": "upsert",
284        //             "encode": "avro"
285        //         },
286        //     }],
287        // }
288
289        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        // We don't need to notify the frontend, because the Init subscription is not send to frontend.
394        Ok(())
395    }
396
397    /// `clean_dirty_creating_jobs` cleans up creating jobs that are creating in Foreground mode or in Initial status.
398    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        // Filter out dummy objs for replacement.
447        // todo: we'd better introduce a new dummy object type for replacement.
448        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        // Only notify delete for failed materialized views.
465        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        // The source ids for dirty tables with connector.
476        // FIXME: we should also clean dirty sources.
477        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
608/// `CatalogStats` is a struct to store the statistics of all catalogs.
609pub 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    /// `list_all_tables` return all tables and internal tables.
726    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    /// `list_tables` return all `CREATED` tables, `CREATING` materialized views and internal tables that belong to them.
739    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                // Correctly set the stream job status for creating materialized views and internal tables.
780                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    /// `list_sources` return all sources and `CREATED` ones if contains any streaming jobs.
795    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        // filter out inner connector sources that are still under creating.
808        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    /// `list_sinks` return all `CREATED` sinks.
835    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    /// `list_subscriptions` return all `CREATED` subscriptions.
850    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    /// `list_indexes` return all `CREATED` indexes.
873    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    /// Since the tables have been dropped from both meta store and streaming jobs, this method removes those table copies.
987    /// Returns the removed table copies.
988    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}