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