risingwave_meta/controller/catalog/
list_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 risingwave_common::catalog::FragmentTypeMask;
16use risingwave_common::id::JobId;
17use risingwave_meta_model::refresh_job::{self, RefreshState};
18use risingwave_pb::meta::ObjectDependency as PbObjectDependency;
19use sea_orm::prelude::DateTime;
20
21use super::*;
22use crate::controller::fragment::FragmentTypeMaskExt;
23use crate::controller::utils::load_streaming_jobs_by_ids;
24
25impl CatalogController {
26    pub async fn list_time_travel_table_ids(&self) -> MetaResult<Vec<TableId>> {
27        self.inner.read().await.list_time_travel_table_ids().await
28    }
29
30    pub async fn list_refresh_jobs(&self) -> MetaResult<Vec<refresh_job::Model>> {
31        let inner = self.inner.read().await;
32        Ok(RefreshJob::find().all(&inner.db).await?)
33    }
34
35    /// The refresh jobs that are not idle, of one database or of all.
36    pub async fn list_refreshing_jobs(
37        &self,
38        database_id: Option<DatabaseId>,
39    ) -> MetaResult<Vec<refresh_job::Model>> {
40        let inner = self.inner.read().await;
41        let database_cond = database_id
42            .map(|database_id| object::Column::DatabaseId.eq(database_id))
43            .unwrap_or_else(|| SimpleExpr::from(true));
44        Ok(RefreshJob::find()
45            .join(JoinType::InnerJoin, refresh_job::Relation::Table.def())
46            .join(JoinType::InnerJoin, table::Relation::Object1.def())
47            .filter(
48                refresh_job::Column::CurrentStatus
49                    .ne(RefreshState::Idle)
50                    .and(database_cond),
51            )
52            .all(&inner.db)
53            .await?)
54    }
55
56    /// The refresh state of a table, or `None` if the table is not refreshable.
57    pub async fn get_refresh_job_state(
58        &self,
59        table_id: TableId,
60    ) -> MetaResult<Option<RefreshState>> {
61        let inner = self.inner.read().await;
62        let state: Option<(RefreshState,)> = RefreshJob::find_by_id(table_id)
63            .select_only()
64            .select_column(refresh_job::Column::CurrentStatus)
65            .into_tuple()
66            .one(&inner.db)
67            .await?;
68        Ok(state.map(|(state,)| state))
69    }
70
71    pub async fn list_refreshable_table_ids(&self) -> MetaResult<Vec<TableId>> {
72        let inner = self.inner.read().await;
73        Ok(Table::find()
74            .select_only()
75            .column(table::Column::TableId)
76            .filter(table::Column::Refreshable.eq(true))
77            .into_tuple()
78            .all(&inner.db)
79            .await?)
80    }
81
82    pub async fn list_stream_job_desc_for_telemetry(
83        &self,
84    ) -> MetaResult<Vec<MetaTelemetryJobDesc>> {
85        let inner = self.inner.read().await;
86        let info: Vec<(TableId, Option<Property>)> = Table::find()
87            .select_only()
88            .column(table::Column::TableId)
89            .column(source::Column::WithProperties)
90            .join(JoinType::LeftJoin, table::Relation::Source.def())
91            .filter(
92                table::Column::TableType
93                    .eq(TableType::Table)
94                    .or(table::Column::TableType.eq(TableType::MaterializedView)),
95            )
96            .into_tuple()
97            .all(&inner.db)
98            .await?;
99
100        Ok(info
101            .into_iter()
102            .map(|(table_id, properties)| {
103                let connector_info = if let Some(inner_props) = properties {
104                    inner_props
105                        .inner_ref()
106                        .get(UPSTREAM_SOURCE_KEY)
107                        .map(|v| v.to_lowercase())
108                } else {
109                    None
110                };
111                MetaTelemetryJobDesc {
112                    table_id,
113                    connector: connector_info,
114                    optimization: vec![],
115                }
116            })
117            .collect())
118    }
119
120    pub async fn list_creating_jobs(
121        &self,
122        include_initial: bool,
123        database_id: Option<DatabaseId>,
124    ) -> MetaResult<Vec<(JobId, String, DateTime, CreateType, bool)>> {
125        let inner = self.inner.read().await;
126        let status_cond = if include_initial {
127            streaming_job::Column::JobStatus.is_in([JobStatus::Initial, JobStatus::Creating])
128        } else {
129            streaming_job::Column::JobStatus.eq(JobStatus::Creating)
130        };
131        let database_cond = database_id
132            .map(|database_id| object::Column::DatabaseId.eq(database_id))
133            .unwrap_or_else(|| SimpleExpr::from(true));
134        let filter_cond = status_cond.and(database_cond);
135        let object_columns = [object::Column::InitializedAt];
136        let streaming_job_columns = [
137            streaming_job::Column::CreateType,
138            streaming_job::Column::IsServerlessBackfill,
139        ];
140        let mut table_info: Vec<(JobId, String, DateTime, CreateType, bool)> = Table::find()
141            .select_only()
142            .columns([table::Column::TableId, table::Column::Definition])
143            .columns(object_columns)
144            .columns(streaming_job_columns)
145            .join(JoinType::LeftJoin, table::Relation::Object1.def())
146            .join(JoinType::LeftJoin, object::Relation::StreamingJob.def())
147            .filter(filter_cond.clone())
148            .into_tuple()
149            .all(&inner.db)
150            .await?;
151        let sink_info: Vec<(JobId, String, DateTime, CreateType, bool)> = Sink::find()
152            .select_only()
153            .columns([sink::Column::SinkId, sink::Column::Definition])
154            .columns(object_columns)
155            .columns(streaming_job_columns)
156            .join(JoinType::LeftJoin, sink::Relation::Object.def())
157            .join(JoinType::LeftJoin, object::Relation::StreamingJob.def())
158            .filter(filter_cond)
159            .into_tuple()
160            .all(&inner.db)
161            .await?;
162
163        table_info.extend(sink_info);
164
165        Ok(table_info)
166    }
167
168    pub async fn list_databases(&self) -> MetaResult<Vec<PbDatabase>> {
169        let inner = self.inner.read().await;
170        inner.list_databases().await
171    }
172
173    pub async fn list_all_object_dependencies(&self) -> MetaResult<Vec<PbObjectDependency>> {
174        let inner = self.inner.read().await;
175        let txn = inner.db.begin().await?;
176        let dependencies = list_object_dependencies(&txn, true).await?;
177        txn.commit().await?;
178        Ok(dependencies)
179    }
180
181    pub async fn list_created_object_dependencies(&self) -> MetaResult<Vec<PbObjectDependency>> {
182        let inner = self.inner.read().await;
183        let txn = inner.db.begin().await?;
184        let dependencies = list_object_dependencies(&txn, false).await?;
185        txn.commit().await?;
186        Ok(dependencies)
187    }
188
189    pub async fn list_schemas(&self) -> MetaResult<Vec<PbSchema>> {
190        let inner = self.inner.read().await;
191        inner.list_schemas().await
192    }
193
194    /// [`Self::list_tables_by_type`] with all types.
195    pub async fn list_all_state_tables(&self) -> MetaResult<Vec<PbTable>> {
196        let inner = self.inner.read().await;
197        inner.list_all_state_tables().await
198    }
199
200    pub async fn list_readonly_table_ids(&self, schema_id: SchemaId) -> MetaResult<Vec<TableId>> {
201        let inner = self.inner.read().await;
202        let table_ids: Vec<TableId> = Table::find()
203            .select_only()
204            .column(table::Column::TableId)
205            .join(JoinType::InnerJoin, table::Relation::Object1.def())
206            .filter(
207                object::Column::SchemaId
208                    .eq(schema_id)
209                    .and(table::Column::TableType.ne(TableType::Table)),
210            )
211            .into_tuple()
212            .all(&inner.db)
213            .await?;
214        Ok(table_ids)
215    }
216
217    pub async fn list_dml_table_ids(&self, schema_id: SchemaId) -> MetaResult<Vec<TableId>> {
218        let inner = self.inner.read().await;
219        let table_ids: Vec<TableId> = Table::find()
220            .select_only()
221            .column(table::Column::TableId)
222            .join(JoinType::InnerJoin, table::Relation::Object1.def())
223            .filter(
224                object::Column::SchemaId
225                    .eq(schema_id)
226                    .and(table::Column::TableType.eq(TableType::Table)),
227            )
228            .into_tuple()
229            .all(&inner.db)
230            .await?;
231        Ok(table_ids)
232    }
233
234    pub async fn list_view_ids(&self, schema_id: SchemaId) -> MetaResult<Vec<ViewId>> {
235        let inner = self.inner.read().await;
236        let view_ids: Vec<ViewId> = View::find()
237            .select_only()
238            .column(view::Column::ViewId)
239            .join(JoinType::InnerJoin, view::Relation::Object.def())
240            .filter(object::Column::SchemaId.eq(schema_id))
241            .into_tuple()
242            .all(&inner.db)
243            .await?;
244        Ok(view_ids)
245    }
246
247    /// Use [`Self::list_all_state_tables`] to get all types.
248    pub async fn list_tables_by_type(&self, table_type: TableType) -> MetaResult<Vec<PbTable>> {
249        let inner = self.inner.read().await;
250        let table_objs = Table::find()
251            .find_also_related(Object)
252            .filter(table::Column::TableType.eq(table_type))
253            .all(&inner.db)
254            .await?;
255        let streaming_jobs = load_streaming_jobs_by_ids(
256            &inner.db,
257            table_objs.iter().map(|(table, _)| table.job_id()),
258        )
259        .await?;
260        Ok(table_objs
261            .into_iter()
262            .map(|(table, obj)| {
263                let job_id = table.job_id();
264                let streaming_job = streaming_jobs.get(&job_id).cloned();
265                ObjectModel(table, obj.unwrap(), streaming_job).into()
266            })
267            .collect())
268    }
269
270    pub async fn list_sources(&self) -> MetaResult<Vec<PbSource>> {
271        let inner = self.inner.read().await;
272        inner.list_sources().await
273    }
274
275    // Return a hashmap to distinguish whether each source is shared or not.
276    pub async fn list_source_id_with_shared_types(&self) -> MetaResult<HashMap<SourceId, bool>> {
277        let inner = self.inner.read().await;
278        let source_ids: Vec<(SourceId, Option<StreamSourceInfo>)> = Source::find()
279            .select_only()
280            .columns([source::Column::SourceId, source::Column::SourceInfo])
281            .into_tuple()
282            .all(&inner.db)
283            .await?;
284
285        Ok(source_ids
286            .into_iter()
287            .map(|(source_id, info)| {
288                (
289                    source_id,
290                    info.map(|info| info.to_protobuf().cdc_source_job)
291                        .unwrap_or(false),
292                )
293            })
294            .collect())
295    }
296
297    pub async fn list_connections(&self) -> MetaResult<Vec<PbConnection>> {
298        let inner = self.inner.read().await;
299        let conn_objs = Connection::find()
300            .find_also_related(Object)
301            .all(&inner.db)
302            .await?;
303        Ok(conn_objs
304            .into_iter()
305            .map(|(conn, obj)| ObjectModel(conn, obj.unwrap(), None).into())
306            .collect())
307    }
308
309    pub async fn list_source_ids(&self, schema_id: SchemaId) -> MetaResult<Vec<SourceId>> {
310        let inner = self.inner.read().await;
311        let source_ids: Vec<SourceId> = Source::find()
312            .select_only()
313            .column(source::Column::SourceId)
314            .join(JoinType::InnerJoin, source::Relation::Object.def())
315            .filter(object::Column::SchemaId.eq(schema_id))
316            .into_tuple()
317            .all(&inner.db)
318            .await?;
319        Ok(source_ids)
320    }
321
322    pub async fn list_indexes(&self) -> MetaResult<Vec<PbIndex>> {
323        let inner = self.inner.read().await;
324        inner.list_indexes().await
325    }
326
327    pub async fn list_sinks(&self) -> MetaResult<Vec<PbSink>> {
328        let inner = self.inner.read().await;
329        inner.list_sinks().await
330    }
331
332    pub async fn list_subscriptions(&self) -> MetaResult<Vec<PbSubscription>> {
333        let inner = self.inner.read().await;
334        inner.list_subscriptions().await
335    }
336
337    pub async fn list_views(&self) -> MetaResult<Vec<PbView>> {
338        let inner = self.inner.read().await;
339        inner.list_views().await
340    }
341
342    pub async fn list_users(&self) -> MetaResult<Vec<PbUserInfo>> {
343        let inner = self.inner.read().await;
344        inner.list_users().await
345    }
346
347    pub async fn list_functions(&self) -> MetaResult<Vec<PbFunction>> {
348        let inner = self.inner.read().await;
349        inner.list_functions().await
350    }
351
352    /// `Unmigrated` refers to table-fragments that have not yet been migrated to the new plan (for now, this means
353    /// table-fragments that do not use `UpstreamSinkUnion` operator to receive multiple upstream sinks)
354    pub async fn list_unmigrated_tables(&self) -> MetaResult<Vec<PbTable>> {
355        let inner = self.inner.read().await;
356
357        let table_objs = Table::find()
358            .filter(table::Column::TableType.eq(TableType::Table))
359            .find_also_related(Object)
360            .join(JoinType::InnerJoin, object::Relation::Fragment.def())
361            .filter(FragmentTypeMask::intersects(FragmentTypeFlag::Mview).and(
362                FragmentTypeMask::disjoint(FragmentTypeFlag::UpstreamSinkUnion),
363            ))
364            .all(&inner.db)
365            .await?;
366        let streaming_jobs = load_streaming_jobs_by_ids(
367            &inner.db,
368            table_objs.iter().map(|(table, _)| table.job_id()),
369        )
370        .await?;
371
372        Ok(table_objs
373            .into_iter()
374            .map(|(table, obj)| {
375                let job_id = table.job_id();
376                let streaming_job = streaming_jobs.get(&job_id).cloned();
377                ObjectModel(table, obj.unwrap(), streaming_job).into()
378            })
379            .collect())
380    }
381
382    pub async fn list_sink_ids(&self, database_id: Option<DatabaseId>) -> MetaResult<Vec<SinkId>> {
383        let inner = self.inner.read().await;
384
385        let mut query = Sink::find().select_only().column(sink::Column::SinkId);
386
387        if let Some(database_id) = database_id {
388            query = query
389                .join(JoinType::InnerJoin, sink::Relation::Object.def())
390                .filter(object::Column::DatabaseId.eq(database_id));
391        }
392
393        let sink_ids: Vec<SinkId> = query.into_tuple().all(&inner.db).await?;
394
395        Ok(sink_ids)
396    }
397}
risingwave_meta/controller/catalog/list_op.rs

risingwave_meta/controller/catalog/
list_op.rs
