risingwave_meta/controller/catalog/
test.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
15#[cfg(test)]
16mod tests {
17    use risingwave_common::hash::VirtualNode;
18    use risingwave_meta_model::FragmentId;
19    use risingwave_meta_model::fragment::DistributionType;
20    use risingwave_meta_model::table::HandleConflictBehavior;
21    use risingwave_pb::catalog::subscription::SubscriptionState;
22    use risingwave_pb::catalog::{PbSinkType, StreamSourceInfo};
23    use risingwave_pb::common::{HostAddress, WorkerNode, WorkerType, worker_node};
24    use risingwave_pb::meta::SubscribeType;
25    use risingwave_pb::stream_plan::PbStreamNode;
26    use tokio::sync::{mpsc, oneshot};
27
28    use crate::controller::catalog::*;
29    use crate::manager::{LocalNotification, WorkerKey};
30    use crate::model::FragmentDownstreamRelation;
31    use crate::serving::ServingVnodeMapping;
32
33    const TEST_DATABASE_ID: DatabaseId = DatabaseId::new(1);
34    const TEST_SCHEMA_ID: SchemaId = SchemaId::new(2);
35    const TEST_OWNER_ID: UserId = UserId::new(1);
36
37    async fn insert_test_table(
38        txn: &DatabaseTransaction,
39        table_id: TableId,
40        name: &str,
41        table_type: TableType,
42        belongs_to_job_id: Option<JobId>,
43        definition: &str,
44    ) -> MetaResult<()> {
45        table::ActiveModel {
46            table_id: Set(table_id),
47            name: Set(name.to_owned()),
48            optional_associated_source_id: Set(None),
49            table_type: Set(table_type),
50            belongs_to_job_id: Set(belongs_to_job_id),
51            columns: Set(vec![].into()),
52            pk: Set(vec![].into()),
53            distribution_key: Set(Vec::<i32>::new().into()),
54            stream_key: Set(Vec::<i32>::new().into()),
55            append_only: Set(false),
56            fragment_id: Set(None),
57            vnode_col_index: Set(None),
58            row_id_index: Set(None),
59            value_indices: Set(Vec::<i32>::new().into()),
60            definition: Set(definition.to_owned()),
61            handle_pk_conflict_behavior: Set(HandleConflictBehavior::NoCheck),
62            version_column_indices: Set(None),
63            read_prefix_len_hint: Set(0),
64            watermark_indices: Set(Vec::<i32>::new().into()),
65            dist_key_in_pk: Set(Vec::<i32>::new().into()),
66            dml_fragment_id: Set(None),
67            cardinality: Set(None),
68            cleaned_by_watermark: Set(false),
69            description: Set(None),
70            version: Set(None),
71            retention_seconds: Set(None),
72            cdc_table_id: Set(None),
73            vnode_count: Set(1),
74            webhook_info: Set(None),
75            engine: Set(None),
76            clean_watermark_index_in_pk: Set(None),
77            clean_watermark_indices: Set(None),
78            refreshable: Set(false),
79            vector_index_info: Set(None),
80            cdc_table_type: Set(None),
81        }
82        .insert(txn)
83        .await?;
84        Ok(())
85    }
86
87    async fn insert_test_fragment(
88        txn: &DatabaseTransaction,
89        fragment_id: FragmentId,
90        job_id: JobId,
91        state_table_ids: TableIdArray,
92    ) -> MetaResult<()> {
93        fragment::ActiveModel {
94            fragment_id: Set(fragment_id),
95            job_id: Set(job_id),
96            fragment_type_mask: Set(0),
97            distribution_type: Set(fragment::DistributionType::Hash),
98            stream_node: Set(StreamNode::from(&PbStreamNode::default())),
99            state_table_ids: Set(state_table_ids),
100            upstream_fragment_id: Set(I32Array::default()),
101            vnode_count: Set(1),
102            parallelism: Set(None),
103        }
104        .insert(txn)
105        .await?;
106        Ok(())
107    }
108
109    async fn insert_test_streaming_job(
110        txn: &DatabaseTransaction,
111        name: &str,
112        has_result_table: bool,
113        policy: Option<CacheRefillPolicy>,
114    ) -> MetaResult<(JobId, Option<TableId>, TableId)> {
115        let object_type = if has_result_table {
116            ObjectType::Table
117        } else {
118            ObjectType::Sink
119        };
120        let job_id = CatalogController::create_object(
121            txn,
122            object_type,
123            TEST_OWNER_ID,
124            Some(TEST_DATABASE_ID),
125            Some(TEST_SCHEMA_ID),
126        )
127        .await?
128        .oid
129        .as_job_id();
130        let result_table_id = has_result_table.then_some(job_id.as_mv_table_id());
131        if let Some(table_id) = result_table_id {
132            insert_test_table(txn, table_id, name, TableType::MaterializedView, None, "").await?;
133        }
134
135        let internal_table_id = CatalogController::create_object(
136            txn,
137            ObjectType::Table,
138            TEST_OWNER_ID,
139            Some(TEST_DATABASE_ID),
140            Some(TEST_SCHEMA_ID),
141        )
142        .await?
143        .oid
144        .as_table_id();
145        insert_test_table(
146            txn,
147            internal_table_id,
148            &format!("__internal_{name}"),
149            TableType::Internal,
150            Some(job_id),
151            "",
152        )
153        .await?;
154
155        insert_test_streaming_job_model(txn, job_id, policy).await?;
156
157        Ok((job_id, result_table_id, internal_table_id))
158    }
159
160    async fn insert_test_streaming_job_model(
161        txn: &DatabaseTransaction,
162        job_id: JobId,
163        policy: Option<CacheRefillPolicy>,
164    ) -> MetaResult<()> {
165        streaming_job::ActiveModel {
166            job_id: Set(job_id),
167            job_status: Set(JobStatus::Created),
168            create_type: Set(CreateType::Foreground),
169            timezone: Set(None),
170            config_override: Set(policy.map(|policy| {
171                format!(
172                    "[streaming.developer]\ncache_refill_policy = \"{}\"\n",
173                    policy
174                )
175            })),
176            adaptive_parallelism_strategy: Set(None),
177            parallelism: Set(StreamingParallelism::Adaptive),
178            backfill_parallelism: Set(None),
179            backfill_adaptive_parallelism_strategy: Set(None),
180            backfill_orders: Set(None),
181            max_parallelism: Set(1),
182            specific_resource_group: Set(None),
183            is_serverless_backfill: Set(false),
184            refresh_interval_sec: Set(None),
185        }
186        .insert(txn)
187        .await?;
188
189        Ok(())
190    }
191
192    #[tokio::test]
193    async fn test_table_refill_catalog_snapshot_classifies_table_identity() -> MetaResult<()> {
194        let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
195        let inner = mgr.inner.write().await;
196        let txn = inner.db.begin().await?;
197
198        let (mv_job, Some(mv_result), mv_internal) =
199            insert_test_streaming_job(&txn, "mv_both", true, Some(CacheRefillPolicy::Both)).await?
200        else {
201            unreachable!()
202        };
203        let (default_job, Some(_default_result), default_internal) =
204            insert_test_streaming_job(&txn, "mv_default", true, None).await?
205        else {
206            unreachable!()
207        };
208        let (sink_job, None, sink_internal) = insert_test_streaming_job(
209            &txn,
210            "sink_streaming",
211            false,
212            Some(CacheRefillPolicy::Streaming),
213        )
214        .await?
215        else {
216            unreachable!()
217        };
218
219        let result_fragment = FragmentId::new(100);
220        let internal_fragment = FragmentId::new(101);
221        let sink_fragment = FragmentId::new(102);
222        for (fragment_id, job_id, table_ids) in [
223            (result_fragment, mv_job, vec![mv_result, mv_internal]),
224            (internal_fragment, default_job, vec![default_internal]),
225            (sink_fragment, sink_job, vec![sink_internal]),
226        ] {
227            insert_test_fragment(&txn, fragment_id, job_id, TableIdArray(table_ids)).await?;
228        }
229        txn.commit().await?;
230        drop(inner);
231
232        let serving_infos = mgr.fragment_serving_infos().await?;
233        assert_eq!(serving_infos.len(), 3);
234        assert_eq!(
235            serving_infos[&result_fragment].result_table_id,
236            Some(mv_result)
237        );
238        assert_eq!(serving_infos[&internal_fragment].result_table_id, None);
239        assert_eq!(serving_infos[&sink_fragment].result_table_id, None);
240
241        let policies = mgr.table_cache_refill_policies_snapshot().await?;
242        assert_eq!(
243            policies
244                .table_policies
245                .into_iter()
246                .map(|policy| (policy.table_id, policy.policy))
247                .collect::<HashMap<_, _>>(),
248            HashMap::from([(
249                mv_result.as_raw_id(),
250                CacheRefillPolicy::Both.to_protobuf() as i32,
251            )])
252        );
253        assert_eq!(
254            policies
255                .internal_table_policies
256                .into_iter()
257                .map(|policy| (policy.table_id, policy.policy))
258                .collect::<HashMap<_, _>>(),
259            HashMap::from([
260                (
261                    mv_internal.as_raw_id(),
262                    CacheRefillPolicy::Both.to_protobuf() as i32,
263                ),
264                (
265                    sink_internal.as_raw_id(),
266                    CacheRefillPolicy::Streaming.to_protobuf() as i32,
267                ),
268            ])
269        );
270
271        Ok(())
272    }
273
274    #[tokio::test]
275    async fn test_foreground_creating_catalog_lifecycle() -> MetaResult<()> {
276        let env = MetaSrvEnv::for_test().await;
277        let (tx, mut notification_rx) = mpsc::unbounded_channel();
278        env.notification_manager().insert_sender(
279            SubscribeType::Frontend,
280            WorkerKey(HostAddress {
281                host: "localhost".to_owned(),
282                port: 1234,
283            }),
284            tx,
285        );
286        let mgr = CatalogController::new(env).await?;
287        let inner = mgr.inner.write().await;
288        let txn = inner.db.begin().await?;
289
290        // A foreground table and its internal table are both visible while creating.
291        let (job_id, Some(table_id), internal_table_id) =
292            insert_test_streaming_job(&txn, "creating_table", true, None).await?
293        else {
294            unreachable!()
295        };
296        let associated_source_id = CatalogController::create_object(
297            &txn,
298            ObjectType::Source,
299            TEST_OWNER_ID,
300            Some(TEST_DATABASE_ID),
301            Some(TEST_SCHEMA_ID),
302        )
303        .await?
304        .oid
305        .as_source_id();
306        Source::insert(source::ActiveModel::from(PbSource {
307            id: associated_source_id,
308            schema_id: TEST_SCHEMA_ID,
309            database_id: TEST_DATABASE_ID,
310            name: "creating_table_source".to_owned(),
311            owner: TEST_OWNER_ID as _,
312            optional_associated_table_id: Some(
313                risingwave_pb::catalog::source::OptionalAssociatedTableId::AssociatedTableId(
314                    table_id,
315                ),
316            ),
317            ..Default::default()
318        }))
319        .exec(&txn)
320        .await?;
321        table::ActiveModel {
322            table_id: Set(table_id),
323            table_type: Set(TableType::Table),
324            optional_associated_source_id: Set(Some(associated_source_id)),
325            ..Default::default()
326        }
327        .update(&txn)
328        .await?;
329        streaming_job::ActiveModel {
330            job_id: Set(job_id),
331            job_status: Set(JobStatus::Initial),
332            ..Default::default()
333        }
334        .update(&txn)
335        .await?;
336
337        // A foreground index and its index table are both visible while creating.
338        let (_primary_job_id, Some(primary_table_id), _) =
339            insert_test_streaming_job(&txn, "primary_table", true, None).await?
340        else {
341            unreachable!()
342        };
343        let index_job_id = CatalogController::create_object(
344            &txn,
345            ObjectType::Index,
346            TEST_OWNER_ID,
347            Some(TEST_DATABASE_ID),
348            Some(TEST_SCHEMA_ID),
349        )
350        .await?
351        .oid
352        .as_job_id();
353        let index_table_id = index_job_id.as_mv_table_id();
354        insert_test_table(
355            &txn,
356            index_table_id,
357            "creating_index",
358            TableType::Index,
359            None,
360            "",
361        )
362        .await?;
363        index::ActiveModel {
364            index_id: Set(index_job_id.as_index_id()),
365            name: Set("creating_index".to_owned()),
366            index_table_id: Set(index_table_id),
367            primary_table_id: Set(primary_table_id),
368            index_items: Set(vec![].into()),
369            index_column_properties: Set(None),
370            index_columns_len: Set(0),
371        }
372        .insert(&txn)
373        .await?;
374        insert_test_streaming_job_model(&txn, index_job_id, None).await?;
375        streaming_job::ActiveModel {
376            job_id: Set(index_job_id),
377            job_status: Set(JobStatus::Initial),
378            ..Default::default()
379        }
380        .update(&txn)
381        .await?;
382
383        // A creating shared source is included in restart snapshots as well.
384        let source_job_id = CatalogController::create_object(
385            &txn,
386            ObjectType::Source,
387            TEST_OWNER_ID,
388            Some(TEST_DATABASE_ID),
389            Some(TEST_SCHEMA_ID),
390        )
391        .await?
392        .oid
393        .as_job_id();
394        Source::insert(source::ActiveModel::from(PbSource {
395            id: source_job_id.as_shared_source_id(),
396            schema_id: TEST_SCHEMA_ID,
397            database_id: TEST_DATABASE_ID,
398            name: "creating_shared_source".to_owned(),
399            owner: TEST_OWNER_ID as _,
400            info: Some(StreamSourceInfo {
401                cdc_source_job: true,
402                ..Default::default()
403            }),
404            ..Default::default()
405        }))
406        .exec(&txn)
407        .await?;
408        insert_test_streaming_job_model(&txn, source_job_id, None).await?;
409        streaming_job::ActiveModel {
410            job_id: Set(source_job_id),
411            job_status: Set(JobStatus::Initial),
412            ..Default::default()
413        }
414        .update(&txn)
415        .await?;
416
417        let sink_job_id = CatalogController::create_object(
418            &txn,
419            ObjectType::Sink,
420            TEST_OWNER_ID,
421            Some(TEST_DATABASE_ID),
422            Some(TEST_SCHEMA_ID),
423        )
424        .await?
425        .oid
426        .as_job_id();
427        Sink::insert(sink::ActiveModel::from(PbSink {
428            id: sink_job_id.as_sink_id(),
429            schema_id: TEST_SCHEMA_ID,
430            database_id: TEST_DATABASE_ID,
431            name: "creating_sink".to_owned(),
432            owner: TEST_OWNER_ID as _,
433            sink_type: PbSinkType::AppendOnly as i32,
434            ..Default::default()
435        }))
436        .exec(&txn)
437        .await?;
438        insert_test_streaming_job_model(&txn, sink_job_id, None).await?;
439        streaming_job::ActiveModel {
440            job_id: Set(sink_job_id),
441            job_status: Set(JobStatus::Initial),
442            ..Default::default()
443        }
444        .update(&txn)
445        .await?;
446
447        txn.commit().await?;
448
449        let (catalog, _) = inner.snapshot().await?;
450        assert!(!catalog.2.iter().any(|table| table.id == table_id));
451        assert!(!catalog.2.iter().any(|table| table.id == internal_table_id));
452        assert!(!catalog.2.iter().any(|table| table.id == index_table_id));
453        assert!(
454            !catalog
455                .3
456                .iter()
457                .any(|source| source.id == associated_source_id)
458        );
459        assert!(
460            !catalog
461                .3
462                .iter()
463                .any(|source| source.id == source_job_id.as_shared_source_id())
464        );
465        assert!(
466            !catalog
467                .6
468                .iter()
469                .any(|index| index.id == index_job_id.as_index_id())
470        );
471        assert!(
472            !catalog
473                .4
474                .iter()
475                .any(|sink| sink.id == sink_job_id.as_sink_id())
476        );
477
478        drop(inner);
479
480        let downstreams = FragmentDownstreamRelation::new();
481        let mut add_notifications = vec![];
482        for creating_job_id in [job_id, index_job_id, source_job_id, sink_job_id] {
483            mgr.post_collect_job_fragments(creating_job_id, &downstreams, None, None, None, true)
484                .await?;
485            let notification = notification_rx
486                .recv()
487                .await
488                .expect("frontend should receive a creating notification")
489                .expect("creating notification should be valid");
490            assert_eq!(notification.operation(), NotificationOperation::Add);
491            let object_group = match notification.info {
492                Some(NotificationInfo::ObjectGroup(object_group)) => object_group,
493                other => panic!("unexpected notification: {other:?}"),
494            };
495            add_notifications.push(object_group);
496        }
497
498        assert!(add_notifications[0].objects.iter().any(|object| matches!(
499            &object.object_info,
500            Some(PbObjectInfo::Table(table)) if table.id == table_id
501        )));
502        assert!(add_notifications[0].objects.iter().any(|object| matches!(
503            &object.object_info,
504            Some(PbObjectInfo::Table(table)) if table.id == internal_table_id
505        )));
506        assert!(add_notifications[0].objects.iter().any(|object| matches!(
507            &object.object_info,
508            Some(PbObjectInfo::Source(source)) if source.id == associated_source_id
509        )));
510        assert!(add_notifications[1].objects.iter().any(|object| matches!(
511            &object.object_info,
512            Some(PbObjectInfo::Table(table)) if table.id == index_table_id
513        )));
514        assert!(add_notifications[1].objects.iter().any(|object| matches!(
515            &object.object_info,
516            Some(PbObjectInfo::Index(index)) if index.id == index_job_id.as_index_id()
517        )));
518        assert!(add_notifications[2].objects.iter().any(|object| matches!(
519            &object.object_info,
520            Some(PbObjectInfo::Source(source)) if source.id == source_job_id.as_shared_source_id()
521        )));
522        assert!(add_notifications[3].objects.iter().any(|object| matches!(
523            &object.object_info,
524            Some(PbObjectInfo::Sink(sink)) if sink.id == sink_job_id.as_sink_id()
525        )));
526
527        let inner = mgr.inner.write().await;
528        let (catalog, _) = inner.snapshot().await?;
529        assert!(catalog.2.iter().any(|table| table.id == table_id));
530        assert!(catalog.2.iter().any(|table| table.id == internal_table_id));
531        assert!(catalog.2.iter().any(|table| table.id == index_table_id));
532        assert!(
533            catalog
534                .3
535                .iter()
536                .any(|source| source.id == source_job_id.as_shared_source_id())
537        );
538        assert!(
539            catalog
540                .6
541                .iter()
542                .any(|index| index.id == index_job_id.as_index_id())
543        );
544        assert!(
545            catalog
546                .4
547                .iter()
548                .any(|sink| sink.id == sink_job_id.as_sink_id())
549        );
550
551        let txn = inner.db.begin().await?;
552        let (operation, _, _, _) = mgr.finish_streaming_job_inner(&txn, job_id).await?;
553        assert_eq!(operation, NotificationOperation::Update);
554        let (operation, _, _, _) = mgr.finish_streaming_job_inner(&txn, index_job_id).await?;
555        assert_eq!(operation, NotificationOperation::Update);
556        let (operation, _, _, _) = mgr.finish_streaming_job_inner(&txn, source_job_id).await?;
557        assert_eq!(operation, NotificationOperation::Update);
558        let (operation, _, _, _) = mgr.finish_streaming_job_inner(&txn, sink_job_id).await?;
559        assert_eq!(operation, NotificationOperation::Update);
560        txn.commit().await?;
561
562        Ok(())
563    }
564
565    #[tokio::test]
566    async fn test_alter_streaming_job_cache_refill_policy_notifies_hummock() -> MetaResult<()> {
567        let env = MetaSrvEnv::for_test().await;
568        let (tx, mut rx) = mpsc::unbounded_channel();
569        env.notification_manager().insert_sender(
570            SubscribeType::Hummock,
571            WorkerKey(HostAddress {
572                host: "localhost".to_owned(),
573                port: 1234,
574            }),
575            tx,
576        );
577        let mgr = CatalogController::new(env).await?;
578
579        let inner = mgr.inner.write().await;
580        let txn = inner.db.begin().await?;
581        let (_job, Some(result_table_id), internal_table_id) =
582            insert_test_streaming_job(&txn, "mv_cache_refill", true, None).await?
583        else {
584            unreachable!()
585        };
586        txn.commit().await?;
587        drop(inner);
588
589        mgr.alter_streaming_job_config(
590            result_table_id.as_job_id(),
591            HashMap::from([(
592                STREAMING_CACHE_REFILL_POLICY_CONFIG_PATH.to_owned(),
593                "\"both\"".to_owned(),
594            )]),
595            vec![],
596        )
597        .await?;
598
599        let response = rx
600            .recv()
601            .await
602            .expect("should receive hummock notification")
603            .expect("notification should be ok");
604        assert_eq!(response.operation(), NotificationOperation::Update);
605        let info = response.info;
606        let Some(NotificationInfo::TableRefillRuntimeConfig(config)) = info else {
607            panic!("unexpected notification: {:?}", info);
608        };
609        assert!(config.serving_table_vnode_mappings.is_none());
610        let policies = config
611            .table_cache_refill_policies
612            .expect("policy snapshot should be present");
613        assert_eq!(
614            policies
615                .table_policies
616                .into_iter()
617                .map(|policy| (policy.table_id, policy.policy))
618                .collect::<HashMap<_, _>>(),
619            HashMap::from([(
620                result_table_id.as_raw_id(),
621                CacheRefillPolicy::Both.to_protobuf() as i32,
622            )])
623        );
624        assert_eq!(
625            policies
626                .internal_table_policies
627                .into_iter()
628                .map(|policy| (policy.table_id, policy.policy))
629                .collect::<HashMap<_, _>>(),
630            HashMap::from([(
631                internal_table_id.as_raw_id(),
632                CacheRefillPolicy::Both.to_protobuf() as i32,
633            )])
634        );
635
636        Ok(())
637    }
638
639    async fn insert_dirty_creating_job_with_fragment(
640        mgr: &CatalogController,
641        fragment_id: FragmentId,
642        vnode_count: i32,
643    ) -> MetaResult<(JobId, TableId)> {
644        let inner = mgr.inner.write().await;
645        let txn = inner.db.begin().await?;
646        let job_obj = CatalogController::create_object(
647            &txn,
648            ObjectType::Table,
649            TEST_OWNER_ID,
650            Some(TEST_DATABASE_ID),
651            Some(TEST_SCHEMA_ID),
652        )
653        .await?;
654        let job_id = job_obj.oid.as_job_id();
655        let table_id = job_id.as_mv_table_id();
656        insert_test_table(
657            &txn,
658            table_id,
659            "mv_dirty_serving_mapping",
660            TableType::MaterializedView,
661            None,
662            "CREATE MATERIALIZED VIEW mv_dirty_serving_mapping AS SELECT 1",
663        )
664        .await?;
665        streaming_job::ActiveModel {
666            job_id: Set(job_id),
667            job_status: Set(JobStatus::Creating),
668            create_type: Set(CreateType::Foreground),
669            timezone: Set(None),
670            config_override: Set(None),
671            adaptive_parallelism_strategy: Set(None),
672            parallelism: Set(StreamingParallelism::Adaptive),
673            backfill_parallelism: Set(None),
674            backfill_adaptive_parallelism_strategy: Set(None),
675            backfill_orders: Set(None),
676            max_parallelism: Set(1),
677            specific_resource_group: Set(None),
678            is_serverless_backfill: Set(false),
679            refresh_interval_sec: Set(None),
680        }
681        .insert(&txn)
682        .await?;
683        fragment::ActiveModel {
684            fragment_id: Set(fragment_id),
685            job_id: Set(job_id),
686            fragment_type_mask: Set(0),
687            distribution_type: Set(DistributionType::Hash),
688            stream_node: Set(StreamNode::default()),
689            state_table_ids: Set(Vec::<TableId>::new().into()),
690            upstream_fragment_id: Set(Vec::<i32>::new().into()),
691            vnode_count: Set(vnode_count),
692            parallelism: Set(None),
693        }
694        .insert(&txn)
695        .await?;
696        txn.commit().await?;
697        drop(inner);
698
699        Ok((job_id, table_id))
700    }
701
702    #[tokio::test]
703    async fn test_dirty_cleanup_reconcile_removes_stale_serving_vnode_mapping() -> MetaResult<()> {
704        let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
705        let fragment_id = FragmentId::new(42);
706        insert_dirty_creating_job_with_fragment(
707            &mgr,
708            fragment_id,
709            VirtualNode::COUNT_FOR_TEST as i32,
710        )
711        .await?;
712
713        let worker = WorkerNode {
714            id: WorkerId::new(1),
715            r#type: WorkerType::ComputeNode.into(),
716            host: Some(HostAddress {
717                host: "localhost".to_owned(),
718                port: 1,
719            }),
720            state: worker_node::State::Running as i32,
721            property: Some(worker_node::Property {
722                is_serving: true,
723                parallelism: 1,
724                ..Default::default()
725            }),
726            ..Default::default()
727        };
728        let serving_vnode_mapping = ServingVnodeMapping::default();
729        let initial_snapshot = mgr.fragment_serving_infos().await?;
730        serving_vnode_mapping.upsert(&initial_snapshot, std::slice::from_ref(&worker), None);
731        assert!(serving_vnode_mapping.all().contains_key(&fragment_id));
732
733        mgr.clean_dirty_creating_jobs(Some(TEST_DATABASE_ID))
734            .await?;
735        let current_snapshot = mgr.fragment_serving_infos().await?;
736        assert!(!current_snapshot.contains_key(&fragment_id));
737
738        serving_vnode_mapping.reconcile(&current_snapshot, &[worker], None);
739        assert!(!serving_vnode_mapping.all().contains_key(&fragment_id));
740
741        Ok(())
742    }
743
744    #[tokio::test]
745    async fn test_database_func() -> MetaResult<()> {
746        let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
747        let pb_database = PbDatabase {
748            name: "db1".to_owned(),
749            owner: TEST_OWNER_ID as _,
750            ..Default::default()
751        };
752        mgr.create_database(pb_database).await?;
753
754        let database_id: DatabaseId = Database::find()
755            .select_only()
756            .column(database::Column::DatabaseId)
757            .filter(database::Column::Name.eq("db1"))
758            .into_tuple()
759            .one(&mgr.inner.read().await.db)
760            .await?
761            .unwrap();
762
763        mgr.alter_name(ObjectType::Database, database_id, "db2")
764            .await?;
765        let database = Database::find_by_id(database_id)
766            .one(&mgr.inner.read().await.db)
767            .await?
768            .unwrap();
769        assert_eq!(database.name, "db2");
770
771        mgr.drop_object(ObjectType::Database, database_id, DropMode::Cascade)
772            .await?;
773
774        Ok(())
775    }
776
777    #[tokio::test]
778    async fn test_schema_func() -> MetaResult<()> {
779        let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
780        let pb_schema = PbSchema {
781            database_id: TEST_DATABASE_ID,
782            name: "schema1".to_owned(),
783            owner: TEST_OWNER_ID as _,
784            ..Default::default()
785        };
786        mgr.create_schema(pb_schema.clone()).await?;
787        assert!(mgr.create_schema(pb_schema).await.is_err());
788
789        let schema_id: SchemaId = Schema::find()
790            .select_only()
791            .column(schema::Column::SchemaId)
792            .filter(schema::Column::Name.eq("schema1"))
793            .into_tuple()
794            .one(&mgr.inner.read().await.db)
795            .await?
796            .unwrap();
797
798        mgr.alter_name(ObjectType::Schema, schema_id, "schema2")
799            .await?;
800        let schema = Schema::find_by_id(schema_id)
801            .one(&mgr.inner.read().await.db)
802            .await?
803            .unwrap();
804        assert_eq!(schema.name, "schema2");
805        mgr.drop_object(ObjectType::Schema, schema_id, DropMode::Restrict)
806            .await?;
807
808        Ok(())
809    }
810
811    #[tokio::test]
812    async fn test_create_view() -> MetaResult<()> {
813        let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
814        let pb_view = PbView {
815            schema_id: TEST_SCHEMA_ID,
816            database_id: TEST_DATABASE_ID,
817            name: "view".to_owned(),
818            owner: TEST_OWNER_ID as _,
819            sql: "CREATE VIEW view AS SELECT 1".to_owned(),
820            ..Default::default()
821        };
822        mgr.create_view(pb_view.clone(), HashSet::new()).await?;
823        assert!(mgr.create_view(pb_view, HashSet::new()).await.is_err());
824
825        let view = View::find().one(&mgr.inner.read().await.db).await?.unwrap();
826        mgr.drop_object(ObjectType::View, view.view_id, DropMode::Cascade)
827            .await?;
828        assert!(
829            View::find_by_id(view.view_id)
830                .one(&mgr.inner.read().await.db)
831                .await?
832                .is_none()
833        );
834
835        Ok(())
836    }
837
838    #[tokio::test]
839    async fn test_create_function() -> MetaResult<()> {
840        let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
841        let test_data_type = risingwave_pb::data::DataType {
842            type_name: risingwave_pb::data::data_type::TypeName::Int32 as _,
843            ..Default::default()
844        };
845        let arg_types = vec![test_data_type.clone()];
846        let pb_function = PbFunction {
847            schema_id: TEST_SCHEMA_ID,
848            database_id: TEST_DATABASE_ID,
849            name: "test_function".to_owned(),
850            owner: TEST_OWNER_ID as _,
851            arg_types,
852            return_type: Some(test_data_type.clone()),
853            language: "python".to_owned(),
854            kind: Some(risingwave_pb::catalog::function::Kind::Scalar(
855                Default::default(),
856            )),
857            ..Default::default()
858        };
859        mgr.create_function(pb_function.clone()).await?;
860        assert!(mgr.create_function(pb_function).await.is_err());
861
862        let function = Function::find()
863            .inner_join(Object)
864            .filter(
865                object::Column::DatabaseId
866                    .eq(TEST_DATABASE_ID)
867                    .and(object::Column::SchemaId.eq(TEST_SCHEMA_ID))
868                    .add(function::Column::Name.eq("test_function")),
869            )
870            .one(&mgr.inner.read().await.db)
871            .await?
872            .unwrap();
873        assert_eq!(function.return_type.to_protobuf(), test_data_type);
874        assert_eq!(function.arg_types.to_protobuf().len(), 1);
875        assert_eq!(function.language, "python");
876
877        mgr.drop_object(
878            ObjectType::Function,
879            function.function_id,
880            DropMode::Restrict,
881        )
882        .await?;
883        assert!(
884            Object::find_by_id(function.function_id)
885                .one(&mgr.inner.read().await.db)
886                .await?
887                .is_none()
888        );
889
890        Ok(())
891    }
892
893    #[tokio::test]
894    async fn test_alter_relation_rename() -> MetaResult<()> {
895        let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
896        let pb_source = PbSource {
897            schema_id: TEST_SCHEMA_ID,
898            database_id: TEST_DATABASE_ID,
899            name: "s1".to_owned(),
900            owner: TEST_OWNER_ID as _,
901            definition: r#"CREATE SOURCE s1 (v1 int) with (
902  connector = 'kafka',
903  topic = 'kafka_alter',
904  properties.bootstrap.server = 'message_queue:29092',
905  scan.startup.mode = 'earliest'
906) FORMAT PLAIN ENCODE JSON"#
907                .to_owned(),
908            info: Some(StreamSourceInfo {
909                ..Default::default()
910            }),
911            ..Default::default()
912        };
913        mgr.create_source(pb_source).await?;
914        let source_id: SourceId = Source::find()
915            .select_only()
916            .column(source::Column::SourceId)
917            .filter(source::Column::Name.eq("s1"))
918            .into_tuple()
919            .one(&mgr.inner.read().await.db)
920            .await?
921            .unwrap();
922
923        let pb_view = PbView {
924            schema_id: TEST_SCHEMA_ID,
925            database_id: TEST_DATABASE_ID,
926            name: "view_1".to_owned(),
927            owner: TEST_OWNER_ID as _,
928            sql: "CREATE VIEW view_1 AS SELECT v1 FROM s1".to_owned(),
929            ..Default::default()
930        };
931        mgr.create_view(pb_view, HashSet::from([source_id.as_object_id()]))
932            .await?;
933        let view_id: ViewId = View::find()
934            .select_only()
935            .column(view::Column::ViewId)
936            .filter(view::Column::Name.eq("view_1"))
937            .into_tuple()
938            .one(&mgr.inner.read().await.db)
939            .await?
940            .unwrap();
941
942        mgr.alter_name(ObjectType::Source, source_id, "s2").await?;
943        let source = Source::find_by_id(source_id)
944            .one(&mgr.inner.read().await.db)
945            .await?
946            .unwrap();
947        assert_eq!(source.name, "s2");
948        assert_eq!(
949            source.definition,
950            "CREATE SOURCE s2 (v1 INT) WITH (\
951  connector = 'kafka', \
952  topic = 'kafka_alter', \
953  properties.bootstrap.server = 'message_queue:29092', \
954  scan.startup.mode = 'earliest'\
955) FORMAT PLAIN ENCODE JSON"
956        );
957
958        let view = View::find_by_id(view_id)
959            .one(&mgr.inner.read().await.db)
960            .await?
961            .unwrap();
962        assert_eq!(
963            view.definition,
964            "CREATE VIEW view_1 AS SELECT v1 FROM s2 AS s1"
965        );
966
967        mgr.drop_object(ObjectType::Source, source_id, DropMode::Cascade)
968            .await?;
969        assert!(
970            View::find_by_id(view_id)
971                .one(&mgr.inner.read().await.db)
972                .await?
973                .is_none()
974        );
975
976        Ok(())
977    }
978
979    #[tokio::test]
980    async fn test_cancel_creating_table_deletes_associated_source() -> MetaResult<()> {
981        let env = MetaSrvEnv::for_test().await;
982        let (tx, mut notification_rx) = mpsc::unbounded_channel();
983        env.notification_manager().insert_sender(
984            SubscribeType::Frontend,
985            WorkerKey(HostAddress {
986                host: "localhost".to_owned(),
987                port: 1234,
988            }),
989            tx,
990        );
991        let mgr = CatalogController::new(env).await?;
992
993        let mut inner = mgr.inner.write().await;
994        let txn = inner.db.begin().await?;
995        let source_obj = CatalogController::create_object(
996            &txn,
997            ObjectType::Source,
998            TEST_OWNER_ID,
999            Some(TEST_DATABASE_ID),
1000            Some(TEST_SCHEMA_ID),
1001        )
1002        .await?;
1003        Source::insert(source::ActiveModel::from(PbSource {
1004            id: source_obj.oid.as_source_id(),
1005            schema_id: TEST_SCHEMA_ID,
1006            database_id: TEST_DATABASE_ID,
1007            name: "source_abort_initial".to_owned(),
1008            owner: TEST_OWNER_ID as _,
1009            ..Default::default()
1010        }))
1011        .exec(&txn)
1012        .await?;
1013        let obj = CatalogController::create_object(
1014            &txn,
1015            ObjectType::Table,
1016            TEST_OWNER_ID,
1017            Some(TEST_DATABASE_ID),
1018            Some(TEST_SCHEMA_ID),
1019        )
1020        .await?;
1021        let job_id = obj.oid.as_job_id();
1022
1023        table::ActiveModel {
1024            table_id: Set(obj.oid.as_table_id()),
1025            name: Set("table_abort_initial".to_owned()),
1026            optional_associated_source_id: Set(Some(source_obj.oid.as_source_id())),
1027            table_type: Set(TableType::Table),
1028            belongs_to_job_id: Set(None),
1029            columns: Set(vec![].into()),
1030            pk: Set(vec![].into()),
1031            distribution_key: Set(Vec::<i32>::new().into()),
1032            stream_key: Set(Vec::<i32>::new().into()),
1033            append_only: Set(false),
1034            fragment_id: Set(None),
1035            vnode_col_index: Set(None),
1036            row_id_index: Set(None),
1037            value_indices: Set(Vec::<i32>::new().into()),
1038            definition: Set("CREATE TABLE table_abort_initial (v1 INT)".to_owned()),
1039            handle_pk_conflict_behavior: Set(HandleConflictBehavior::NoCheck),
1040            version_column_indices: Set(None),
1041            read_prefix_len_hint: Set(0),
1042            watermark_indices: Set(Vec::<i32>::new().into()),
1043            dist_key_in_pk: Set(Vec::<i32>::new().into()),
1044            dml_fragment_id: Set(None),
1045            cardinality: Set(None),
1046            cleaned_by_watermark: Set(false),
1047            description: Set(None),
1048            version: Set(None),
1049            retention_seconds: Set(None),
1050            cdc_table_id: Set(None),
1051            vnode_count: Set(1),
1052            webhook_info: Set(None),
1053            engine: Set(None),
1054            clean_watermark_index_in_pk: Set(None),
1055            clean_watermark_indices: Set(None),
1056            refreshable: Set(false),
1057            vector_index_info: Set(None),
1058            cdc_table_type: Set(None),
1059        }
1060        .insert(&txn)
1061        .await?;
1062
1063        streaming_job::ActiveModel {
1064            job_id: Set(job_id),
1065            job_status: Set(JobStatus::Creating),
1066            create_type: Set(CreateType::Foreground),
1067            timezone: Set(None),
1068            config_override: Set(None),
1069            adaptive_parallelism_strategy: Set(None),
1070            parallelism: Set(StreamingParallelism::Adaptive),
1071            backfill_parallelism: Set(None),
1072            backfill_adaptive_parallelism_strategy: Set(None),
1073            backfill_orders: Set(None),
1074            max_parallelism: Set(1),
1075            specific_resource_group: Set(None),
1076            is_serverless_backfill: Set(false),
1077            refresh_interval_sec: Set(None),
1078        }
1079        .insert(&txn)
1080        .await?;
1081
1082        let (tx, rx) = oneshot::channel();
1083        inner.register_finish_notifier(TEST_DATABASE_ID, job_id, tx);
1084        txn.commit().await?;
1085        drop(inner);
1086
1087        let abort_result = mgr.try_abort_creating_streaming_job(job_id, true).await?;
1088        assert!(abort_result.aborted);
1089        assert_eq!(abort_result.database_id, Some(TEST_DATABASE_ID));
1090
1091        let err = rx
1092            .await
1093            .expect("finish notifier should be notified")
1094            .expect_err("creating job cancellation should fail the create wait");
1095        assert!(err.contains("cancelled"));
1096
1097        let db = &mgr.inner.read().await.db;
1098        assert!(Object::find_by_id(job_id).one(db).await?.is_none());
1099        assert!(StreamingJob::find_by_id(job_id).one(db).await?.is_none());
1100        assert!(
1101            Table::find_by_id(job_id.as_mv_table_id())
1102                .one(db)
1103                .await?
1104                .is_none()
1105        );
1106        assert!(
1107            Source::find_by_id(source_obj.oid.as_source_id())
1108                .one(db)
1109                .await?
1110                .is_none()
1111        );
1112
1113        let notification = notification_rx
1114            .recv()
1115            .await
1116            .expect("frontend should receive an abort notification")
1117            .expect("abort notification should be valid");
1118        assert_eq!(notification.operation(), NotificationOperation::Delete);
1119        let object_group = match notification.info {
1120            Some(NotificationInfo::ObjectGroup(object_group)) => object_group,
1121            other => panic!("unexpected notification: {other:?}"),
1122        };
1123        assert!(object_group.objects.iter().any(|object| matches!(
1124            &object.object_info,
1125            Some(PbObjectInfo::Table(table)) if table.id == job_id.as_mv_table_id()
1126        )));
1127        assert!(object_group.objects.iter().any(|object| matches!(
1128            &object.object_info,
1129            Some(PbObjectInfo::Source(source)) if source.id == source_obj.oid.as_source_id()
1130        )));
1131
1132        Ok(())
1133    }
1134
1135    #[tokio::test]
1136    async fn test_clean_dirty_creating_jobs_records_dropped_tables_for_per_db_recovery()
1137    -> MetaResult<()> {
1138        let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
1139
1140        let inner = mgr.inner.write().await;
1141        let txn = inner.db.begin().await?;
1142        let mv_obj = CatalogController::create_object(
1143            &txn,
1144            ObjectType::Table,
1145            TEST_OWNER_ID,
1146            Some(TEST_DATABASE_ID),
1147            Some(TEST_SCHEMA_ID),
1148        )
1149        .await?;
1150        let job_id = mv_obj.oid.as_job_id();
1151        let mv_table_id = job_id.as_mv_table_id();
1152        insert_test_table(
1153            &txn,
1154            mv_table_id,
1155            "mv_dirty",
1156            TableType::MaterializedView,
1157            None,
1158            "CREATE MATERIALIZED VIEW mv_dirty AS SELECT 1",
1159        )
1160        .await?;
1161
1162        let internal_obj = CatalogController::create_object(
1163            &txn,
1164            ObjectType::Table,
1165            TEST_OWNER_ID,
1166            Some(TEST_DATABASE_ID),
1167            Some(TEST_SCHEMA_ID),
1168        )
1169        .await?;
1170        let internal_table_id = internal_obj.oid.as_table_id();
1171        insert_test_table(
1172            &txn,
1173            internal_table_id,
1174            "__internal_mv_dirty",
1175            TableType::Internal,
1176            Some(job_id),
1177            "",
1178        )
1179        .await?;
1180
1181        streaming_job::ActiveModel {
1182            job_id: Set(job_id),
1183            job_status: Set(JobStatus::Creating),
1184            create_type: Set(CreateType::Foreground),
1185            timezone: Set(None),
1186            config_override: Set(None),
1187            adaptive_parallelism_strategy: Set(None),
1188            parallelism: Set(StreamingParallelism::Adaptive),
1189            backfill_parallelism: Set(None),
1190            backfill_adaptive_parallelism_strategy: Set(None),
1191            backfill_orders: Set(None),
1192            max_parallelism: Set(1),
1193            specific_resource_group: Set(None),
1194            is_serverless_backfill: Set(false),
1195            refresh_interval_sec: Set(None),
1196        }
1197        .insert(&txn)
1198        .await?;
1199        txn.commit().await?;
1200        drop(inner);
1201
1202        let cleaned = mgr
1203            .clean_dirty_creating_jobs(Some(TEST_DATABASE_ID))
1204            .await?;
1205        assert_eq!(cleaned.streaming_job_ids, vec![job_id]);
1206        assert!(cleaned.source_ids.is_empty());
1207        let mut dropped_table_ids = cleaned.dropped_table_ids;
1208        dropped_table_ids.sort_unstable();
1209        assert_eq!(dropped_table_ids, vec![mv_table_id, internal_table_id]);
1210
1211        let inner = mgr.inner.read().await;
1212        assert!(inner.dropped_tables.contains_key(&mv_table_id));
1213        assert!(inner.dropped_tables.contains_key(&internal_table_id));
1214        assert!(Object::find_by_id(job_id).one(&inner.db).await?.is_none());
1215        assert!(
1216            StreamingJob::find_by_id(job_id)
1217                .one(&inner.db)
1218                .await?
1219                .is_none()
1220        );
1221        assert!(
1222            Table::find_by_id(mv_table_id)
1223                .one(&inner.db)
1224                .await?
1225                .is_none()
1226        );
1227        assert!(
1228            Table::find_by_id(internal_table_id)
1229                .one(&inner.db)
1230                .await?
1231                .is_none()
1232        );
1233
1234        Ok(())
1235    }
1236
1237    #[tokio::test]
1238    async fn test_clean_dirty_creating_jobs_notifies_serving_mapping_fragment_delete()
1239    -> MetaResult<()> {
1240        let env = MetaSrvEnv::for_test().await;
1241        let (local_notification_tx, mut local_notification_rx) = mpsc::unbounded_channel();
1242        env.notification_manager()
1243            .insert_local_sender(local_notification_tx);
1244        let mgr = CatalogController::new(env).await?;
1245        let fragment_id = FragmentId::new(3);
1246        let (job_id, mv_table_id) =
1247            insert_dirty_creating_job_with_fragment(&mgr, fragment_id, 1).await?;
1248
1249        assert!(
1250            mgr.fragment_serving_infos()
1251                .await?
1252                .contains_key(&fragment_id)
1253        );
1254
1255        let cleaned = mgr
1256            .clean_dirty_creating_jobs(Some(TEST_DATABASE_ID))
1257            .await?;
1258        assert_eq!(cleaned.streaming_job_ids, vec![job_id]);
1259
1260        let inner = mgr.inner.read().await;
1261        assert!(Object::find_by_id(job_id).one(&inner.db).await?.is_none());
1262        assert!(
1263            StreamingJob::find_by_id(job_id)
1264                .one(&inner.db)
1265                .await?
1266                .is_none()
1267        );
1268        assert!(
1269            Table::find_by_id(mv_table_id)
1270                .one(&inner.db)
1271                .await?
1272                .is_none()
1273        );
1274        drop(inner);
1275        assert!(
1276            !mgr.fragment_serving_infos()
1277                .await?
1278                .contains_key(&fragment_id)
1279        );
1280
1281        let notification = local_notification_rx.try_recv().expect(
1282            "dirty-job cleanup must notify the serving mapping worker about deleted fragments",
1283        );
1284        match notification {
1285            LocalNotification::ServingFragmentMappingsDelete(fragment_ids) => {
1286                assert_eq!(fragment_ids, vec![fragment_id]);
1287            }
1288            notification => panic!("unexpected local notification: {notification:?}"),
1289        }
1290
1291        Ok(())
1292    }
1293
1294    #[tokio::test]
1295    async fn test_abort_creating_subscription_commits_delete() -> MetaResult<()> {
1296        let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
1297        let pb_view = PbView {
1298            schema_id: TEST_SCHEMA_ID,
1299            database_id: TEST_DATABASE_ID,
1300            name: "subscription_dep_view".to_owned(),
1301            owner: TEST_OWNER_ID as _,
1302            sql: "CREATE VIEW subscription_dep_view AS SELECT 1".to_owned(),
1303            ..Default::default()
1304        };
1305        mgr.create_view(pb_view, HashSet::new()).await?;
1306
1307        let view_id: ViewId = View::find()
1308            .select_only()
1309            .column(view::Column::ViewId)
1310            .filter(view::Column::Name.eq("subscription_dep_view"))
1311            .into_tuple()
1312            .one(&mgr.inner.read().await.db)
1313            .await?
1314            .unwrap();
1315
1316        let mut pb_subscription = PbSubscription {
1317            name: "subscription_to_abort".to_owned(),
1318            definition: "CREATE SUBSCRIPTION subscription_to_abort FROM subscription_dep_view"
1319                .to_owned(),
1320            retention_seconds: 86400,
1321            database_id: TEST_DATABASE_ID,
1322            schema_id: TEST_SCHEMA_ID,
1323            dependent_table_id: view_id.as_object_id().as_table_id(),
1324            owner: TEST_OWNER_ID as _,
1325            subscription_state: SubscriptionState::Init as _,
1326            ..Default::default()
1327        };
1328        mgr.create_subscription_catalog(&mut pb_subscription)
1329            .await?;
1330
1331        mgr.try_abort_creating_subscription(pb_subscription.id)
1332            .await?;
1333
1334        assert!(
1335            Subscription::find_by_id(pb_subscription.id)
1336                .one(&mgr.inner.read().await.db)
1337                .await?
1338                .is_none()
1339        );
1340
1341        Ok(())
1342    }
1343
1344    #[tokio::test]
1345    async fn test_drop_table_cascade_drops_dependent_subscription() -> MetaResult<()> {
1346        let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
1347
1348        let inner = mgr.inner.write().await;
1349        let txn = inner.db.begin().await?;
1350        let table_obj = CatalogController::create_object(
1351            &txn,
1352            ObjectType::Table,
1353            TEST_OWNER_ID,
1354            Some(TEST_DATABASE_ID),
1355            Some(TEST_SCHEMA_ID),
1356        )
1357        .await?;
1358        let table_id = table_obj.oid.as_table_id();
1359        insert_test_table(
1360            &txn,
1361            table_id,
1362            "subscription_dep_table",
1363            TableType::Table,
1364            None,
1365            "CREATE TABLE subscription_dep_table (v1 INT)",
1366        )
1367        .await?;
1368        txn.commit().await?;
1369        drop(inner);
1370
1371        let mut pb_subscription = PbSubscription {
1372            name: "subscription_to_drop_with_table".to_owned(),
1373            definition:
1374                "CREATE SUBSCRIPTION subscription_to_drop_with_table FROM subscription_dep_table"
1375                    .to_owned(),
1376            retention_seconds: 86400,
1377            database_id: TEST_DATABASE_ID,
1378            schema_id: TEST_SCHEMA_ID,
1379            dependent_table_id: table_id,
1380            owner: TEST_OWNER_ID as _,
1381            subscription_state: SubscriptionState::Created as _,
1382            ..Default::default()
1383        };
1384        mgr.create_subscription_catalog(&mut pb_subscription)
1385            .await?;
1386
1387        mgr.drop_object(ObjectType::Table, table_id, DropMode::Cascade)
1388            .await?;
1389
1390        let db = &mgr.inner.read().await.db;
1391        assert!(Table::find_by_id(table_id).one(db).await?.is_none());
1392        assert!(
1393            Object::find_by_id(table_id.as_object_id())
1394                .one(db)
1395                .await?
1396                .is_none()
1397        );
1398        assert!(
1399            Subscription::find_by_id(pb_subscription.id)
1400                .one(db)
1401                .await?
1402                .is_none()
1403        );
1404        assert!(
1405            Object::find_by_id(pb_subscription.id.as_object_id())
1406                .one(db)
1407                .await?
1408                .is_none()
1409        );
1410
1411        Ok(())
1412    }
1413}
risingwave_meta/controller/catalog/test.rs

risingwave_meta/controller/catalog/
test.rs
