risingwave_meta/controller/catalog/
get_op.rs

1// Copyright 2024 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
15use std::collections::{HashMap, HashSet};
16
17use risingwave_common::catalog::ColumnCatalog;
18use risingwave_common::id::JobId;
19use sea_orm::ConnectionTrait;
20
21use super::*;
22use crate::controller::utils::{
23    StreamingJobExtraInfo, get_database_resource_group, get_existing_job_resource_group,
24    get_streaming_job_extra_info as fetch_streaming_job_extra_info, get_table_columns,
25    load_streaming_jobs_by_ids,
26};
27
28impl CatalogController {
29    pub async fn get_secret_by_id(&self, secret_id: SecretId) -> MetaResult<PbSecret> {
30        let inner = self.inner.read().await;
31        let (secret, obj) = Secret::find_by_id(secret_id)
32            .find_also_related(Object)
33            .one(&inner.db)
34            .await?
35            .ok_or_else(|| MetaError::catalog_id_not_found("secret", secret_id))?;
36        Ok(ObjectModel(secret, obj.unwrap(), None).into())
37    }
38
39    pub async fn get_object_database_id(
40        &self,
41        object_id: impl Into<ObjectId>,
42    ) -> MetaResult<DatabaseId> {
43        let object_id = object_id.into();
44        let inner = self.inner.read().await;
45        let (database_id,): (Option<DatabaseId>,) = Object::find_by_id(object_id)
46            .select_only()
47            .select_column(object::Column::DatabaseId)
48            .into_tuple()
49            .one(&inner.db)
50            .await?
51            .ok_or_else(|| MetaError::catalog_id_not_found("object", object_id))?;
52        Ok(database_id.ok_or_else(|| anyhow!("object has no database id: {object_id}"))?)
53    }
54
55    pub async fn get_connection_by_id(
56        &self,
57        connection_id: ConnectionId,
58    ) -> MetaResult<PbConnection> {
59        let inner = self.inner.read().await;
60        let (conn, obj) = Connection::find_by_id(connection_id)
61            .find_also_related(Object)
62            .one(&inner.db)
63            .await?
64            .ok_or_else(|| MetaError::catalog_id_not_found("connection", connection_id))?;
65
66        Ok(ObjectModel(conn, obj.unwrap(), None).into())
67    }
68
69    pub async fn get_table_catalog_by_name(
70        &self,
71        database_id: DatabaseId,
72        schema_id: SchemaId,
73        name: &str,
74    ) -> MetaResult<Option<PbTable>> {
75        let inner = self.inner.read().await;
76        let table_obj = Table::find()
77            .find_also_related(Object)
78            .filter(
79                table::Column::Name
80                    .eq(name)
81                    .and(object::Column::DatabaseId.eq(database_id))
82                    .and(object::Column::SchemaId.eq(schema_id)),
83            )
84            .one(&inner.db)
85            .await?;
86        if let Some((table, obj)) = table_obj {
87            let streaming_job = streaming_job::Entity::find_by_id(table.job_id())
88                .one(&inner.db)
89                .await?;
90            Ok(Some(ObjectModel(table, obj.unwrap(), streaming_job).into()))
91        } else {
92            Ok(None)
93        }
94    }
95
96    pub async fn get_table_by_name(
97        &self,
98        database_name: &str,
99        table_name: &str,
100    ) -> MetaResult<Option<PbTable>> {
101        let inner = self.inner.read().await;
102        let table_obj = Table::find()
103            .find_also_related(Object)
104            .join(JoinType::InnerJoin, object::Relation::Database2.def())
105            .filter(
106                table::Column::Name
107                    .eq(table_name)
108                    .and(database::Column::Name.eq(database_name)),
109            )
110            .one(&inner.db)
111            .await?;
112        if let Some((table, obj)) = table_obj {
113            let streaming_job = streaming_job::Entity::find_by_id(table.job_id())
114                .one(&inner.db)
115                .await?;
116            Ok(Some(ObjectModel(table, obj.unwrap(), streaming_job).into()))
117        } else {
118            Ok(None)
119        }
120    }
121
122    pub async fn get_table_associated_source_id(
123        &self,
124        table_id: TableId,
125    ) -> MetaResult<Option<SourceId>> {
126        let inner = self.inner.read().await;
127        Table::find_by_id(table_id)
128            .select_only()
129            .select_column(table::Column::OptionalAssociatedSourceId)
130            .into_tuple()
131            .one(&inner.db)
132            .await?
133            .ok_or_else(|| MetaError::catalog_id_not_found("table", table_id))
134    }
135
136    pub async fn get_table_by_associate_source_id(
137        &self,
138        associated_source_id: SourceId,
139    ) -> MetaResult<PbTable> {
140        let inner = self.inner.read().await;
141        let table_obj = Table::find()
142            .find_also_related(Object)
143            .filter(table::Column::OptionalAssociatedSourceId.eq(associated_source_id))
144            .one(&inner.db)
145            .await?
146            .ok_or_else(|| {
147                MetaError::catalog_id_not_found("table associated source", associated_source_id)
148            })?;
149        let (table, obj) = table_obj;
150        let streaming_job = streaming_job::Entity::find_by_id(table.job_id())
151            .one(&inner.db)
152            .await?;
153        Ok(ObjectModel(table, obj.unwrap(), streaming_job).into())
154    }
155
156    pub async fn get_table_by_id(&self, table_id: TableId) -> MetaResult<PbTable> {
157        let inner = self.inner.read().await;
158        let table_obj = Table::find_by_id(table_id)
159            .find_also_related(Object)
160            .one(&inner.db)
161            .await?;
162        if let Some((table, obj)) = table_obj {
163            let streaming_job = streaming_job::Entity::find_by_id(table.job_id())
164                .one(&inner.db)
165                .await?;
166            Ok(ObjectModel(table, obj.unwrap(), streaming_job).into())
167        } else {
168            Err(MetaError::catalog_id_not_found("table", table_id))
169        }
170    }
171
172    pub async fn get_user_created_table_by_ids(
173        &self,
174        job_ids: impl Iterator<Item = JobId>,
175    ) -> MetaResult<Vec<PbTable>> {
176        let inner = self.inner.read().await;
177        self.get_user_created_table_by_ids_in_txn(&inner.db, job_ids)
178            .await
179    }
180
181    pub async fn get_user_created_table_by_ids_in_txn<C>(
182        &self,
183        txn: &C,
184        job_ids: impl Iterator<Item = JobId>,
185    ) -> MetaResult<Vec<PbTable>>
186    where
187        C: ConnectionTrait,
188    {
189        let table_objs = Table::find()
190            .find_also_related(Object)
191            .filter(
192                table::Column::TableId
193                    .is_in(job_ids.map(|job_id| job_id.as_mv_table_id()).collect_vec())
194                    .and(table::Column::TableType.eq(TableType::Table)),
195            )
196            .all(txn)
197            .await?;
198        let streaming_jobs = streaming_job::Entity::find()
199            .filter(
200                streaming_job::Column::JobId.is_in(
201                    table_objs
202                        .iter()
203                        .map(|(table, _)| table.job_id())
204                        .collect_vec(),
205                ),
206            )
207            .all(txn)
208            .await?
209            .into_iter()
210            .map(|job| (job.job_id, job))
211            .collect::<HashMap<JobId, streaming_job::Model>>();
212        Ok(table_objs
213            .into_iter()
214            .map(|(table, obj)| {
215                let job_id = table.job_id();
216                let streaming_job = streaming_jobs.get(&job_id).cloned();
217                ObjectModel(table, obj.unwrap(), streaming_job).into()
218            })
219            .collect())
220    }
221
222    pub async fn get_table_by_ids(
223        &self,
224        table_ids: Vec<TableId>,
225        include_dropped_table: bool,
226    ) -> MetaResult<Vec<PbTable>> {
227        let inner = self.inner.read().await;
228        let table_objs = Table::find()
229            .find_also_related(Object)
230            .filter(table::Column::TableId.is_in(table_ids.clone()))
231            .all(&inner.db)
232            .await?;
233        let streaming_jobs = load_streaming_jobs_by_ids(
234            &inner.db,
235            table_objs.iter().map(|(table, _)| table.job_id()),
236        )
237        .await?;
238        let tables = table_objs.into_iter().map(|(table, obj)| {
239            let job_id = table.job_id();
240            let streaming_job = streaming_jobs.get(&job_id).cloned();
241            ObjectModel(table, obj.unwrap(), streaming_job).into()
242        });
243        let tables = if include_dropped_table {
244            tables
245                .chain(inner.dropped_tables.iter().filter_map(|(id, t)| {
246                    if table_ids.contains(id) {
247                        Some(t.clone())
248                    } else {
249                        None
250                    }
251                }))
252                .collect()
253        } else {
254            tables.collect()
255        };
256        Ok(tables)
257    }
258
259    pub async fn get_table_columns(&self, id: TableId) -> MetaResult<Vec<ColumnCatalog>> {
260        let inner = self.inner.read().await;
261        Ok(get_table_columns(&inner.db, id)
262            .await?
263            .to_protobuf()
264            .into_iter()
265            .map(|col| col.into())
266            .collect())
267    }
268
269    pub async fn get_sink_by_id(&self, sink_id: SinkId) -> MetaResult<Option<PbSink>> {
270        let inner = self.inner.read().await;
271        inner.get_sink_by_id(sink_id).await
272    }
273
274    pub async fn get_sink_auto_refresh_schema_from(
275        &self,
276        table_id: TableId,
277    ) -> MetaResult<Vec<PbSink>> {
278        let inner = self.inner.read().await;
279        let sink_objs = Sink::find()
280            .find_also_related(Object)
281            .filter(sink::Column::AutoRefreshSchemaFromTable.eq(table_id))
282            .all(&inner.db)
283            .await?;
284        let streaming_jobs = load_streaming_jobs_by_ids(
285            &inner.db,
286            sink_objs.iter().map(|(sink, _)| sink.sink_id.as_job_id()),
287        )
288        .await?;
289        Ok(sink_objs
290            .into_iter()
291            .map(|(sink, obj)| {
292                let streaming_job = streaming_jobs.get(&sink.sink_id.as_job_id()).cloned();
293                ObjectModel(sink, obj.unwrap(), streaming_job).into()
294            })
295            .collect())
296    }
297
298    pub async fn get_sink_state_table_ids(&self, sink_id: SinkId) -> MetaResult<Vec<TableId>> {
299        let inner = self.inner.read().await;
300        let tables: Vec<I32Array> = Fragment::find()
301            .select_only()
302            .column(fragment::Column::StateTableIds)
303            .filter(fragment::Column::JobId.eq(sink_id))
304            .into_tuple()
305            .all(&inner.db)
306            .await?;
307        Ok(tables
308            .into_iter()
309            .flat_map(|ids| {
310                ids.into_inner()
311                    .into_iter()
312                    .map(|table_id| TableId::new(table_id as _))
313            })
314            .collect())
315    }
316
317    pub async fn get_subscription_by_id(
318        &self,
319        subscription_id: SubscriptionId,
320    ) -> MetaResult<PbSubscription> {
321        let inner = self.inner.read().await;
322        let subscription_objs = Subscription::find()
323            .find_also_related(Object)
324            .filter(subscription::Column::SubscriptionId.eq(subscription_id))
325            .all(&inner.db)
326            .await?;
327        let subscription: PbSubscription = subscription_objs
328            .into_iter()
329            .map(|(subscription, obj)| ObjectModel(subscription, obj.unwrap(), None).into())
330            .find_or_first(|_| true)
331            .ok_or_else(|| anyhow!("cannot find subscription with id {}", subscription_id))?;
332
333        Ok(subscription)
334    }
335
336    pub async fn get_mv_depended_subscriptions(
337        &self,
338        database_id: Option<DatabaseId>,
339    ) -> MetaResult<HashMap<TableId, HashMap<SubscriptionId, u64>>> {
340        let inner = self.inner.read().await;
341        let select = Subscription::find()
342            .select_only()
343            .select_column(subscription::Column::SubscriptionId)
344            .select_column(subscription::Column::DependentTableId)
345            .select_column(subscription::Column::RetentionSeconds);
346        let select = if let Some(database_id) = database_id {
347            select
348                .join(JoinType::InnerJoin, subscription::Relation::Object.def())
349                .filter(object::Column::DatabaseId.eq(database_id))
350        } else {
351            select
352        };
353        let subscription_objs: Vec<(SubscriptionId, TableId, i64)> =
354            select.into_tuple().all(&inner.db).await?;
355        let mut map: HashMap<_, HashMap<_, _>> = HashMap::new();
356        // Write object at the same time we write subscription, so we must be able to get obj
357        for (subscription_id, dependent_table_id, retention_seconds) in subscription_objs {
358            map.entry(dependent_table_id)
359                .or_default()
360                .insert(subscription_id, retention_seconds as _);
361        }
362        Ok(map)
363    }
364
365    pub async fn get_all_table_options(&self) -> MetaResult<HashMap<TableId, TableOption>> {
366        let inner = self.inner.read().await;
367        let table_options: Vec<(TableId, Option<i32>)> = Table::find()
368            .select_only()
369            .columns([table::Column::TableId, table::Column::RetentionSeconds])
370            .into_tuple::<(TableId, Option<i32>)>()
371            .all(&inner.db)
372            .await?;
373
374        Ok(table_options
375            .into_iter()
376            .map(|(id, retention_seconds)| {
377                (
378                    id,
379                    TableOption {
380                        retention_seconds: retention_seconds.map(|i| i.try_into().unwrap()),
381                    },
382                )
383            })
384            .collect())
385    }
386
387    pub async fn get_all_streaming_parallelisms(
388        &self,
389    ) -> MetaResult<HashMap<ObjectId, StreamingParallelism>> {
390        let inner = self.inner.read().await;
391
392        let job_parallelisms = StreamingJob::find()
393            .select_only()
394            .columns([
395                streaming_job::Column::JobId,
396                streaming_job::Column::Parallelism,
397            ])
398            .into_tuple::<(ObjectId, StreamingParallelism)>()
399            .all(&inner.db)
400            .await?;
401
402        Ok(job_parallelisms
403            .into_iter()
404            .collect::<HashMap<ObjectId, StreamingParallelism>>())
405    }
406
407    pub async fn get_table_name_type_mapping(
408        &self,
409    ) -> MetaResult<HashMap<TableId, (String, String)>> {
410        let inner = self.inner.read().await;
411        let table_name_types: Vec<(TableId, String, TableType)> = Table::find()
412            .select_only()
413            .columns([
414                table::Column::TableId,
415                table::Column::Name,
416                table::Column::TableType,
417            ])
418            .into_tuple()
419            .all(&inner.db)
420            .await?;
421        Ok(table_name_types
422            .into_iter()
423            .map(|(id, name, table_type)| {
424                (
425                    id,
426                    (name, PbTableType::from(table_type).as_str_name().to_owned()),
427                )
428            })
429            .collect())
430    }
431
432    pub async fn get_table_by_cdc_table_id(
433        &self,
434        cdc_table_id: &String,
435    ) -> MetaResult<Vec<PbTable>> {
436        let inner = self.inner.read().await;
437        let table_objs = Table::find()
438            .find_also_related(Object)
439            .filter(table::Column::CdcTableId.eq(cdc_table_id))
440            .all(&inner.db)
441            .await?;
442        let streaming_jobs = load_streaming_jobs_by_ids(
443            &inner.db,
444            table_objs.iter().map(|(table, _)| table.job_id()),
445        )
446        .await?;
447        Ok(table_objs
448            .into_iter()
449            .map(|(table, obj)| {
450                let job_id = table.job_id();
451                let streaming_job = streaming_jobs.get(&job_id).cloned();
452                ObjectModel(table, obj.unwrap(), streaming_job).into()
453            })
454            .collect())
455    }
456
457    pub async fn get_created_table_ids(&self) -> MetaResult<Vec<TableId>> {
458        let inner = self.inner.read().await;
459
460        // created table ids.
461        let mut table_ids: Vec<TableId> = StreamingJob::find()
462            .select_only()
463            .column(streaming_job::Column::JobId)
464            .filter(streaming_job::Column::JobStatus.eq(JobStatus::Created))
465            .into_tuple()
466            .all(&inner.db)
467            .await?;
468
469        // internal table ids.
470        let internal_table_ids: Vec<TableId> = Table::find()
471            .select_only()
472            .column(table::Column::TableId)
473            .filter(table::Column::BelongsToJobId.is_in(table_ids.clone()))
474            .into_tuple()
475            .all(&inner.db)
476            .await?;
477        table_ids.extend(internal_table_ids);
478
479        Ok(table_ids)
480    }
481
482    /// Returns column ids of versioned tables.
483    /// Being versioned implies using `ColumnAwareSerde`.
484    pub async fn get_versioned_table_schemas(
485        &self,
486    ) -> MetaResult<HashMap<risingwave_common::catalog::TableId, Vec<i32>>> {
487        let res = self
488            .list_all_state_tables()
489            .await?
490            .into_iter()
491            .filter_map(|t| {
492                if t.version.is_some() {
493                    let ret = (
494                        t.id,
495                        t.columns
496                            .iter()
497                            .map(|c| c.column_desc.as_ref().unwrap().column_id)
498                            .collect_vec(),
499                    );
500                    return Some(ret);
501                }
502                None
503            })
504            .collect();
505        Ok(res)
506    }
507
508    pub async fn get_existing_job_resource_group(
509        &self,
510        streaming_job_id: JobId,
511    ) -> MetaResult<String> {
512        let inner = self.inner.read().await;
513        get_existing_job_resource_group(&inner.db, streaming_job_id).await
514    }
515
516    pub async fn get_database_resource_group(&self, database_id: DatabaseId) -> MetaResult<String> {
517        let inner = self.inner.read().await;
518        get_database_resource_group(&inner.db, database_id).await
519    }
520
521    pub async fn get_existing_job_resource_groups(
522        &self,
523        streaming_job_ids: Vec<JobId>,
524    ) -> MetaResult<HashMap<JobId, String>> {
525        let inner = self.inner.read().await;
526        let mut resource_groups = HashMap::new();
527        for job_id in streaming_job_ids {
528            let resource_group = get_existing_job_resource_group(&inner.db, job_id).await?;
529            resource_groups.insert(job_id, resource_group);
530        }
531
532        Ok(resource_groups)
533    }
534
535    pub async fn get_existing_job_database_resource_group(
536        &self,
537        streaming_job_id: JobId,
538    ) -> MetaResult<String> {
539        let inner = self.inner.read().await;
540        let database_id: DatabaseId = StreamingJob::find_by_id(streaming_job_id)
541            .select_only()
542            .join(JoinType::InnerJoin, streaming_job::Relation::Object.def())
543            .column(object::Column::DatabaseId)
544            .into_tuple()
545            .one(&inner.db)
546            .await?
547            .ok_or_else(|| MetaError::catalog_id_not_found("streaming job", streaming_job_id))?;
548
549        get_database_resource_group(&inner.db, database_id).await
550    }
551
552    pub async fn get_job_streaming_parallelisms(
553        &self,
554        streaming_job_id: JobId,
555    ) -> MetaResult<StreamingParallelism> {
556        let inner = self.inner.read().await;
557
558        let job_parallelism: StreamingParallelism = StreamingJob::find_by_id(streaming_job_id)
559            .select_only()
560            .column(streaming_job::Column::Parallelism)
561            .into_tuple()
562            .one(&inner.db)
563            .await?
564            .ok_or_else(|| MetaError::catalog_id_not_found("streaming job", streaming_job_id))?;
565
566        Ok(job_parallelism)
567    }
568
569    pub async fn get_job_parallelisms(
570        &self,
571        streaming_job_id: JobId,
572    ) -> MetaResult<
573        Option<(
574            StreamingParallelism,
575            Option<String>,
576            Option<StreamingParallelism>,
577            Option<String>,
578        )>,
579    > {
580        let inner = self.inner.read().await;
581
582        let parallelisms = StreamingJob::find_by_id(streaming_job_id)
583            .select_only()
584            .columns([
585                streaming_job::Column::Parallelism,
586                streaming_job::Column::AdaptiveParallelismStrategy,
587                streaming_job::Column::BackfillParallelism,
588                streaming_job::Column::BackfillAdaptiveParallelismStrategy,
589            ])
590            .into_tuple::<(
591                StreamingParallelism,
592                Option<String>,
593                Option<StreamingParallelism>,
594                Option<String>,
595            )>()
596            .one(&inner.db)
597            .await?;
598
599        Ok(parallelisms)
600    }
601
602    pub async fn get_fragment_streaming_job_id(
603        &self,
604        fragment_id: FragmentId,
605    ) -> MetaResult<JobId> {
606        let inner = self.inner.read().await;
607        let job_id: JobId = Fragment::find_by_id(fragment_id)
608            .select_only()
609            .column(fragment::Column::JobId)
610            .into_tuple()
611            .one(&inner.db)
612            .await?
613            .ok_or_else(|| MetaError::catalog_id_not_found("fragment", fragment_id))?;
614        Ok(job_id)
615    }
616
617    pub async fn list_streaming_job_with_database(
618        &self,
619    ) -> MetaResult<HashMap<DatabaseId, Vec<JobId>>> {
620        let inner = self.inner.read().await;
621        let database_objects: Vec<(DatabaseId, JobId)> = StreamingJob::find()
622            .select_only()
623            .column(object::Column::DatabaseId)
624            .column(streaming_job::Column::JobId)
625            .join(JoinType::LeftJoin, streaming_job::Relation::Object.def())
626            .into_tuple()
627            .all(&inner.db)
628            .await?;
629
630        Ok(database_objects.into_iter().into_group_map())
631    }
632
633    // Output: Vec<(table id, db name, schema name, table name, resource group, table type)>
634    pub async fn list_table_objects(
635        &self,
636    ) -> MetaResult<Vec<(TableId, String, String, String, String, TableType)>> {
637        let inner = self.inner.read().await;
638        Ok(Object::find()
639            .select_only()
640            .join(JoinType::InnerJoin, object::Relation::Table.def())
641            .join(JoinType::InnerJoin, object::Relation::Database2.def())
642            .join(JoinType::InnerJoin, object::Relation::Schema2.def())
643            .column(object::Column::Oid)
644            .column(database::Column::Name)
645            .column(schema::Column::Name)
646            .column(table::Column::Name)
647            .column(database::Column::ResourceGroup)
648            .column(table::Column::TableType)
649            .into_tuple()
650            .all(&inner.db)
651            .await?)
652    }
653
654    // Output: Vec<(source id, db name, schema name, source name, resource group)>
655    pub async fn list_source_objects(
656        &self,
657    ) -> MetaResult<Vec<(TableId, String, String, String, String)>> {
658        let inner = self.inner.read().await;
659        Ok(Object::find()
660            .select_only()
661            .join(JoinType::InnerJoin, object::Relation::Source.def())
662            .join(JoinType::InnerJoin, object::Relation::Database2.def())
663            .join(JoinType::InnerJoin, object::Relation::Schema2.def())
664            .column(object::Column::Oid)
665            .column(database::Column::Name)
666            .column(schema::Column::Name)
667            .column(source::Column::Name)
668            .column(database::Column::ResourceGroup)
669            .into_tuple()
670            .all(&inner.db)
671            .await?)
672    }
673
674    // Output: Vec<(sink id, db name, schema name, sink name, resource group)>
675    pub async fn list_sink_objects(
676        &self,
677    ) -> MetaResult<Vec<(TableId, String, String, String, String)>> {
678        let inner = self.inner.read().await;
679        Ok(Object::find()
680            .select_only()
681            .join(JoinType::InnerJoin, object::Relation::Sink.def())
682            .join(JoinType::InnerJoin, object::Relation::Database2.def())
683            .join(JoinType::InnerJoin, object::Relation::Schema2.def())
684            .column(object::Column::Oid)
685            .column(database::Column::Name)
686            .column(schema::Column::Name)
687            .column(sink::Column::Name)
688            .column(database::Column::ResourceGroup)
689            .into_tuple()
690            .all(&inner.db)
691            .await?)
692    }
693
694    // Output: Vec<(id, db name, schema name, relation name, type)>
695    pub async fn list_relation_objects_by_ids(
696        &self,
697        ids: &HashSet<ObjectId>,
698    ) -> MetaResult<Vec<(ObjectId, String, String, String, String)>> {
699        if ids.is_empty() {
700            return Ok(vec![]);
701        }
702        let inner = self.inner.read().await;
703        let ids: Vec<_> = ids.iter().cloned().collect();
704
705        let tables: Vec<(ObjectId, String, String, String, TableType)> = Object::find()
706            .select_only()
707            .join(JoinType::InnerJoin, object::Relation::Table.def())
708            .join(JoinType::InnerJoin, object::Relation::Database2.def())
709            .join(JoinType::InnerJoin, object::Relation::Schema2.def())
710            .filter(object::Column::Oid.is_in(ids.clone()))
711            .column(object::Column::Oid)
712            .column(database::Column::Name)
713            .column(schema::Column::Name)
714            .column(table::Column::Name)
715            .column(table::Column::TableType)
716            .into_tuple()
717            .all(&inner.db)
718            .await?;
719
720        let sources: Vec<(ObjectId, String, String, String)> = Object::find()
721            .select_only()
722            .join(JoinType::InnerJoin, object::Relation::Source.def())
723            .join(JoinType::InnerJoin, object::Relation::Database2.def())
724            .join(JoinType::InnerJoin, object::Relation::Schema2.def())
725            .filter(object::Column::Oid.is_in(ids.clone()))
726            .column(object::Column::Oid)
727            .column(database::Column::Name)
728            .column(schema::Column::Name)
729            .column(source::Column::Name)
730            .into_tuple()
731            .all(&inner.db)
732            .await?;
733
734        let mut result = Vec::with_capacity(tables.len() + sources.len());
735        result.extend(
736            tables
737                .into_iter()
738                .map(|(id, db, schema, name, table_type)| {
739                    let relation_type = PbTableType::from(table_type).as_str_name();
740                    (id, db, schema, name, relation_type.to_owned())
741                }),
742        );
743        result.extend(
744            sources
745                .into_iter()
746                .map(|(id, db, schema, name)| (id, db, schema, name, "source".to_owned())),
747        );
748        Ok(result)
749    }
750
751    pub async fn get_streaming_job_status(&self, streaming_job_id: JobId) -> MetaResult<JobStatus> {
752        let inner = self.inner.read().await;
753        let status = StreamingJob::find_by_id(streaming_job_id)
754            .select_only()
755            .column(streaming_job::Column::JobStatus)
756            .into_tuple()
757            .one(&inner.db)
758            .await?
759            .ok_or_else(|| MetaError::catalog_id_not_found("streaming job", streaming_job_id))?;
760        Ok(status)
761    }
762
763    pub async fn get_streaming_job_extra_info(
764        &self,
765        job_ids: Vec<JobId>,
766    ) -> MetaResult<HashMap<JobId, StreamingJobExtraInfo>> {
767        let inner = self.inner.read().await;
768        let txn = inner.db.begin().await?;
769
770        self.get_streaming_job_extra_info_in_txn(&txn, job_ids)
771            .await
772    }
773
774    pub async fn get_streaming_job_extra_info_in_txn<C>(
775        &self,
776        txn: &C,
777        job_ids: Vec<JobId>,
778    ) -> MetaResult<HashMap<JobId, StreamingJobExtraInfo>>
779    where
780        C: ConnectionTrait,
781    {
782        fetch_streaming_job_extra_info(txn, job_ids).await
783    }
784}
785
786impl CatalogControllerInner {
787    pub(crate) async fn get_sink_by_id(&self, sink_id: SinkId) -> MetaResult<Option<PbSink>> {
788        let sink_objs = Sink::find_by_id(sink_id)
789            .find_also_related(Object)
790            .one(&self.db)
791            .await?;
792        if let Some((sink, obj)) = sink_objs {
793            let streaming_job = streaming_job::Entity::find_by_id(sink.sink_id.as_job_id())
794                .one(&self.db)
795                .await?;
796            Ok(Some(ObjectModel(sink, obj.unwrap(), streaming_job).into()))
797        } else {
798            Ok(None)
799        }
800    }
801}
risingwave_meta/controller/catalog/get_op.rs

risingwave_meta/controller/catalog/
get_op.rs
