risingwave_meta/controller/catalog/
create_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::cast::datetime_to_timestamp_millis;
16use risingwave_common::system_param::{OverrideValidate, Validate};
17use risingwave_common::util::epoch::Epoch;
18use risingwave_pb::common::PbObjectType;
19use risingwave_pb::meta::PbObjectDependency;
20
21use super::*;
22use crate::barrier::SnapshotBackfillInfo;
23
24impl CatalogController {
25    pub(crate) async fn create_object(
26        txn: &DatabaseTransaction,
27        obj_type: ObjectType,
28        owner_id: UserId,
29        belong_to_oid: Option<ObjectId>,
30    ) -> MetaResult<object::Model> {
31        let (database_id, schema_id) = match (obj_type, belong_to_oid) {
32            (ObjectType::Database, None) => (None, None),
33            (ObjectType::Database, Some(_)) => {
34                return Err(MetaError::invalid_parameter(
35                    "database object must not have a parent",
36                ));
37            }
38            (ObjectType::Schema, Some(database_oid)) => {
39                let database = Object::find_by_id(database_oid)
40                    .one(txn)
41                    .await?
42                    .ok_or_else(|| MetaError::catalog_id_not_found("database", database_oid))?;
43                if database.obj_type != ObjectType::Database {
44                    return Err(MetaError::invalid_parameter(
45                        "schema object must belong to a database",
46                    ));
47                }
48                (Some(database_oid.as_database_id()), None)
49            }
50            (ObjectType::Schema, None) => {
51                return Err(MetaError::invalid_parameter(
52                    "schema object must belong to a database",
53                ));
54            }
55            (_, Some(parent_oid)) => {
56                let parent = Object::find_by_id(parent_oid)
57                    .one(txn)
58                    .await?
59                    .ok_or_else(|| MetaError::catalog_id_not_found("parent object", parent_oid))?;
60                if parent.obj_type == ObjectType::Schema {
61                    (parent.database_id, Some(parent.oid.as_schema_id()))
62                } else {
63                    (parent.database_id, parent.schema_id)
64                }
65            }
66            (_, None) => {
67                return Err(MetaError::invalid_parameter(
68                    "non-database object must have a parent",
69                ));
70            }
71        };
72        let active_db = object::ActiveModel {
73            oid: Default::default(),
74            obj_type: Set(obj_type),
75            owner_id: Set(owner_id),
76            schema_id: Set(schema_id),
77            database_id: Set(database_id),
78            belong_to_oid: Set(belong_to_oid),
79            initialized_at: Default::default(),
80            created_at: Default::default(),
81            initialized_at_cluster_version: Set(Some(current_cluster_version())),
82            created_at_cluster_version: Set(Some(current_cluster_version())),
83        };
84        Ok(active_db.insert(txn).await?)
85    }
86
87    pub async fn create_database(
88        &self,
89        db: PbDatabase,
90    ) -> MetaResult<(NotificationVersion, risingwave_meta_model::database::Model)> {
91        // validate first
92        if let Some(ref interval) = db.barrier_interval_ms {
93            OverrideValidate::barrier_interval_ms(interval).map_err(|e| anyhow::anyhow!(e))?;
94        }
95        if let Some(ref frequency) = db.checkpoint_frequency {
96            OverrideValidate::checkpoint_frequency(frequency).map_err(|e| anyhow::anyhow!(e))?;
97        }
98
99        let inner = self.inner.write().await;
100        let owner_id = db.owner as _;
101        let txn = inner.db.begin().await?;
102        ensure_user_id(owner_id, &txn).await?;
103        check_database_name_duplicate(&db.name, &txn).await?;
104
105        let db_obj = Self::create_object(&txn, ObjectType::Database, owner_id, None).await?;
106        let mut db: database::ActiveModel = db.into();
107        db.database_id = Set(db_obj.oid.as_database_id());
108        let db = db.insert(&txn).await?;
109
110        let mut schemas = vec![];
111        for schema_name in iter::once(DEFAULT_SCHEMA_NAME).chain(SYSTEM_SCHEMAS) {
112            let schema_obj =
113                Self::create_object(&txn, ObjectType::Schema, owner_id, Some(db_obj.oid)).await?;
114            let schema = schema::ActiveModel {
115                schema_id: Set(schema_obj.oid.as_schema_id()),
116                name: Set(schema_name.into()),
117            };
118            let schema = schema.insert(&txn).await?;
119            schemas.push(ObjectModel(schema, schema_obj, None).into());
120        }
121        txn.commit().await?;
122
123        let mut notifications = self.frontend_notification_batch();
124        notifications.add(
125            NotificationOperation::Add,
126            NotificationInfo::Database(ObjectModel(db.clone(), db_obj, None).into()),
127        );
128        for schema in schemas {
129            notifications.add(NotificationOperation::Add, NotificationInfo::Schema(schema));
130        }
131        let version = notifications
132            .publish()
133            .await
134            .expect("the database notification batch contains the database");
135
136        Ok((version, db))
137    }
138
139    pub async fn create_schema(&self, schema: PbSchema) -> MetaResult<NotificationVersion> {
140        let inner = self.inner.write().await;
141        let owner_id = schema.owner as _;
142        let txn = inner.db.begin().await?;
143        ensure_user_id(owner_id, &txn).await?;
144        ensure_object_id(ObjectType::Database, schema.database_id, &txn).await?;
145        check_schema_name_duplicate(&schema.name, schema.database_id, &txn).await?;
146
147        let schema_obj = Self::create_object(
148            &txn,
149            ObjectType::Schema,
150            owner_id,
151            Some(schema.database_id.as_object_id()),
152        )
153        .await?;
154        let mut schema: schema::ActiveModel = schema.into();
155        schema.schema_id = Set(schema_obj.oid.as_schema_id());
156        let schema = schema.insert(&txn).await?;
157
158        let updated_user_info =
159            grant_default_privileges_automatically(&txn, schema_obj.oid).await?;
160
161        txn.commit().await?;
162
163        let mut notifications = self.frontend_notification_batch();
164        notifications.add(
165            NotificationOperation::Add,
166            NotificationInfo::Schema(ObjectModel(schema, schema_obj, None).into()),
167        );
168
169        // notify default privileges for schemas
170        Self::add_users_update_notifications(&mut notifications, updated_user_info);
171        let version = notifications
172            .publish()
173            .await
174            .expect("the schema notification batch contains the schema");
175
176        Ok(version)
177    }
178
179    pub async fn create_subscription_catalog(
180        &self,
181        pb_subscription: &mut PbSubscription,
182    ) -> MetaResult<()> {
183        let inner = self.inner.write().await;
184        let txn = inner.db.begin().await?;
185
186        ensure_user_id(pb_subscription.owner as _, &txn).await?;
187        ensure_object_id(ObjectType::Database, pb_subscription.database_id, &txn).await?;
188        ensure_object_id(ObjectType::Schema, pb_subscription.schema_id, &txn).await?;
189        check_subscription_name_duplicate(pb_subscription, &txn).await?;
190
191        let obj = Self::create_object(
192            &txn,
193            ObjectType::Subscription,
194            pb_subscription.owner as _,
195            Some(pb_subscription.schema_id.as_object_id()),
196        )
197        .await?;
198        pb_subscription.id = obj.oid.as_subscription_id();
199        let subscription: subscription::ActiveModel = pb_subscription.clone().into();
200        Subscription::insert(subscription).exec(&txn).await?;
201
202        // record object dependency.
203        ObjectDependency::insert(object_dependency::ActiveModel {
204            oid: Set(pb_subscription.dependent_table_id.into()),
205            used_by: Set(pb_subscription.id.into()),
206            ..Default::default()
207        })
208        .exec(&txn)
209        .await?;
210        txn.commit().await?;
211        Ok(())
212    }
213
214    pub async fn create_source(
215        &self,
216        mut pb_source: PbSource,
217        iceberg_table_id: Option<TableId>,
218    ) -> MetaResult<(SourceId, NotificationVersion)> {
219        let mut inner = self.inner.write().await;
220        let owner_id = pb_source.owner as _;
221        let txn = inner.db.begin().await?;
222        ensure_user_id(owner_id, &txn).await?;
223        ensure_object_id(ObjectType::Database, pb_source.database_id, &txn).await?;
224        ensure_object_id(ObjectType::Schema, pb_source.schema_id, &txn).await?;
225        check_relation_name_duplicate(
226            &pb_source.name,
227            pb_source.database_id,
228            pb_source.schema_id,
229            &txn,
230        )
231        .await?;
232
233        // handle secret ref
234        let secret_ids = get_referred_secret_ids_from_source(&pb_source)?;
235        let connection_ids = get_referred_connection_ids_from_source(&pb_source);
236        let cdc_source_id = pb_source
237            .info
238            .as_ref()
239            .and_then(|info| info.external_table.as_ref())
240            .map(|table| table.source_id);
241        if let Some(cdc_source_id) = cdc_source_id {
242            ensure_object_id(ObjectType::Source, cdc_source_id, &txn).await?;
243        }
244
245        let source_obj = Self::create_object(
246            &txn,
247            ObjectType::Source,
248            owner_id,
249            iceberg_table_id
250                .map(TableId::as_object_id)
251                .or(Some(pb_source.schema_id.as_object_id())),
252        )
253        .await?;
254        let source_id = source_obj.oid.as_source_id();
255        pb_source.id = source_id;
256        if let Some(external_table) = pb_source
257            .info
258            .as_mut()
259            .and_then(|info| info.external_table.as_mut())
260        {
261            external_table.table_id = source_id.as_cdc_table_id();
262            // The connect properties were only needed to validate the descriptor. Consumers
263            // resolve them from the upstream shared source when they are planned, so persisting
264            // a copy here would only make credentials and secret references stale after
265            // `ALTER SOURCE`.
266            external_table.connect_properties.clear();
267            external_table.secret_refs.clear();
268        }
269        let source: source::ActiveModel = pb_source.clone().into();
270        Source::insert(source).exec(&txn).await?;
271
272        // Add secret, connection, and upstream shared CDC source dependencies.
273        let dep_relation_ids = secret_ids
274            .iter()
275            .copied()
276            .map_into()
277            .chain(connection_ids.iter().copied().map_into())
278            .chain(cdc_source_id.map(|id| id.as_object_id()))
279            .collect_vec();
280        let dependencies = if !dep_relation_ids.is_empty() {
281            ObjectDependency::insert_many(dep_relation_ids.into_iter().map(|id| {
282                object_dependency::ActiveModel {
283                    oid: Set(id),
284                    used_by: Set(source_id.as_object_id()),
285                    ..Default::default()
286                }
287            }))
288            .exec(&txn)
289            .await?;
290            secret_ids
291                .iter()
292                .map(|id| PbObjectDependency {
293                    object_id: source_id.as_object_id(),
294                    referenced_object_id: id.as_object_id(),
295                    referenced_object_type: PbObjectType::Secret as _,
296                })
297                .chain(connection_ids.iter().map(|id| PbObjectDependency {
298                    object_id: source_id.as_object_id(),
299                    referenced_object_id: id.as_object_id(),
300                    referenced_object_type: PbObjectType::Connection as _,
301                }))
302                .chain(cdc_source_id.map(|id| PbObjectDependency {
303                    object_id: source_id.as_object_id(),
304                    referenced_object_id: id.as_object_id(),
305                    referenced_object_type: PbObjectType::Source as _,
306                }))
307                .collect_vec()
308        } else {
309            vec![]
310        };
311
312        let mut job_notifications = vec![];
313        let mut updated_user_info = vec![];
314        // check if it belongs to iceberg table
315        if let Some(table_id) = iceberg_table_id {
316            // 1. finish iceberg table job.
317            let table_notifications = self
318                .finish_streaming_job_inner(&txn, table_id.as_job_id())
319                .await?;
320            job_notifications.push((table_id.as_job_id(), table_notifications));
321
322            // 2. finish iceberg sink job.
323            let sink_id = Sink::find()
324                .select_only()
325                .column(sink::Column::SinkId)
326                .join(JoinType::InnerJoin, sink::Relation::Object.def())
327                .filter(
328                    object::Column::BelongToOid
329                        .eq(table_id)
330                        .and(object::Column::ObjType.eq(ObjectType::Sink)),
331                )
332                .into_tuple::<SinkId>()
333                .one(&txn)
334                .await?
335                .ok_or_else(|| {
336                    MetaError::catalog_id_not_found("iceberg sink for table", table_id)
337                })?;
338            let sink_job_id = sink_id.as_job_id();
339            let sink_notifications = self.finish_streaming_job_inner(&txn, sink_job_id).await?;
340            job_notifications.push((sink_job_id, sink_notifications));
341        } else {
342            updated_user_info = grant_default_privileges_automatically(&txn, source_id).await?;
343        }
344
345        txn.commit().await?;
346
347        let mut notifications = self.frontend_notification_batch();
348        let mut finished_job_ids = Vec::with_capacity(job_notifications.len());
349        for (job_id, (op, objects, user_info, dependencies)) in job_notifications {
350            notifications.add(
351                op,
352                NotificationInfo::ObjectGroup(PbObjectGroup {
353                    objects,
354                    dependencies,
355                }),
356            );
357            Self::add_users_update_notifications(&mut notifications, user_info);
358            finished_job_ids.push(job_id);
359        }
360
361        notifications.add(
362            NotificationOperation::Add,
363            NotificationInfo::ObjectGroup(PbObjectGroup {
364                objects: vec![PbObject {
365                    object_info: Some(PbObjectInfo::Source(pb_source)),
366                }],
367                dependencies,
368            }),
369        );
370
371        // notify default privileges for source
372        Self::add_users_update_notifications(&mut notifications, updated_user_info);
373        let version = notifications
374            .publish()
375            .await
376            .expect("the source notification batch contains the source");
377
378        for job_id in finished_job_ids {
379            inner
380                .creating_table_finish_notifier
381                .values_mut()
382                .for_each(|creating_tables| {
383                    if let Some(txs) = creating_tables.remove(&job_id) {
384                        for tx in txs {
385                            let _ = tx.send(Ok(version));
386                        }
387                    }
388                });
389        }
390
391        Ok((source_id, version))
392    }
393
394    pub async fn create_function(
395        &self,
396        mut pb_function: PbFunction,
397    ) -> MetaResult<NotificationVersion> {
398        let inner = self.inner.write().await;
399        let owner_id = pb_function.owner as _;
400        let txn = inner.db.begin().await?;
401        ensure_user_id(owner_id, &txn).await?;
402        ensure_object_id(ObjectType::Database, pb_function.database_id, &txn).await?;
403        ensure_object_id(ObjectType::Schema, pb_function.schema_id, &txn).await?;
404        check_function_signature_duplicate(&pb_function, &txn).await?;
405
406        let function_obj = Self::create_object(
407            &txn,
408            ObjectType::Function,
409            owner_id,
410            Some(pb_function.schema_id.as_object_id()),
411        )
412        .await?;
413        pb_function.id = function_obj.oid.as_function_id();
414        pb_function.created_at_epoch = Some(
415            Epoch::from_unix_millis(datetime_to_timestamp_millis(function_obj.created_at) as _).0,
416        );
417        pb_function.created_at_cluster_version = function_obj.created_at_cluster_version;
418        let function: function::ActiveModel = pb_function.clone().into();
419        Function::insert(function).exec(&txn).await?;
420
421        let updated_user_info =
422            grant_default_privileges_automatically(&txn, function_obj.oid).await?;
423
424        txn.commit().await?;
425
426        let mut notifications = self.frontend_notification_batch();
427        notifications.add(
428            NotificationOperation::Add,
429            NotificationInfo::Function(pb_function),
430        );
431
432        // notify default privileges for functions
433        Self::add_users_update_notifications(&mut notifications, updated_user_info);
434        let version = notifications
435            .publish()
436            .await
437            .expect("the function notification batch contains the function");
438
439        Ok(version)
440    }
441
442    pub async fn create_connection(
443        &self,
444        mut pb_connection: PbConnection,
445    ) -> MetaResult<NotificationVersion> {
446        let inner = self.inner.write().await;
447        let owner_id = pb_connection.owner as _;
448        let txn = inner.db.begin().await?;
449        ensure_user_id(owner_id, &txn).await?;
450        ensure_object_id(ObjectType::Database, pb_connection.database_id, &txn).await?;
451        ensure_object_id(ObjectType::Schema, pb_connection.schema_id, &txn).await?;
452        check_connection_name_duplicate(&pb_connection, &txn).await?;
453
454        let mut dep_secrets: HashSet<SecretId> = HashSet::new();
455        if let Some(ConnectionInfo::ConnectionParams(params)) = &pb_connection.info {
456            dep_secrets.extend(
457                params
458                    .secret_refs
459                    .values()
460                    .map(|secret_ref| secret_ref.secret_id),
461            );
462        }
463
464        let conn_obj = Self::create_object(
465            &txn,
466            ObjectType::Connection,
467            owner_id,
468            Some(pb_connection.schema_id.as_object_id()),
469        )
470        .await?;
471        pb_connection.id = conn_obj.oid.as_connection_id();
472        let connection: connection::ActiveModel = pb_connection.clone().into();
473        Connection::insert(connection).exec(&txn).await?;
474
475        for secret_id in &dep_secrets {
476            ObjectDependency::insert(object_dependency::ActiveModel {
477                oid: Set(secret_id.as_object_id()),
478                used_by: Set(conn_obj.oid),
479                ..Default::default()
480            })
481            .exec(&txn)
482            .await?;
483        }
484
485        let updated_user_info = grant_default_privileges_automatically(&txn, conn_obj.oid).await?;
486
487        txn.commit().await?;
488
489        {
490            // call meta telemetry here to report the connection creation
491            report_event(
492                PbTelemetryEventStage::Unspecified,
493                "connection_create",
494                pb_connection.get_id().as_raw_id() as i64,
495                {
496                    pb_connection.info.as_ref().and_then(|info| match info {
497                        ConnectionInfo::ConnectionParams(params) => {
498                            Some(params.connection_type().as_str_name().to_owned())
499                        }
500                        _ => None,
501                    })
502                },
503                None,
504                None,
505            );
506        }
507
508        let mut notifications = self.frontend_notification_batch();
509        notifications.add(
510            NotificationOperation::Add,
511            NotificationInfo::ObjectGroup(PbObjectGroup {
512                objects: vec![PbObject {
513                    object_info: Some(PbObjectInfo::Connection(pb_connection)),
514                }],
515                dependencies: dep_secrets
516                    .iter()
517                    .map(|secret_id| PbObjectDependency {
518                        object_id: conn_obj.oid,
519                        referenced_object_id: secret_id.as_object_id(),
520                        referenced_object_type: PbObjectType::Secret as _,
521                    })
522                    .collect(),
523            }),
524        );
525
526        // notify default privileges for connections
527        Self::add_users_update_notifications(&mut notifications, updated_user_info);
528        let version = notifications
529            .publish()
530            .await
531            .expect("the connection notification batch contains the connection");
532
533        Ok(version)
534    }
535
536    pub async fn create_secret(
537        &self,
538        mut pb_secret: PbSecret,
539        secret_plain_payload: Vec<u8>,
540    ) -> MetaResult<NotificationVersion> {
541        let inner = self.inner.write().await;
542        let owner_id = pb_secret.owner as _;
543        let txn = inner.db.begin().await?;
544        ensure_user_id(owner_id, &txn).await?;
545        ensure_object_id(ObjectType::Database, pb_secret.database_id, &txn).await?;
546        ensure_object_id(ObjectType::Schema, pb_secret.schema_id, &txn).await?;
547        check_secret_name_duplicate(&pb_secret, &txn).await?;
548
549        let secret_obj = Self::create_object(
550            &txn,
551            ObjectType::Secret,
552            owner_id,
553            Some(pb_secret.schema_id.as_object_id()),
554        )
555        .await?;
556        pb_secret.id = secret_obj.oid.as_secret_id();
557        let secret: secret::ActiveModel = pb_secret.clone().into();
558        Secret::insert(secret).exec(&txn).await?;
559
560        let updated_user_info =
561            grant_default_privileges_automatically(&txn, secret_obj.oid).await?;
562
563        txn.commit().await?;
564
565        // Notify the compute and frontend node plain secret
566        let mut secret_plain = pb_secret;
567        secret_plain.value.clone_from(&secret_plain_payload);
568
569        LocalSecretManager::global().add_secret(secret_plain.id, secret_plain_payload);
570        self.env
571            .notification_manager()
572            .notify_compute_without_version(Operation::Add, Info::Secret(secret_plain.clone()));
573
574        let mut notifications = self.frontend_notification_batch();
575        notifications.add(
576            NotificationOperation::Add,
577            NotificationInfo::Secret(secret_plain),
578        );
579
580        // notify default privileges for secrets
581        Self::add_users_update_notifications(&mut notifications, updated_user_info);
582        let version = notifications
583            .publish()
584            .await
585            .expect("the secret notification batch contains the secret");
586
587        Ok(version)
588    }
589
590    pub async fn create_view(
591        &self,
592        mut pb_view: PbView,
593        dependencies: HashSet<ObjectId>,
594    ) -> MetaResult<NotificationVersion> {
595        let inner = self.inner.write().await;
596        let owner_id = pb_view.owner as _;
597        let txn = inner.db.begin().await?;
598        ensure_user_id(owner_id, &txn).await?;
599        ensure_object_id(ObjectType::Database, pb_view.database_id, &txn).await?;
600        ensure_object_id(ObjectType::Schema, pb_view.schema_id, &txn).await?;
601        check_relation_name_duplicate(&pb_view.name, pb_view.database_id, pb_view.schema_id, &txn)
602            .await?;
603        ensure_object_id(ObjectType::Schema, pb_view.schema_id, &txn).await?;
604        check_relation_name_duplicate(&pb_view.name, pb_view.database_id, pb_view.schema_id, &txn)
605            .await?;
606
607        let view_obj = Self::create_object(
608            &txn,
609            ObjectType::View,
610            owner_id,
611            Some(pb_view.schema_id.as_object_id()),
612        )
613        .await?;
614        pb_view.id = view_obj.oid.as_view_id();
615        pb_view.created_at_epoch =
616            Some(Epoch::from_unix_millis(datetime_to_timestamp_millis(view_obj.created_at) as _).0);
617        pb_view.created_at_cluster_version = view_obj.created_at_cluster_version;
618
619        let view: view::ActiveModel = pb_view.clone().into();
620        View::insert(view).exec(&txn).await?;
621
622        let dependencies = if !dependencies.is_empty() {
623            ObjectDependency::insert_many(dependencies.into_iter().map(|obj_id| {
624                object_dependency::ActiveModel {
625                    oid: Set(obj_id),
626                    used_by: Set(view_obj.oid),
627                    ..Default::default()
628                }
629            }))
630            .exec(&txn)
631            .await?;
632            list_object_dependencies_by_object_id(&txn, view_obj.oid).await?
633        } else {
634            vec![]
635        };
636
637        let updated_user_info = grant_default_privileges_automatically(&txn, view_obj.oid).await?;
638
639        txn.commit().await?;
640        let mut notifications = self.frontend_notification_batch();
641        notifications.add(
642            NotificationOperation::Add,
643            NotificationInfo::ObjectGroup(PbObjectGroup {
644                objects: vec![PbObject {
645                    object_info: Some(PbObjectInfo::View(pb_view)),
646                }],
647                dependencies,
648            }),
649        );
650
651        // notify default privileges for views
652        Self::add_users_update_notifications(&mut notifications, updated_user_info);
653        let version = notifications
654            .publish()
655            .await
656            .expect("the view notification batch contains the view");
657
658        Ok(version)
659    }
660
661    pub async fn validate_cross_db_snapshot_backfill(
662        &self,
663        cross_db_snapshot_backfill_info: &SnapshotBackfillInfo,
664    ) -> MetaResult<()> {
665        if cross_db_snapshot_backfill_info
666            .upstream_mv_table_id_to_backfill_epoch
667            .is_empty()
668        {
669            return Ok(());
670        }
671
672        let inner = self.inner.read().await;
673        let table_ids = cross_db_snapshot_backfill_info
674            .upstream_mv_table_id_to_backfill_epoch
675            .keys()
676            .copied()
677            .map_into()
678            .collect_vec();
679        let cnt = Subscription::find()
680            .select_only()
681            .column(subscription::Column::DependentTableId)
682            .distinct()
683            .filter(subscription::Column::DependentTableId.is_in::<TableId, _>(table_ids))
684            .count(&inner.db)
685            .await? as usize;
686
687        if cnt
688            < cross_db_snapshot_backfill_info
689                .upstream_mv_table_id_to_backfill_epoch
690                .keys()
691                .count()
692        {
693            return Err(MetaError::permission_denied(
694                "Some upstream tables are not subscribed".to_owned(),
695            ));
696        }
697
698        Ok(())
699    }
700}
risingwave_meta/controller/catalog/create_op.rs

risingwave_meta/controller/catalog/
create_op.rs
