1#[cfg(test)]
16mod tests {
17 use risingwave_common::catalog::{FragmentTypeFlag, FragmentTypeMask};
18 use risingwave_common::hash::VirtualNode;
19 use risingwave_meta_model::FragmentId;
20 use risingwave_meta_model::fragment::DistributionType;
21 use risingwave_meta_model::table::HandleConflictBehavior;
22 use risingwave_pb::catalog::subscription::SubscriptionState;
23 use risingwave_pb::catalog::{PbSinkType, StreamSourceInfo};
24 use risingwave_pb::common::{HostAddress, WorkerNode, WorkerType, worker_node};
25 use risingwave_pb::meta::SubscribeType;
26 use risingwave_pb::meta::table_fragments::fragment::PbFragmentDistributionType;
27 use risingwave_pb::stream_plan::PbStreamNode;
28 use tokio::sync::{mpsc, oneshot};
29
30 use crate::barrier::Command;
31 use crate::controller::catalog::*;
32 use crate::manager::{LocalNotification, MetaOpts, WorkerKey};
33 use crate::model::{Fragment, FragmentDownstreamRelation};
34 use crate::serving::ServingVnodeMapping;
35
36 const TEST_DATABASE_ID: DatabaseId = DatabaseId::new(1);
37 const TEST_SCHEMA_ID: SchemaId = SchemaId::new(2);
38 const TEST_OWNER_ID: UserId = UserId::new(1);
39
40 async fn insert_test_table(
41 txn: &DatabaseTransaction,
42 table_id: TableId,
43 name: &str,
44 table_type: TableType,
45 belongs_to_job_id: Option<JobId>,
46 definition: &str,
47 ) -> MetaResult<()> {
48 table::ActiveModel {
49 table_id: Set(table_id),
50 name: Set(name.to_owned()),
51 optional_associated_source_id: Set(None),
52 table_type: Set(table_type),
53 belongs_to_job_id: Set(belongs_to_job_id),
54 columns: Set(vec![].into()),
55 pk: Set(vec![].into()),
56 distribution_key: Set(Vec::<i32>::new().into()),
57 stream_key: Set(Vec::<i32>::new().into()),
58 append_only: Set(false),
59 fragment_id: Set(None),
60 vnode_col_index: Set(None),
61 row_id_index: Set(None),
62 value_indices: Set(Vec::<i32>::new().into()),
63 definition: Set(definition.to_owned()),
64 handle_pk_conflict_behavior: Set(HandleConflictBehavior::NoCheck),
65 version_column_indices: Set(None),
66 read_prefix_len_hint: Set(0),
67 watermark_indices: Set(Vec::<i32>::new().into()),
68 dist_key_in_pk: Set(Vec::<i32>::new().into()),
69 dml_fragment_id: Set(None),
70 cardinality: Set(None),
71 cleaned_by_watermark: Set(false),
72 description: Set(None),
73 version: Set(None),
74 retention_seconds: Set(None),
75 cdc_table_id: Set(None),
76 vnode_count: Set(1),
77 webhook_info: Set(None),
78 engine: Set(None),
79 clean_watermark_index_in_pk: Set(None),
80 clean_watermark_indices: Set(None),
81 refreshable: Set(false),
82 vector_index_info: Set(None),
83 cdc_table_type: Set(None),
84 }
85 .insert(txn)
86 .await?;
87 Ok(())
88 }
89
90 async fn insert_test_fragment(
91 txn: &DatabaseTransaction,
92 fragment_id: FragmentId,
93 job_id: JobId,
94 state_table_ids: TableIdArray,
95 ) -> MetaResult<()> {
96 fragment::ActiveModel {
97 fragment_id: Set(fragment_id),
98 job_id: Set(job_id),
99 fragment_type_mask: Set(0),
100 distribution_type: Set(fragment::DistributionType::Hash),
101 stream_node: Set(StreamNode::from(&PbStreamNode::default())),
102 state_table_ids: Set(state_table_ids),
103 upstream_fragment_id: Set(I32Array::default()),
104 vnode_count: Set(1),
105 parallelism: Set(None),
106 }
107 .insert(txn)
108 .await?;
109 Ok(())
110 }
111
112 async fn insert_test_streaming_job(
113 txn: &DatabaseTransaction,
114 name: &str,
115 has_result_table: bool,
116 policy: Option<CacheRefillPolicy>,
117 ) -> MetaResult<(JobId, Option<TableId>, TableId)> {
118 let object_type = if has_result_table {
119 ObjectType::Table
120 } else {
121 ObjectType::Sink
122 };
123 let job_id = CatalogController::create_object(
124 txn,
125 object_type,
126 TEST_OWNER_ID,
127 Some(TEST_SCHEMA_ID.as_object_id()),
128 )
129 .await?
130 .oid
131 .as_job_id();
132 let result_table_id = has_result_table.then_some(job_id.as_mv_table_id());
133 if let Some(table_id) = result_table_id {
134 insert_test_table(txn, table_id, name, TableType::MaterializedView, None, "").await?;
135 }
136
137 let internal_table_id = CatalogController::create_object(
138 txn,
139 ObjectType::Table,
140 TEST_OWNER_ID,
141 Some(job_id.as_object_id()),
142 )
143 .await?
144 .oid
145 .as_table_id();
146 insert_test_table(
147 txn,
148 internal_table_id,
149 &format!("__internal_{name}"),
150 TableType::Internal,
151 Some(job_id),
152 "",
153 )
154 .await?;
155
156 insert_test_streaming_job_model(txn, job_id, policy).await?;
157
158 Ok((job_id, result_table_id, internal_table_id))
159 }
160
161 async fn insert_test_streaming_job_model(
162 txn: &DatabaseTransaction,
163 job_id: JobId,
164 policy: Option<CacheRefillPolicy>,
165 ) -> MetaResult<()> {
166 streaming_job::ActiveModel {
167 job_id: Set(job_id),
168 job_status: Set(JobStatus::Created),
169 create_type: Set(CreateType::Foreground),
170 timezone: Set(None),
171 config_override: Set(policy.map(|policy| {
172 format!(
173 "[streaming.developer]\ncache_refill_policy = \"{}\"\n",
174 policy
175 )
176 })),
177 adaptive_parallelism_strategy: Set(None),
178 parallelism: Set(StreamingParallelism::Adaptive),
179 backfill_parallelism: Set(None),
180 backfill_adaptive_parallelism_strategy: Set(None),
181 backfill_orders: Set(None),
182 max_parallelism: Set(1),
183 specific_resource_group: Set(None),
184 is_serverless_backfill: Set(false),
185 refresh_interval_sec: Set(None),
186 }
187 .insert(txn)
188 .await?;
189
190 Ok(())
191 }
192
193 #[tokio::test]
194 async fn test_cancel_creating_job_includes_belonging_streaming_jobs() -> MetaResult<()> {
195 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
196 let mut inner = mgr.inner.write().await;
197 let txn = inner.db.begin().await?;
198
199 let table_job_id = CatalogController::create_object(
200 &txn,
201 ObjectType::Table,
202 TEST_OWNER_ID,
203 Some(TEST_SCHEMA_ID.as_object_id()),
204 )
205 .await?
206 .oid
207 .as_job_id();
208 insert_test_table(
209 &txn,
210 table_job_id.as_mv_table_id(),
211 "cancel_table",
212 TableType::Table,
213 None,
214 "",
215 )
216 .await?;
217 let sink_job_id = CatalogController::create_object(
218 &txn,
219 ObjectType::Sink,
220 TEST_OWNER_ID,
221 Some(table_job_id.as_object_id()),
222 )
223 .await?
224 .oid
225 .as_job_id();
226 Sink::insert(sink::ActiveModel::from(PbSink {
227 id: sink_job_id.as_sink_id(),
228 schema_id: TEST_SCHEMA_ID,
229 database_id: TEST_DATABASE_ID,
230 name: "cancel_sink".to_owned(),
231 owner: TEST_OWNER_ID as _,
232 sink_type: PbSinkType::AppendOnly as i32,
233 ..Default::default()
234 }))
235 .exec(&txn)
236 .await?;
237 for job_id in [table_job_id, sink_job_id] {
238 insert_test_streaming_job_model(&txn, job_id, None).await?;
239 StreamingJob::update(streaming_job::ActiveModel {
240 job_id: Set(job_id),
241 job_status: Set(JobStatus::Creating),
242 ..Default::default()
243 })
244 .exec(&txn)
245 .await?;
246 }
247
248 let table_state_id = TableId::new(1000);
249 let sink_state_id = TableId::new(1001);
250 insert_test_fragment(
251 &txn,
252 FragmentId::new(100),
253 table_job_id,
254 TableIdArray(vec![table_state_id]),
255 )
256 .await?;
257 insert_test_fragment(
258 &txn,
259 FragmentId::new(101),
260 sink_job_id,
261 TableIdArray(vec![sink_state_id]),
262 )
263 .await?;
264 let (table_finish_tx, table_finish_rx) = oneshot::channel();
265 inner.register_finish_notifier(TEST_DATABASE_ID, table_job_id, table_finish_tx);
266 let (sink_finish_tx, sink_finish_rx) = oneshot::channel();
267 inner.register_finish_notifier(TEST_DATABASE_ID, sink_job_id, sink_finish_tx);
268 txn.commit().await?;
269 drop(inner);
270
271 let abort_result = mgr
272 .try_abort_creating_streaming_job(table_job_id, true)
273 .await?;
274 assert!(abort_result.aborted);
275 assert_eq!(
276 abort_result.aborted_sink_ids,
277 vec![sink_job_id.as_sink_id()]
278 );
279 let cancel_info = abort_result
280 .cancel_info
281 .expect("cancelled table job should have cleanup information");
282 assert_eq!(
283 cancel_info
284 .streaming_job_ids
285 .iter()
286 .copied()
287 .collect::<HashSet<_>>(),
288 HashSet::from([table_job_id, sink_job_id])
289 );
290 assert_eq!(
291 cancel_info
292 .state_table_ids
293 .iter()
294 .copied()
295 .collect::<HashSet<_>>(),
296 HashSet::from([table_state_id, sink_state_id])
297 );
298 let Command::DropStreamingJobs {
299 streaming_job_ids,
300 unregistered_state_table_ids,
301 ..
302 } = cancel_info.command
303 else {
304 unreachable!()
305 };
306 assert_eq!(
307 streaming_job_ids,
308 HashSet::from([table_job_id, sink_job_id])
309 );
310 assert_eq!(
311 unregistered_state_table_ids,
312 HashSet::from([table_state_id, sink_state_id])
313 );
314
315 for finish_rx in [table_finish_rx, sink_finish_rx] {
316 let err = finish_rx
317 .await
318 .expect("aborted job should notify its finish waiter")
319 .expect_err("aborted job should not finish successfully");
320 assert!(err.contains("cancelled"));
321 }
322 let db = &mgr.inner.read().await.db;
323 assert!(Object::find_by_id(table_job_id).one(db).await?.is_none());
324 assert!(Object::find_by_id(sink_job_id).one(db).await?.is_none());
325
326 Ok(())
327 }
328
329 #[tokio::test]
330 async fn test_create_multiple_sinks_into_same_table_and_drop_table() -> MetaResult<()> {
331 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
332 let inner = mgr.inner.write().await;
333 let txn = inner.db.begin().await?;
334 let (_, Some(target_table_id), _) =
335 insert_test_streaming_job(&txn, "mvt", true, None).await?
336 else {
337 unreachable!()
338 };
339 let (mv1_id, Some(_), _) = insert_test_streaming_job(&txn, "mv1", true, None).await? else {
340 unreachable!()
341 };
342 let (mv2_id, Some(_), _) = insert_test_streaming_job(&txn, "mv2", true, None).await? else {
343 unreachable!()
344 };
345 txn.commit().await?;
346 drop(inner);
347
348 let mut sink_ids = Vec::new();
349 let test_sink_tuples = [("s1", mv1_id), ("s2", mv2_id)];
350
351 fn assert_incoming_sink_drop_error<T>(error: &MetaError, test_sink_tuples: &[(&str, T)]) {
352 let message = error.to_string();
353
354 assert!(
355 message.contains("sink") && message.contains("depends on it"),
356 "expected an incoming-sink dependency error, got: {message}"
357 );
358
359 assert!(
360 test_sink_tuples
361 .iter()
362 .all(|(sink_name, _)| message.contains(*sink_name)),
363 "expected the error to mention all incoming sinks, got: {message}"
364 );
365 }
366
367 for (name, source) in test_sink_tuples {
368 let mut job = crate::manager::StreamingJob::Sink(
369 PbSink {
370 name: name.to_owned(),
371 database_id: TEST_DATABASE_ID,
372 schema_id: TEST_SCHEMA_ID,
373 owner: TEST_OWNER_ID as _,
374 target_table: Some(target_table_id),
375 sink_type: PbSinkType::AppendOnly as i32,
376 ..Default::default()
377 },
378 None,
379 );
380 tokio::time::timeout(
384 std::time::Duration::from_secs(3),
385 mgr.create_job_catalog(
386 &mut job,
387 &crate::model::StreamContext::default(),
388 &None,
389 1,
390 HashSet::from([source.as_object_id()]),
391 risingwave_pb::ddl_service::streaming_job_resource_type::ResourceType::Regular(
392 true,
393 ),
394 &None,
395 None,
396 None,
397 None,
398 None,
399 ),
400 )
401 .await
402 .expect("creating a second sink into the same table should not hang")?;
403 sink_ids.push(job.id().as_object_id());
404 }
405 let owned_sink_name = "owned_sink_without_iceberg_prefix";
406 let mut owned_sink = crate::manager::StreamingJob::Sink(
407 PbSink {
408 name: owned_sink_name.to_owned(),
409 database_id: TEST_DATABASE_ID,
410 schema_id: TEST_SCHEMA_ID,
411 owner: TEST_OWNER_ID as _,
412 target_table: Some(target_table_id),
413 sink_type: PbSinkType::AppendOnly as i32,
414 ..Default::default()
415 },
416 Some(target_table_id),
417 );
418 mgr.create_job_catalog(
419 &mut owned_sink,
420 &crate::model::StreamContext::default(),
421 &None,
422 1,
423 HashSet::from([mv1_id.as_object_id()]),
424 risingwave_pb::ddl_service::streaming_job_resource_type::ResourceType::Regular(true),
425 &None,
426 None,
427 None,
428 None,
429 None,
430 )
431 .await?;
432 let owned_sink_id = owned_sink.id().as_object_id();
433 sink_ids.push(owned_sink_id);
434
435 assert_eq!(sink_ids.len(), test_sink_tuples.len() + 1);
437
438 let inner = mgr.inner.read().await;
439 for sink_id in &sink_ids {
440 streaming_job::ActiveModel {
441 job_id: Set(sink_id.as_job_id()),
442 job_status: Set(JobStatus::Created),
443 ..Default::default()
444 }
445 .update(&inner.db)
446 .await?;
447 }
448 let object_dependency_count = ObjectDependency::find()
451 .filter(object_dependency::Column::Oid.eq(target_table_id.as_object_id()))
452 .filter(object_dependency::Column::UsedBy.is_in(sink_ids.clone()))
453 .count(&inner.db)
454 .await?;
455 assert_eq!(object_dependency_count, 0);
456 assert_eq!(
457 Object::find_by_id(owned_sink_id)
458 .one(&inner.db)
459 .await?
460 .unwrap()
461 .belong_to_oid,
462 Some(target_table_id.as_object_id())
463 );
464 drop(inner);
465
466 let error = mgr
467 .drop_object(ObjectType::Table, target_table_id, DropMode::Restrict)
468 .await
469 .expect_err("RESTRICT drop should fail for a table with incoming sinks");
470 assert_incoming_sink_drop_error(&error, &test_sink_tuples);
471 assert!(
472 !error.to_string().contains(owned_sink_name),
473 "an owned incoming sink should not prevent a RESTRICT drop"
474 );
475 mgr.drop_object(ObjectType::Table, target_table_id, DropMode::Cascade)
476 .await
477 .expect("CASCADE drop should succeed");
478
479 let inner = mgr.inner.read().await;
480 let db = &inner.db;
481 assert!(Object::find_by_id(target_table_id).one(db).await?.is_none());
483 assert_eq!(
484 Object::find()
485 .filter(object::Column::Oid.is_in(sink_ids))
486 .count(db)
487 .await?,
488 0
489 );
490 assert!(Object::find_by_id(mv1_id).one(db).await?.is_some());
492 assert!(Object::find_by_id(mv2_id).one(db).await?.is_some());
493
494 Ok(())
495 }
496
497 #[tokio::test]
498 async fn test_replace_upstream_object_rejects_creating_incoming_sink() -> MetaResult<()> {
499 fn assert_replace_concurrency_error(error: &MetaError) {
500 let message = error.to_string();
501 assert!(
503 message.contains("referenced by some creating jobs"),
504 "expected a replace concurrency error, got: {message}"
505 );
506 }
507
508 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
509 let inner = mgr.inner.write().await;
510 let txn = inner.db.begin().await?;
511 let (target_mv_id, Some(target_mv_table_id), _) =
512 insert_test_streaming_job(&txn, "target_mv", true, None).await?
513 else {
514 unreachable!()
515 };
516 let (source_mv_id, Some(_), _) =
517 insert_test_streaming_job(&txn, "source_mv", true, None).await?
518 else {
519 unreachable!()
520 };
521 txn.commit().await?;
522 drop(inner);
523
524 let mut sink = crate::manager::StreamingJob::Sink(
525 PbSink {
526 name: "creating_sink".to_owned(),
527 database_id: TEST_DATABASE_ID,
528 schema_id: TEST_SCHEMA_ID,
529 owner: TEST_OWNER_ID as _,
530 target_table: Some(target_mv_table_id),
531 sink_type: PbSinkType::AppendOnly as i32,
532 ..Default::default()
533 },
534 None,
535 );
536 let creating_sink = mgr
537 .create_job_catalog(
538 &mut sink,
539 &crate::model::StreamContext::default(),
540 &None,
541 1,
542 HashSet::from([source_mv_id.as_object_id()]),
543 risingwave_pb::ddl_service::streaming_job_resource_type::ResourceType::Regular(
544 true,
545 ),
546 &None,
547 None,
548 None,
549 None,
550 None,
551 )
552 .await?;
553 assert_ne!(creating_sink.job_status, JobStatus::Created);
555
556 let replacement = crate::manager::StreamingJob::MaterializedView(PbTable {
557 id: target_mv_table_id,
558 name: "target_mv".to_owned(),
559 database_id: TEST_DATABASE_ID,
560 schema_id: TEST_SCHEMA_ID,
561 owner: TEST_OWNER_ID as _,
562 ..Default::default()
563 });
564 assert_eq!(replacement.id(), target_mv_id);
566
567 let error = mgr
569 .create_job_catalog_for_replace(&replacement, None, None, None)
570 .await
571 .expect_err("replacement should reject a creating incoming sink");
572 assert_replace_concurrency_error(&error);
574
575 Ok(())
576 }
577
578 #[tokio::test]
579 async fn test_replace_upstream_object_with_created_incoming_sink() -> MetaResult<()> {
580 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
581 let inner = mgr.inner.write().await;
582 let txn = inner.db.begin().await?;
583 let (_, Some(target_table_id), _) =
584 insert_test_streaming_job(&txn, "target_table", true, None).await?
585 else {
586 unreachable!()
587 };
588 let (source_mv_id, Some(source_mv_table_id), _) =
589 insert_test_streaming_job(&txn, "source_mv", true, None).await?
590 else {
591 unreachable!()
592 };
593 txn.commit().await?;
594 drop(inner);
595
596 let mut sink = crate::manager::StreamingJob::Sink(
597 PbSink {
598 name: "created_sink".to_owned(),
599 database_id: TEST_DATABASE_ID,
600 schema_id: TEST_SCHEMA_ID,
601 owner: TEST_OWNER_ID as _,
602 target_table: Some(target_table_id),
603 sink_type: PbSinkType::AppendOnly as i32,
604 ..Default::default()
605 },
606 None,
607 );
608 let creating_sink = mgr
609 .create_job_catalog(
610 &mut sink,
611 &crate::model::StreamContext::default(),
612 &None,
613 1,
614 HashSet::from([source_mv_id.as_object_id()]),
615 risingwave_pb::ddl_service::streaming_job_resource_type::ResourceType::Regular(
616 true,
617 ),
618 &None,
619 None,
620 None,
621 None,
622 None,
623 )
624 .await?;
625 assert_ne!(creating_sink.job_status, JobStatus::Created);
627
628 let sink_id = sink.id();
629 let inner = mgr.inner.read().await;
630 streaming_job::ActiveModel {
631 job_id: Set(sink_id),
632 job_status: Set(JobStatus::Created),
633 ..Default::default()
634 }
635 .update(&inner.db)
636 .await?;
637 let sink_model = risingwave_meta_model::prelude::StreamingJob::find_by_id(sink_id)
638 .one(&inner.db)
639 .await?
640 .expect("sink should exist");
641 assert_eq!(sink_model.job_id, sink_id);
643 assert_eq!(sink_model.job_status, JobStatus::Created);
645 drop(inner);
646
647 let replacement = crate::manager::StreamingJob::MaterializedView(PbTable {
648 id: source_mv_table_id,
649 name: "source_mv".to_owned(),
650 database_id: TEST_DATABASE_ID,
651 schema_id: TEST_SCHEMA_ID,
652 owner: TEST_OWNER_ID as _,
653 ..Default::default()
654 });
655 assert_eq!(replacement.id(), source_mv_id);
657
658 let tmp_model = mgr
659 .create_job_catalog_for_replace(&replacement, None, None, None)
660 .await?;
661
662 assert_ne!(tmp_model.job_id, source_mv_id);
664 assert_eq!(tmp_model.job_status, JobStatus::Initial);
666
667 let inner = mgr.inner.read().await;
668 let db = &inner.db;
669 assert_eq!(
671 ObjectDependency::find()
672 .filter(object_dependency::Column::Oid.eq(source_mv_id.as_object_id()))
673 .filter(object_dependency::Column::UsedBy.eq(sink_id.as_object_id()))
674 .count(db)
675 .await?,
676 1
677 );
678 assert_eq!(
680 ObjectDependency::find()
681 .filter(object_dependency::Column::Oid.eq(source_mv_id.as_object_id()))
682 .filter(object_dependency::Column::UsedBy.eq(tmp_model.job_id.as_object_id()))
683 .count(db)
684 .await?,
685 1
686 );
687
688 Ok(())
689 }
690
691 #[tokio::test]
692 async fn test_table_refill_catalog_snapshot_classifies_table_identity() -> MetaResult<()> {
693 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
694 let inner = mgr.inner.write().await;
695 let txn = inner.db.begin().await?;
696
697 let (mv_job, Some(mv_result), mv_internal) =
698 insert_test_streaming_job(&txn, "mv_both", true, Some(CacheRefillPolicy::Both)).await?
699 else {
700 unreachable!()
701 };
702 let (default_job, Some(_default_result), default_internal) =
703 insert_test_streaming_job(&txn, "mv_default", true, None).await?
704 else {
705 unreachable!()
706 };
707 let (sink_job, None, sink_internal) = insert_test_streaming_job(
708 &txn,
709 "sink_streaming",
710 false,
711 Some(CacheRefillPolicy::Streaming),
712 )
713 .await?
714 else {
715 unreachable!()
716 };
717
718 let result_fragment = FragmentId::new(100);
719 let internal_fragment = FragmentId::new(101);
720 let sink_fragment = FragmentId::new(102);
721 for (fragment_id, job_id, table_ids) in [
722 (result_fragment, mv_job, vec![mv_result, mv_internal]),
723 (internal_fragment, default_job, vec![default_internal]),
724 (sink_fragment, sink_job, vec![sink_internal]),
725 ] {
726 insert_test_fragment(&txn, fragment_id, job_id, TableIdArray(table_ids)).await?;
727 }
728 txn.commit().await?;
729 drop(inner);
730
731 let serving_infos = mgr.fragment_serving_infos().await?;
732 assert_eq!(serving_infos.len(), 3);
733 assert_eq!(
734 serving_infos[&result_fragment].result_table_id,
735 Some(mv_result)
736 );
737 assert_eq!(serving_infos[&internal_fragment].result_table_id, None);
738 assert_eq!(serving_infos[&sink_fragment].result_table_id, None);
739
740 let policies = mgr.table_cache_refill_policies_snapshot().await?;
741 assert_eq!(
742 policies
743 .table_policies
744 .into_iter()
745 .map(|policy| (policy.table_id, policy.policy))
746 .collect::<HashMap<_, _>>(),
747 HashMap::from([(
748 mv_result.as_raw_id(),
749 CacheRefillPolicy::Both.to_protobuf() as i32,
750 )])
751 );
752 assert_eq!(
753 policies
754 .internal_table_policies
755 .into_iter()
756 .map(|policy| (policy.table_id, policy.policy))
757 .collect::<HashMap<_, _>>(),
758 HashMap::from([
759 (
760 mv_internal.as_raw_id(),
761 CacheRefillPolicy::Both.to_protobuf() as i32,
762 ),
763 (
764 sink_internal.as_raw_id(),
765 CacheRefillPolicy::Streaming.to_protobuf() as i32,
766 ),
767 ])
768 );
769
770 Ok(())
771 }
772
773 #[tokio::test]
774 async fn test_foreground_creating_catalog_lifecycle() -> MetaResult<()> {
775 let env = MetaSrvEnv::for_test().await;
776 let (tx, mut notification_rx) = mpsc::unbounded_channel();
777 env.notification_manager().insert_sender(
778 SubscribeType::Frontend,
779 WorkerKey(HostAddress {
780 host: "localhost".to_owned(),
781 port: 1234,
782 }),
783 tx,
784 );
785 let mgr = CatalogController::new(env).await?;
786 let inner = mgr.inner.write().await;
787 let txn = inner.db.begin().await?;
788
789 let (job_id, Some(table_id), internal_table_id) =
791 insert_test_streaming_job(&txn, "creating_table", true, None).await?
792 else {
793 unreachable!()
794 };
795 let associated_source_id = CatalogController::create_object(
796 &txn,
797 ObjectType::Source,
798 TEST_OWNER_ID,
799 Some(TEST_SCHEMA_ID.as_object_id()),
800 )
801 .await?
802 .oid
803 .as_source_id();
804 Source::insert(source::ActiveModel::from(PbSource {
805 id: associated_source_id,
806 schema_id: TEST_SCHEMA_ID,
807 database_id: TEST_DATABASE_ID,
808 name: "creating_table_source".to_owned(),
809 owner: TEST_OWNER_ID as _,
810 optional_associated_table_id: Some(
811 risingwave_pb::catalog::source::OptionalAssociatedTableId::AssociatedTableId(
812 table_id,
813 ),
814 ),
815 ..Default::default()
816 }))
817 .exec(&txn)
818 .await?;
819 table::ActiveModel {
820 table_id: Set(table_id),
821 table_type: Set(TableType::Table),
822 optional_associated_source_id: Set(Some(associated_source_id)),
823 ..Default::default()
824 }
825 .update(&txn)
826 .await?;
827 streaming_job::ActiveModel {
828 job_id: Set(job_id),
829 job_status: Set(JobStatus::Initial),
830 ..Default::default()
831 }
832 .update(&txn)
833 .await?;
834
835 let (_primary_job_id, Some(primary_table_id), _) =
837 insert_test_streaming_job(&txn, "primary_table", true, None).await?
838 else {
839 unreachable!()
840 };
841 let index_job_id = CatalogController::create_object(
842 &txn,
843 ObjectType::Index,
844 TEST_OWNER_ID,
845 Some(TEST_SCHEMA_ID.as_object_id()),
846 )
847 .await?
848 .oid
849 .as_job_id();
850 let index_table_id = index_job_id.as_mv_table_id();
851 insert_test_table(
852 &txn,
853 index_table_id,
854 "creating_index",
855 TableType::Index,
856 None,
857 "",
858 )
859 .await?;
860 index::ActiveModel {
861 index_id: Set(index_job_id.as_index_id()),
862 name: Set("creating_index".to_owned()),
863 index_table_id: Set(index_table_id),
864 primary_table_id: Set(primary_table_id),
865 index_items: Set(vec![].into()),
866 index_column_properties: Set(None),
867 index_columns_len: Set(0),
868 }
869 .insert(&txn)
870 .await?;
871 insert_test_streaming_job_model(&txn, index_job_id, None).await?;
872 streaming_job::ActiveModel {
873 job_id: Set(index_job_id),
874 job_status: Set(JobStatus::Initial),
875 ..Default::default()
876 }
877 .update(&txn)
878 .await?;
879
880 let source_job_id = CatalogController::create_object(
882 &txn,
883 ObjectType::Source,
884 TEST_OWNER_ID,
885 Some(TEST_SCHEMA_ID.as_object_id()),
886 )
887 .await?
888 .oid
889 .as_job_id();
890 Source::insert(source::ActiveModel::from(PbSource {
891 id: source_job_id.as_shared_source_id(),
892 schema_id: TEST_SCHEMA_ID,
893 database_id: TEST_DATABASE_ID,
894 name: "creating_shared_source".to_owned(),
895 owner: TEST_OWNER_ID as _,
896 info: Some(StreamSourceInfo {
897 cdc_source_job: true,
898 ..Default::default()
899 }),
900 ..Default::default()
901 }))
902 .exec(&txn)
903 .await?;
904 insert_test_streaming_job_model(&txn, source_job_id, None).await?;
905 streaming_job::ActiveModel {
906 job_id: Set(source_job_id),
907 job_status: Set(JobStatus::Initial),
908 ..Default::default()
909 }
910 .update(&txn)
911 .await?;
912
913 let sink_job_id = CatalogController::create_object(
914 &txn,
915 ObjectType::Sink,
916 TEST_OWNER_ID,
917 Some(TEST_SCHEMA_ID.as_object_id()),
918 )
919 .await?
920 .oid
921 .as_job_id();
922 Sink::insert(sink::ActiveModel::from(PbSink {
923 id: sink_job_id.as_sink_id(),
924 schema_id: TEST_SCHEMA_ID,
925 database_id: TEST_DATABASE_ID,
926 name: "creating_sink".to_owned(),
927 owner: TEST_OWNER_ID as _,
928 sink_type: PbSinkType::AppendOnly as i32,
929 ..Default::default()
930 }))
931 .exec(&txn)
932 .await?;
933 insert_test_streaming_job_model(&txn, sink_job_id, None).await?;
934 streaming_job::ActiveModel {
935 job_id: Set(sink_job_id),
936 job_status: Set(JobStatus::Initial),
937 ..Default::default()
938 }
939 .update(&txn)
940 .await?;
941
942 txn.commit().await?;
943
944 let (catalog, _) = inner.snapshot().await?;
945 assert!(!catalog.2.iter().any(|table| table.id == table_id));
946 assert!(!catalog.2.iter().any(|table| table.id == internal_table_id));
947 assert!(!catalog.2.iter().any(|table| table.id == index_table_id));
948 assert!(
949 !catalog
950 .3
951 .iter()
952 .any(|source| source.id == associated_source_id)
953 );
954 assert!(
955 !catalog
956 .3
957 .iter()
958 .any(|source| source.id == source_job_id.as_shared_source_id())
959 );
960 assert!(
961 !catalog
962 .6
963 .iter()
964 .any(|index| index.id == index_job_id.as_index_id())
965 );
966 assert!(
967 !catalog
968 .4
969 .iter()
970 .any(|sink| sink.id == sink_job_id.as_sink_id())
971 );
972
973 drop(inner);
974
975 let downstreams = FragmentDownstreamRelation::new();
976 let mut add_notifications = vec![];
977 for creating_job_id in [job_id, index_job_id, source_job_id, sink_job_id] {
978 mgr.post_collect_job_fragments(creating_job_id, &downstreams, None, None, None, true)
979 .await?;
980 let notification = notification_rx
981 .recv()
982 .await
983 .expect("frontend should receive a creating notification")
984 .expect("creating notification should be valid");
985 assert_eq!(notification.operation(), NotificationOperation::Add);
986 let object_group = match notification.info {
987 Some(NotificationInfo::ObjectGroup(object_group)) => object_group,
988 other => panic!("unexpected notification: {other:?}"),
989 };
990 add_notifications.push(object_group);
991 }
992
993 assert!(add_notifications[0].objects.iter().any(|object| matches!(
994 &object.object_info,
995 Some(PbObjectInfo::Table(table)) if table.id == table_id
996 )));
997 assert!(add_notifications[0].objects.iter().any(|object| matches!(
998 &object.object_info,
999 Some(PbObjectInfo::Table(table)) if table.id == internal_table_id
1000 )));
1001 assert!(add_notifications[0].objects.iter().any(|object| matches!(
1002 &object.object_info,
1003 Some(PbObjectInfo::Source(source)) if source.id == associated_source_id
1004 )));
1005 assert!(add_notifications[1].objects.iter().any(|object| matches!(
1006 &object.object_info,
1007 Some(PbObjectInfo::Table(table)) if table.id == index_table_id
1008 )));
1009 assert!(add_notifications[1].objects.iter().any(|object| matches!(
1010 &object.object_info,
1011 Some(PbObjectInfo::Index(index)) if index.id == index_job_id.as_index_id()
1012 )));
1013 assert!(add_notifications[2].objects.iter().any(|object| matches!(
1014 &object.object_info,
1015 Some(PbObjectInfo::Source(source)) if source.id == source_job_id.as_shared_source_id()
1016 )));
1017 assert!(add_notifications[3].objects.iter().any(|object| matches!(
1018 &object.object_info,
1019 Some(PbObjectInfo::Sink(sink)) if sink.id == sink_job_id.as_sink_id()
1020 )));
1021
1022 let inner = mgr.inner.write().await;
1023 let (catalog, _) = inner.snapshot().await?;
1024 assert!(catalog.2.iter().any(|table| table.id == table_id));
1025 assert!(catalog.2.iter().any(|table| table.id == internal_table_id));
1026 assert!(catalog.2.iter().any(|table| table.id == index_table_id));
1027 assert!(
1028 catalog
1029 .3
1030 .iter()
1031 .any(|source| source.id == source_job_id.as_shared_source_id())
1032 );
1033 assert!(
1034 catalog
1035 .6
1036 .iter()
1037 .any(|index| index.id == index_job_id.as_index_id())
1038 );
1039 assert!(
1040 catalog
1041 .4
1042 .iter()
1043 .any(|sink| sink.id == sink_job_id.as_sink_id())
1044 );
1045
1046 let txn = inner.db.begin().await?;
1047 let (operation, _, _, _) = mgr.finish_streaming_job_inner(&txn, job_id).await?;
1048 assert_eq!(operation, NotificationOperation::Update);
1049 let (operation, _, _, _) = mgr.finish_streaming_job_inner(&txn, index_job_id).await?;
1050 assert_eq!(operation, NotificationOperation::Update);
1051 let (operation, _, _, _) = mgr.finish_streaming_job_inner(&txn, source_job_id).await?;
1052 assert_eq!(operation, NotificationOperation::Update);
1053 let (operation, _, _, _) = mgr.finish_streaming_job_inner(&txn, sink_job_id).await?;
1054 assert_eq!(operation, NotificationOperation::Update);
1055 txn.commit().await?;
1056
1057 Ok(())
1058 }
1059
1060 #[tokio::test]
1061 async fn test_alter_streaming_job_cache_refill_policy_notifies_hummock() -> MetaResult<()> {
1062 let env = MetaSrvEnv::for_test().await;
1063 let (tx, mut rx) = mpsc::unbounded_channel();
1064 env.notification_manager().insert_sender(
1065 SubscribeType::Hummock,
1066 WorkerKey(HostAddress {
1067 host: "localhost".to_owned(),
1068 port: 1234,
1069 }),
1070 tx,
1071 );
1072 let mgr = CatalogController::new(env).await?;
1073
1074 let inner = mgr.inner.write().await;
1075 let txn = inner.db.begin().await?;
1076 let (_job, Some(result_table_id), internal_table_id) =
1077 insert_test_streaming_job(&txn, "mv_cache_refill", true, None).await?
1078 else {
1079 unreachable!()
1080 };
1081 insert_test_fragment(
1082 &txn,
1083 FragmentId::new(200),
1084 result_table_id.as_job_id(),
1085 TableIdArray(vec![result_table_id, internal_table_id]),
1086 )
1087 .await?;
1088 txn.commit().await?;
1089 drop(inner);
1090
1091 mgr.alter_streaming_job_config(
1092 result_table_id.as_job_id(),
1093 HashMap::from([(
1094 STREAMING_CACHE_REFILL_POLICY_CONFIG_PATH.to_owned(),
1095 "\"both\"".to_owned(),
1096 )]),
1097 vec![],
1098 )
1099 .await?;
1100
1101 let response = rx
1102 .recv()
1103 .await
1104 .expect("should receive hummock notification")
1105 .expect("notification should be ok");
1106 assert_eq!(response.operation(), NotificationOperation::Update);
1107 let info = response.info;
1108 let Some(NotificationInfo::TableRefillRuntimeConfig(config)) = info else {
1109 panic!("unexpected notification: {:?}", info);
1110 };
1111 assert!(config.serving_table_vnode_mappings.is_none());
1112 let policies = config
1113 .table_cache_refill_policies
1114 .expect("policy snapshot should be present");
1115 assert_eq!(
1116 policies
1117 .table_policies
1118 .into_iter()
1119 .map(|policy| (policy.table_id, policy.policy))
1120 .collect::<HashMap<_, _>>(),
1121 HashMap::from([(
1122 result_table_id.as_raw_id(),
1123 CacheRefillPolicy::Both.to_protobuf() as i32,
1124 )])
1125 );
1126 assert_eq!(
1127 policies
1128 .internal_table_policies
1129 .into_iter()
1130 .map(|policy| (policy.table_id, policy.policy))
1131 .collect::<HashMap<_, _>>(),
1132 HashMap::from([(
1133 internal_table_id.as_raw_id(),
1134 CacheRefillPolicy::Both.to_protobuf() as i32,
1135 )])
1136 );
1137
1138 Ok(())
1139 }
1140
1141 #[tokio::test]
1142 async fn test_prepare_streaming_job_cache_refill_policy_notifies_hummock() -> MetaResult<()> {
1143 let env = MetaSrvEnv::for_test().await;
1144 let (tx, mut rx) = mpsc::unbounded_channel();
1145 env.notification_manager().insert_sender(
1146 SubscribeType::Hummock,
1147 WorkerKey(HostAddress {
1148 host: "localhost".to_owned(),
1149 port: 1234,
1150 }),
1151 tx,
1152 );
1153 let (local_notification_tx, mut local_notification_rx) = mpsc::unbounded_channel();
1154 env.notification_manager()
1155 .insert_local_sender(local_notification_tx);
1156 let mgr = CatalogController::new(env).await?;
1157
1158 let inner = mgr.inner.write().await;
1159 let txn = inner.db.begin().await?;
1160 let (job_id, Some(result_table_id), internal_table_id) = insert_test_streaming_job(
1161 &txn,
1162 "mv_initial_cache_refill",
1163 true,
1164 Some(CacheRefillPolicy::Both),
1165 )
1166 .await?
1167 else {
1168 unreachable!()
1169 };
1170 let (unprepared_job_id, Some(unprepared_result_table_id), unprepared_internal_table_id) =
1171 insert_test_streaming_job(
1172 &txn,
1173 "mv_unprepared_cache_refill",
1174 true,
1175 Some(CacheRefillPolicy::Serving),
1176 )
1177 .await?
1178 else {
1179 unreachable!()
1180 };
1181 for job_id in [job_id, unprepared_job_id] {
1182 streaming_job::ActiveModel {
1183 job_id: Set(job_id),
1184 job_status: Set(JobStatus::Initial),
1185 ..Default::default()
1186 }
1187 .update(&txn)
1188 .await?;
1189 }
1190 txn.commit().await?;
1191 drop(inner);
1192
1193 let fragments = [Fragment {
1194 fragment_id: FragmentId::new(300),
1195 fragment_type_mask: FragmentTypeMask::default(),
1196 distribution_type: PbFragmentDistributionType::Hash,
1197 state_table_ids: vec![],
1198 maybe_vnode_count: Some(1),
1199 nodes: PbStreamNode::default(),
1200 }];
1201 mgr.prepare_streaming_job(
1202 job_id,
1203 || fragments.iter(),
1204 &FragmentDownstreamRelation::default(),
1205 true,
1206 None,
1207 None,
1208 )
1209 .await?;
1210
1211 let local_notification = local_notification_rx
1212 .try_recv()
1213 .expect("should receive serving fragment mapping notification");
1214 let LocalNotification::ServingFragmentMappingsUpsert(fragment_ids) = local_notification
1215 else {
1216 panic!(
1217 "unexpected local notification before hummock notification: {:?}",
1218 local_notification
1219 );
1220 };
1221 assert_eq!(fragment_ids, vec![FragmentId::new(300).as_raw_id()]);
1222
1223 let response = rx
1224 .recv()
1225 .await
1226 .expect("should receive hummock notification")
1227 .expect("notification should be ok");
1228 assert_eq!(response.operation(), NotificationOperation::Update);
1229 let info = response.info;
1230 let Some(NotificationInfo::TableRefillRuntimeConfig(config)) = info else {
1231 panic!("unexpected notification: {:?}", info);
1232 };
1233 assert!(config.serving_table_vnode_mappings.is_none());
1234 let policies = config
1235 .table_cache_refill_policies
1236 .expect("policy snapshot should be present");
1237 let table_policies = policies
1238 .table_policies
1239 .into_iter()
1240 .map(|policy| (policy.table_id, policy.policy))
1241 .collect::<HashMap<_, _>>();
1242 assert_eq!(
1243 table_policies,
1244 HashMap::from([(
1245 result_table_id.as_raw_id(),
1246 CacheRefillPolicy::Both.to_protobuf() as i32,
1247 )])
1248 );
1249 assert!(!table_policies.contains_key(&unprepared_result_table_id.as_raw_id()));
1250 let internal_table_policies = policies
1251 .internal_table_policies
1252 .into_iter()
1253 .map(|policy| (policy.table_id, policy.policy))
1254 .collect::<HashMap<_, _>>();
1255 assert_eq!(
1256 internal_table_policies,
1257 HashMap::from([(
1258 internal_table_id.as_raw_id(),
1259 CacheRefillPolicy::Both.to_protobuf() as i32,
1260 )])
1261 );
1262 assert!(!internal_table_policies.contains_key(&unprepared_internal_table_id.as_raw_id()));
1263
1264 Ok(())
1265 }
1266
1267 async fn insert_dirty_creating_job_with_fragment(
1268 mgr: &CatalogController,
1269 fragment_id: FragmentId,
1270 vnode_count: i32,
1271 fragment_type_mask: FragmentTypeMask,
1272 ) -> MetaResult<(JobId, TableId)> {
1273 let inner = mgr.inner.write().await;
1274 let txn = inner.db.begin().await?;
1275 let job_obj = CatalogController::create_object(
1276 &txn,
1277 ObjectType::Table,
1278 TEST_OWNER_ID,
1279 Some(TEST_SCHEMA_ID.as_object_id()),
1280 )
1281 .await?;
1282 let job_id = job_obj.oid.as_job_id();
1283 let table_id = job_id.as_mv_table_id();
1284 insert_test_table(
1285 &txn,
1286 table_id,
1287 "mv_dirty_serving_mapping",
1288 TableType::MaterializedView,
1289 None,
1290 "CREATE MATERIALIZED VIEW mv_dirty_serving_mapping AS SELECT 1",
1291 )
1292 .await?;
1293 table::ActiveModel {
1294 table_id: Set(table_id),
1295 engine: Set(Some(table::Engine::Hummock)),
1296 ..Default::default()
1297 }
1298 .update(&txn)
1299 .await?;
1300 streaming_job::ActiveModel {
1301 job_id: Set(job_id),
1302 job_status: Set(JobStatus::Creating),
1303 create_type: Set(CreateType::Foreground),
1304 timezone: Set(None),
1305 config_override: Set(None),
1306 adaptive_parallelism_strategy: Set(None),
1307 parallelism: Set(StreamingParallelism::Adaptive),
1308 backfill_parallelism: Set(None),
1309 backfill_adaptive_parallelism_strategy: Set(None),
1310 backfill_orders: Set(None),
1311 max_parallelism: Set(1),
1312 specific_resource_group: Set(None),
1313 is_serverless_backfill: Set(false),
1314 refresh_interval_sec: Set(None),
1315 }
1316 .insert(&txn)
1317 .await?;
1318 fragment::ActiveModel {
1319 fragment_id: Set(fragment_id),
1320 job_id: Set(job_id),
1321 fragment_type_mask: Set(fragment_type_mask.into()),
1322 distribution_type: Set(DistributionType::Hash),
1323 stream_node: Set(StreamNode::default()),
1324 state_table_ids: Set(Vec::<TableId>::new().into()),
1325 upstream_fragment_id: Set(Vec::<i32>::new().into()),
1326 vnode_count: Set(vnode_count),
1327 parallelism: Set(None),
1328 }
1329 .insert(&txn)
1330 .await?;
1331 txn.commit().await?;
1332 drop(inner);
1333
1334 Ok((job_id, table_id))
1335 }
1336
1337 #[tokio::test]
1338 async fn test_dirty_cleanup_reconcile_removes_stale_serving_vnode_mapping() -> MetaResult<()> {
1339 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
1340 let fragment_id = FragmentId::new(42);
1341 insert_dirty_creating_job_with_fragment(
1342 &mgr,
1343 fragment_id,
1344 VirtualNode::COUNT_FOR_TEST as i32,
1345 FragmentTypeMask::from(FragmentTypeFlag::Values as u32),
1346 )
1347 .await?;
1348
1349 let worker = WorkerNode {
1350 id: WorkerId::new(1),
1351 r#type: WorkerType::ComputeNode.into(),
1352 host: Some(HostAddress {
1353 host: "localhost".to_owned(),
1354 port: 1,
1355 }),
1356 state: worker_node::State::Running as i32,
1357 property: Some(worker_node::Property {
1358 is_serving: true,
1359 parallelism: 1,
1360 ..Default::default()
1361 }),
1362 ..Default::default()
1363 };
1364 let serving_vnode_mapping = ServingVnodeMapping::default();
1365 let initial_snapshot = mgr.fragment_serving_infos().await?;
1366 serving_vnode_mapping.upsert(&initial_snapshot, std::slice::from_ref(&worker), None);
1367 assert!(serving_vnode_mapping.all().contains_key(&fragment_id));
1368
1369 mgr.clean_dirty_creating_jobs(Some(TEST_DATABASE_ID))
1370 .await?;
1371 let current_snapshot = mgr.fragment_serving_infos().await?;
1372 assert!(!current_snapshot.contains_key(&fragment_id));
1373
1374 serving_vnode_mapping.reconcile(¤t_snapshot, &[worker], None);
1375 assert!(!serving_vnode_mapping.all().contains_key(&fragment_id));
1376
1377 Ok(())
1378 }
1379
1380 #[tokio::test]
1381 async fn test_clean_dirty_creating_jobs_keeps_job_without_values_fragment() -> MetaResult<()> {
1382 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
1383 let fragment_id = FragmentId::new(43);
1384 let (job_id, table_id) = insert_dirty_creating_job_with_fragment(
1385 &mgr,
1386 fragment_id,
1387 1,
1388 FragmentTypeMask::empty(),
1389 )
1390 .await?;
1391
1392 let cleaned = mgr
1393 .clean_dirty_creating_jobs(Some(TEST_DATABASE_ID))
1394 .await?;
1395 assert!(cleaned.streaming_job_ids.is_empty());
1396
1397 let inner = mgr.inner.read().await;
1398 assert!(Object::find_by_id(job_id).one(&inner.db).await?.is_some());
1399 assert!(
1400 StreamingJob::find_by_id(job_id)
1401 .one(&inner.db)
1402 .await?
1403 .is_some()
1404 );
1405 assert!(Table::find_by_id(table_id).one(&inner.db).await?.is_some());
1406
1407 Ok(())
1408 }
1409
1410 #[tokio::test]
1411 async fn test_clean_dirty_creating_jobs_cleans_foreground_job_in_legacy_mode() -> MetaResult<()>
1412 {
1413 let mut opts = MetaOpts::test(false);
1414 opts.clean_all_foreground_jobs_on_recovery = true;
1415 let mgr = CatalogController::new(MetaSrvEnv::for_test_opts(opts, |_| ()).await).await?;
1416 let (job_id, table_id) = insert_dirty_creating_job_with_fragment(
1417 &mgr,
1418 FragmentId::new(44),
1419 1,
1420 FragmentTypeMask::empty(),
1421 )
1422 .await?;
1423
1424 let cleaned = mgr
1425 .clean_dirty_creating_jobs(Some(TEST_DATABASE_ID))
1426 .await?;
1427 assert_eq!(cleaned.streaming_job_ids, vec![job_id]);
1428
1429 let db = &mgr.inner.read().await.db;
1430 assert!(Object::find_by_id(job_id).one(db).await?.is_none());
1431 assert!(StreamingJob::find_by_id(job_id).one(db).await?.is_none());
1432 assert!(Table::find_by_id(table_id).one(db).await?.is_none());
1433
1434 Ok(())
1435 }
1436
1437 #[tokio::test]
1438 async fn test_database_func() -> MetaResult<()> {
1439 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
1440 let pb_database = PbDatabase {
1441 name: "db1".to_owned(),
1442 owner: TEST_OWNER_ID as _,
1443 ..Default::default()
1444 };
1445 mgr.create_database(pb_database).await?;
1446
1447 let database_id: DatabaseId = Database::find()
1448 .select_only()
1449 .column(database::Column::DatabaseId)
1450 .filter(database::Column::Name.eq("db1"))
1451 .into_tuple()
1452 .one(&mgr.inner.read().await.db)
1453 .await?
1454 .unwrap();
1455
1456 mgr.alter_name(ObjectType::Database, database_id, "db2")
1457 .await?;
1458 let database = Database::find_by_id(database_id)
1459 .one(&mgr.inner.read().await.db)
1460 .await?
1461 .unwrap();
1462 assert_eq!(database.name, "db2");
1463
1464 let schema_id: SchemaId = Schema::find()
1465 .inner_join(Object)
1466 .select_only()
1467 .column(schema::Column::SchemaId)
1468 .filter(object::Column::DatabaseId.eq(database_id))
1469 .into_tuple()
1470 .one(&mgr.inner.read().await.db)
1471 .await?
1472 .unwrap();
1473 mgr.create_view(
1474 PbView {
1475 schema_id,
1476 database_id,
1477 name: "cross_db_upstream".to_owned(),
1478 owner: TEST_OWNER_ID as _,
1479 sql: "CREATE VIEW cross_db_upstream AS SELECT 1".to_owned(),
1480 ..Default::default()
1481 },
1482 HashSet::new(),
1483 )
1484 .await?;
1485 let upstream_id: ViewId = View::find()
1486 .inner_join(Object)
1487 .select_only()
1488 .column(view::Column::ViewId)
1489 .filter(
1490 object::Column::DatabaseId
1491 .eq(database_id)
1492 .and(view::Column::Name.eq("cross_db_upstream")),
1493 )
1494 .into_tuple()
1495 .one(&mgr.inner.read().await.db)
1496 .await?
1497 .unwrap();
1498
1499 let inner = mgr.inner.write().await;
1500 let txn = inner.db.begin().await?;
1501 let (dependent_job_id, Some(dependent_table_id), _) =
1502 insert_test_streaming_job(&txn, "cross_db_dependent", true, None).await?
1503 else {
1504 unreachable!()
1505 };
1506 ObjectDependency::insert(object_dependency::ActiveModel {
1507 oid: Set(upstream_id.as_object_id()),
1508 used_by: Set(dependent_job_id.as_object_id()),
1509 ..Default::default()
1510 })
1511 .exec(&txn)
1512 .await?;
1513 txn.commit().await?;
1514 drop(inner);
1515
1516 assert!(
1517 mgr.drop_object(ObjectType::Database, database_id, DropMode::Cascade)
1518 .await
1519 .is_err()
1520 );
1521 mgr.drop_object(ObjectType::Table, dependent_table_id, DropMode::Cascade)
1522 .await?;
1523 mgr.drop_object(ObjectType::Database, database_id, DropMode::Cascade)
1524 .await?;
1525
1526 Ok(())
1527 }
1528
1529 #[tokio::test]
1530 async fn test_schema_func() -> MetaResult<()> {
1531 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
1532 let pb_schema = PbSchema {
1533 database_id: TEST_DATABASE_ID,
1534 name: "schema1".to_owned(),
1535 owner: TEST_OWNER_ID as _,
1536 ..Default::default()
1537 };
1538 mgr.create_schema(pb_schema.clone()).await?;
1539 assert!(mgr.create_schema(pb_schema).await.is_err());
1540
1541 let schema_id: SchemaId = Schema::find()
1542 .select_only()
1543 .column(schema::Column::SchemaId)
1544 .filter(schema::Column::Name.eq("schema1"))
1545 .into_tuple()
1546 .one(&mgr.inner.read().await.db)
1547 .await?
1548 .unwrap();
1549
1550 mgr.alter_name(ObjectType::Schema, schema_id, "schema2")
1551 .await?;
1552 let schema = Schema::find_by_id(schema_id)
1553 .one(&mgr.inner.read().await.db)
1554 .await?
1555 .unwrap();
1556 assert_eq!(schema.name, "schema2");
1557 mgr.drop_object(ObjectType::Schema, schema_id, DropMode::Restrict)
1558 .await?;
1559
1560 Ok(())
1561 }
1562
1563 #[tokio::test]
1564 async fn test_create_view() -> MetaResult<()> {
1565 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
1566 let pb_view = PbView {
1567 schema_id: TEST_SCHEMA_ID,
1568 database_id: TEST_DATABASE_ID,
1569 name: "view".to_owned(),
1570 owner: TEST_OWNER_ID as _,
1571 sql: "CREATE VIEW view AS SELECT 1".to_owned(),
1572 ..Default::default()
1573 };
1574 mgr.create_view(pb_view.clone(), HashSet::new()).await?;
1575 assert!(mgr.create_view(pb_view, HashSet::new()).await.is_err());
1576
1577 let view = View::find().one(&mgr.inner.read().await.db).await?.unwrap();
1578 mgr.drop_object(ObjectType::View, view.view_id, DropMode::Cascade)
1579 .await?;
1580 assert!(
1581 View::find_by_id(view.view_id)
1582 .one(&mgr.inner.read().await.db)
1583 .await?
1584 .is_none()
1585 );
1586
1587 Ok(())
1588 }
1589
1590 #[tokio::test]
1591 async fn test_object_belong_to_cascade() -> MetaResult<()> {
1592 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
1593 mgr.create_schema(PbSchema {
1594 database_id: TEST_DATABASE_ID,
1595 name: "belong_to_target".to_owned(),
1596 owner: TEST_OWNER_ID as _,
1597 ..Default::default()
1598 })
1599 .await?;
1600 let target_schema_id: SchemaId = Schema::find()
1601 .select_only()
1602 .column(schema::Column::SchemaId)
1603 .filter(schema::Column::Name.eq("belong_to_target"))
1604 .into_tuple()
1605 .one(&mgr.inner.read().await.db)
1606 .await?
1607 .unwrap();
1608 let txn = mgr.inner.read().await.db.begin().await?;
1609
1610 let mv_obj = CatalogController::create_object(
1611 &txn,
1612 ObjectType::Table,
1613 TEST_OWNER_ID,
1614 Some(TEST_SCHEMA_ID.as_object_id()),
1615 )
1616 .await?;
1617 assert_eq!(mv_obj.belong_to_oid, Some(TEST_SCHEMA_ID.as_object_id()));
1618 assert_eq!(mv_obj.database_id, Some(TEST_DATABASE_ID));
1619 assert_eq!(mv_obj.schema_id, Some(TEST_SCHEMA_ID));
1620 let job_id = mv_obj.oid.as_job_id();
1621 let mv_table_id = job_id.as_mv_table_id();
1622 insert_test_table(
1623 &txn,
1624 mv_table_id,
1625 "mv_belong_to",
1626 TableType::MaterializedView,
1627 None,
1628 "CREATE MATERIALIZED VIEW mv_belong_to AS SELECT 1",
1629 )
1630 .await?;
1631
1632 let internal_obj = CatalogController::create_object(
1633 &txn,
1634 ObjectType::Table,
1635 TEST_OWNER_ID,
1636 Some(job_id.as_object_id()),
1637 )
1638 .await?;
1639 assert_eq!(internal_obj.belong_to_oid, Some(job_id.as_object_id()));
1640 assert_eq!(internal_obj.database_id, Some(TEST_DATABASE_ID));
1641 assert_eq!(internal_obj.schema_id, Some(TEST_SCHEMA_ID));
1642 let internal_table_id = internal_obj.oid.as_table_id();
1643 insert_test_table(
1644 &txn,
1645 internal_table_id,
1646 "__internal_mv_belong_to",
1647 TableType::Internal,
1648 Some(job_id),
1649 "",
1650 )
1651 .await?;
1652 let nested_obj = CatalogController::create_object(
1653 &txn,
1654 ObjectType::Table,
1655 TEST_OWNER_ID,
1656 Some(internal_table_id.as_object_id()),
1657 )
1658 .await?;
1659 txn.commit().await?;
1660
1661 assert!(
1662 mgr.alter_schema(ObjectType::Sink, job_id.as_object_id(), target_schema_id,)
1663 .await
1664 .is_err()
1665 );
1666 mgr.alter_schema(ObjectType::Table, job_id.as_object_id(), target_schema_id)
1667 .await?;
1668
1669 let db = &mgr.inner.read().await.db;
1670 let belonging_object_ids = get_belong_objects(db, job_id.as_object_id())
1671 .await?
1672 .into_iter()
1673 .map(|object| object.oid)
1674 .collect::<HashSet<_>>();
1675 assert_eq!(
1676 belonging_object_ids,
1677 HashSet::from([internal_table_id.as_object_id(), nested_obj.oid])
1678 );
1679 let moved_objects = Object::find()
1680 .filter(object::Column::Oid.is_in([
1681 job_id.as_object_id(),
1682 internal_table_id.as_object_id(),
1683 nested_obj.oid,
1684 ]))
1685 .all(db)
1686 .await?;
1687 assert!(
1688 moved_objects
1689 .iter()
1690 .all(|object| object.schema_id == Some(target_schema_id))
1691 );
1692 assert_eq!(
1693 Object::find_by_id(internal_table_id)
1694 .one(db)
1695 .await?
1696 .unwrap()
1697 .belong_to_oid,
1698 Some(job_id.as_object_id())
1699 );
1700 assert_eq!(
1701 Object::find_by_id(nested_obj.oid)
1702 .one(db)
1703 .await?
1704 .unwrap()
1705 .belong_to_oid,
1706 Some(internal_table_id.as_object_id())
1707 );
1708
1709 Object::delete_by_id(job_id).exec(db).await?;
1710
1711 assert!(Object::find_by_id(job_id).one(db).await?.is_none());
1712 assert!(
1713 Object::find_by_id(internal_table_id)
1714 .one(db)
1715 .await?
1716 .is_none()
1717 );
1718 assert!(Table::find_by_id(mv_table_id).one(db).await?.is_none());
1719 assert!(
1720 Table::find_by_id(internal_table_id)
1721 .one(db)
1722 .await?
1723 .is_none()
1724 );
1725 assert!(Object::find_by_id(nested_obj.oid).one(db).await?.is_none());
1726
1727 Ok(())
1728 }
1729
1730 #[tokio::test]
1731 async fn test_alter_internal_table_schema_rejected() -> MetaResult<()> {
1732 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
1733 mgr.create_schema(PbSchema {
1734 database_id: TEST_DATABASE_ID,
1735 name: "internal_table_alter_target".to_owned(),
1736 owner: TEST_OWNER_ID as _,
1737 ..Default::default()
1738 })
1739 .await?;
1740 let target_schema_id: SchemaId = Schema::find()
1741 .select_only()
1742 .column(schema::Column::SchemaId)
1743 .filter(schema::Column::Name.eq("internal_table_alter_target"))
1744 .into_tuple()
1745 .one(&mgr.inner.read().await.db)
1746 .await?
1747 .unwrap();
1748
1749 let txn = mgr.inner.read().await.db.begin().await?;
1750 let parent_obj = CatalogController::create_object(
1751 &txn,
1752 ObjectType::Table,
1753 TEST_OWNER_ID,
1754 Some(TEST_SCHEMA_ID.as_object_id()),
1755 )
1756 .await?;
1757 let parent_job_id = parent_obj.oid.as_job_id();
1758 insert_test_table(
1759 &txn,
1760 parent_job_id.as_mv_table_id(),
1761 "internal_table_parent",
1762 TableType::MaterializedView,
1763 None,
1764 "",
1765 )
1766 .await?;
1767 let internal_obj = CatalogController::create_object(
1768 &txn,
1769 ObjectType::Table,
1770 TEST_OWNER_ID,
1771 Some(parent_job_id.as_object_id()),
1772 )
1773 .await?;
1774 let internal_table_id = internal_obj.oid.as_table_id();
1775 insert_test_table(
1776 &txn,
1777 internal_table_id,
1778 "__internal_table_alter_target",
1779 TableType::Internal,
1780 Some(parent_job_id),
1781 "",
1782 )
1783 .await?;
1784 txn.commit().await?;
1785
1786 for new_schema in [TEST_SCHEMA_ID, target_schema_id] {
1787 assert!(
1788 mgr.alter_schema(
1789 ObjectType::Table,
1790 internal_table_id.as_object_id(),
1791 new_schema,
1792 )
1793 .await
1794 .is_err()
1795 );
1796 }
1797
1798 let internal_obj = Object::find_by_id(internal_table_id)
1799 .one(&mgr.inner.read().await.db)
1800 .await?
1801 .unwrap();
1802 assert_eq!(internal_obj.schema_id, Some(TEST_SCHEMA_ID));
1803 assert_eq!(
1804 internal_obj.belong_to_oid,
1805 Some(parent_job_id.as_object_id())
1806 );
1807
1808 Ok(())
1809 }
1810
1811 #[tokio::test]
1812 async fn test_alter_table_schema_moves_indexes_but_not_subscriptions() -> MetaResult<()> {
1813 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
1814 mgr.create_schema(PbSchema {
1815 database_id: TEST_DATABASE_ID,
1816 name: "alter_table_target".to_owned(),
1817 owner: TEST_OWNER_ID as _,
1818 ..Default::default()
1819 })
1820 .await?;
1821 let target_schema_id: SchemaId = Schema::find()
1822 .select_only()
1823 .column(schema::Column::SchemaId)
1824 .filter(schema::Column::Name.eq("alter_table_target"))
1825 .into_tuple()
1826 .one(&mgr.inner.read().await.db)
1827 .await?
1828 .unwrap();
1829
1830 let txn = mgr.inner.read().await.db.begin().await?;
1831 let table_obj = CatalogController::create_object(
1832 &txn,
1833 ObjectType::Table,
1834 TEST_OWNER_ID,
1835 Some(TEST_SCHEMA_ID.as_object_id()),
1836 )
1837 .await?;
1838 let table_id = table_obj.oid.as_table_id();
1839 insert_test_table(
1840 &txn,
1841 table_id,
1842 "mv_with_index_and_subscription",
1843 TableType::MaterializedView,
1844 None,
1845 "CREATE MATERIALIZED VIEW mv_with_index_and_subscription AS SELECT 1",
1846 )
1847 .await?;
1848
1849 let index_obj = CatalogController::create_object(
1850 &txn,
1851 ObjectType::Index,
1852 TEST_OWNER_ID,
1853 Some(TEST_SCHEMA_ID.as_object_id()),
1854 )
1855 .await?;
1856 let index_id = index_obj.oid.as_index_id();
1857 let index_table_id = index_id.as_object_id().as_table_id();
1858 insert_test_table(
1859 &txn,
1860 index_table_id,
1861 "idx_mv_with_index_and_subscription_table",
1862 TableType::Index,
1863 None,
1864 "",
1865 )
1866 .await?;
1867 index::ActiveModel {
1868 index_id: Set(index_id),
1869 name: Set("idx_mv_with_index_and_subscription".to_owned()),
1870 index_table_id: Set(index_table_id),
1871 primary_table_id: Set(table_id),
1872 index_items: Set(Vec::<risingwave_pb::expr::ExprNode>::new().into()),
1873 index_column_properties: Set(None),
1874 index_columns_len: Set(0),
1875 }
1876 .insert(&txn)
1877 .await?;
1878
1879 let index_internal_obj = CatalogController::create_object(
1880 &txn,
1881 ObjectType::Table,
1882 TEST_OWNER_ID,
1883 Some(index_id.as_object_id()),
1884 )
1885 .await?;
1886 let index_internal_table_id = index_internal_obj.oid.as_table_id();
1887 insert_test_table(
1888 &txn,
1889 index_internal_table_id,
1890 "__internal_idx_mv_with_index_and_subscription",
1891 TableType::Internal,
1892 Some(index_id.as_job_id()),
1893 "",
1894 )
1895 .await?;
1896 txn.commit().await?;
1897
1898 let mut subscription = PbSubscription {
1899 name: "subscription_in_original_schema".to_owned(),
1900 definition: "CREATE SUBSCRIPTION subscription_in_original_schema FROM mv_with_index_and_subscription".to_owned(),
1901 retention_seconds: 86400,
1902 database_id: TEST_DATABASE_ID,
1903 schema_id: TEST_SCHEMA_ID,
1904 dependent_table_id: table_id,
1905 owner: TEST_OWNER_ID as _,
1906 subscription_state: SubscriptionState::Created as _,
1907 ..Default::default()
1908 };
1909 mgr.create_subscription_catalog(&mut subscription).await?;
1910
1911 {
1912 let inner = mgr.inner.read().await;
1913 assert_eq!(
1914 Object::find_by_id(index_id)
1915 .one(&inner.db)
1916 .await?
1917 .unwrap()
1918 .belong_to_oid,
1919 Some(TEST_SCHEMA_ID.as_object_id())
1920 );
1921 assert_eq!(
1922 Object::find_by_id(subscription.id)
1923 .one(&inner.db)
1924 .await?
1925 .unwrap()
1926 .belong_to_oid,
1927 Some(TEST_SCHEMA_ID.as_object_id())
1928 );
1929 }
1930
1931 mgr.alter_schema(ObjectType::Table, table_id.as_object_id(), target_schema_id)
1932 .await?;
1933
1934 let db = &mgr.inner.read().await.db;
1935 for object_id in [table_id.as_object_id(), index_id.as_object_id()] {
1936 let object = Object::find_by_id(object_id).one(db).await?.unwrap();
1937 assert_eq!(object.schema_id, Some(target_schema_id));
1938 assert_eq!(object.belong_to_oid, Some(target_schema_id.as_object_id()));
1939 }
1940 let index_internal_object = Object::find_by_id(index_internal_table_id)
1941 .one(db)
1942 .await?
1943 .unwrap();
1944 assert_eq!(index_internal_object.schema_id, Some(target_schema_id));
1945 assert_eq!(
1946 index_internal_object.belong_to_oid,
1947 Some(index_id.as_object_id())
1948 );
1949
1950 let subscription_object = Object::find_by_id(subscription.id).one(db).await?.unwrap();
1951 assert_eq!(subscription_object.schema_id, Some(TEST_SCHEMA_ID));
1952 assert_eq!(
1953 subscription_object.belong_to_oid,
1954 Some(TEST_SCHEMA_ID.as_object_id())
1955 );
1956
1957 Ok(())
1958 }
1959
1960 #[tokio::test]
1961 async fn test_create_function() -> MetaResult<()> {
1962 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
1963 let test_data_type = risingwave_pb::data::DataType {
1964 type_name: risingwave_pb::data::data_type::TypeName::Int32 as _,
1965 ..Default::default()
1966 };
1967 let arg_types = vec![test_data_type.clone()];
1968 let pb_function = PbFunction {
1969 schema_id: TEST_SCHEMA_ID,
1970 database_id: TEST_DATABASE_ID,
1971 name: "test_function".to_owned(),
1972 owner: TEST_OWNER_ID as _,
1973 arg_types,
1974 return_type: Some(test_data_type.clone()),
1975 language: "python".to_owned(),
1976 kind: Some(risingwave_pb::catalog::function::Kind::Scalar(
1977 Default::default(),
1978 )),
1979 ..Default::default()
1980 };
1981 mgr.create_function(pb_function.clone()).await?;
1982 assert!(mgr.create_function(pb_function).await.is_err());
1983
1984 let function = Function::find()
1985 .inner_join(Object)
1986 .filter(
1987 object::Column::DatabaseId
1988 .eq(TEST_DATABASE_ID)
1989 .and(object::Column::SchemaId.eq(TEST_SCHEMA_ID))
1990 .add(function::Column::Name.eq("test_function")),
1991 )
1992 .one(&mgr.inner.read().await.db)
1993 .await?
1994 .unwrap();
1995 assert_eq!(function.return_type.to_protobuf(), test_data_type);
1996 assert_eq!(function.arg_types.to_protobuf().len(), 1);
1997 assert_eq!(function.language, "python");
1998
1999 mgr.create_schema(PbSchema {
2000 database_id: TEST_DATABASE_ID,
2001 name: "function_target".to_owned(),
2002 owner: TEST_OWNER_ID as _,
2003 ..Default::default()
2004 })
2005 .await?;
2006 let target_schema_id: SchemaId = Schema::find()
2007 .select_only()
2008 .column(schema::Column::SchemaId)
2009 .filter(schema::Column::Name.eq("function_target"))
2010 .into_tuple()
2011 .one(&mgr.inner.read().await.db)
2012 .await?
2013 .unwrap();
2014 mgr.alter_schema(
2015 ObjectType::Function,
2016 function.function_id.as_object_id(),
2017 target_schema_id,
2018 )
2019 .await?;
2020 assert_eq!(
2021 Object::find_by_id(function.function_id)
2022 .one(&mgr.inner.read().await.db)
2023 .await?
2024 .unwrap()
2025 .schema_id,
2026 Some(target_schema_id)
2027 );
2028
2029 mgr.drop_object(
2030 ObjectType::Function,
2031 function.function_id,
2032 DropMode::Restrict,
2033 )
2034 .await?;
2035 assert!(
2036 Object::find_by_id(function.function_id)
2037 .one(&mgr.inner.read().await.db)
2038 .await?
2039 .is_none()
2040 );
2041
2042 Ok(())
2043 }
2044
2045 #[tokio::test]
2046 async fn test_alter_relation_rename() -> MetaResult<()> {
2047 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
2048 let pb_source = PbSource {
2049 schema_id: TEST_SCHEMA_ID,
2050 database_id: TEST_DATABASE_ID,
2051 name: "s1".to_owned(),
2052 owner: TEST_OWNER_ID as _,
2053 definition: r#"CREATE SOURCE s1 (v1 int) with (
2054 connector = 'kafka',
2055 topic = 'kafka_alter',
2056 properties.bootstrap.server = 'message_queue:29092',
2057 scan.startup.mode = 'earliest'
2058) FORMAT PLAIN ENCODE JSON"#
2059 .to_owned(),
2060 info: Some(StreamSourceInfo {
2061 ..Default::default()
2062 }),
2063 ..Default::default()
2064 };
2065 mgr.create_source(pb_source, None).await?;
2066 let source_id: SourceId = Source::find()
2067 .select_only()
2068 .column(source::Column::SourceId)
2069 .filter(source::Column::Name.eq("s1"))
2070 .into_tuple()
2071 .one(&mgr.inner.read().await.db)
2072 .await?
2073 .unwrap();
2074
2075 let pb_view = PbView {
2076 schema_id: TEST_SCHEMA_ID,
2077 database_id: TEST_DATABASE_ID,
2078 name: "view_1".to_owned(),
2079 owner: TEST_OWNER_ID as _,
2080 sql: "CREATE VIEW view_1 AS SELECT v1 FROM s1".to_owned(),
2081 ..Default::default()
2082 };
2083 mgr.create_view(pb_view, HashSet::from([source_id.as_object_id()]))
2084 .await?;
2085 let view_id: ViewId = View::find()
2086 .select_only()
2087 .column(view::Column::ViewId)
2088 .filter(view::Column::Name.eq("view_1"))
2089 .into_tuple()
2090 .one(&mgr.inner.read().await.db)
2091 .await?
2092 .unwrap();
2093
2094 mgr.alter_name(ObjectType::Source, source_id, "s2").await?;
2095 let source = Source::find_by_id(source_id)
2096 .one(&mgr.inner.read().await.db)
2097 .await?
2098 .unwrap();
2099 assert_eq!(source.name, "s2");
2100 assert_eq!(
2101 source.definition,
2102 "CREATE SOURCE s2 (v1 INT) WITH (\
2103 connector = 'kafka', \
2104 topic = 'kafka_alter', \
2105 properties.bootstrap.server = 'message_queue:29092', \
2106 scan.startup.mode = 'earliest'\
2107) FORMAT PLAIN ENCODE JSON"
2108 );
2109
2110 let view = View::find_by_id(view_id)
2111 .one(&mgr.inner.read().await.db)
2112 .await?
2113 .unwrap();
2114 assert_eq!(
2115 view.definition,
2116 "CREATE VIEW view_1 AS SELECT v1 FROM s2 AS s1"
2117 );
2118
2119 mgr.drop_object(ObjectType::Source, source_id, DropMode::Cascade)
2120 .await?;
2121 assert!(
2122 View::find_by_id(view_id)
2123 .one(&mgr.inner.read().await.db)
2124 .await?
2125 .is_none()
2126 );
2127
2128 Ok(())
2129 }
2130
2131 #[tokio::test]
2132 async fn test_cancel_creating_table_deletes_associated_source() -> MetaResult<()> {
2133 let env = MetaSrvEnv::for_test().await;
2134 let (tx, mut notification_rx) = mpsc::unbounded_channel();
2135 env.notification_manager().insert_sender(
2136 SubscribeType::Frontend,
2137 WorkerKey(HostAddress {
2138 host: "localhost".to_owned(),
2139 port: 1234,
2140 }),
2141 tx,
2142 );
2143 let mgr = CatalogController::new(env).await?;
2144
2145 let mut inner = mgr.inner.write().await;
2146 let txn = inner.db.begin().await?;
2147 let obj = CatalogController::create_object(
2148 &txn,
2149 ObjectType::Table,
2150 TEST_OWNER_ID,
2151 Some(TEST_SCHEMA_ID.as_object_id()),
2152 )
2153 .await?;
2154 let job_id = obj.oid.as_job_id();
2155 let source_obj = CatalogController::create_object(
2156 &txn,
2157 ObjectType::Source,
2158 TEST_OWNER_ID,
2159 Some(job_id.as_object_id()),
2160 )
2161 .await?;
2162 Source::insert(source::ActiveModel::from(PbSource {
2163 id: source_obj.oid.as_source_id(),
2164 schema_id: TEST_SCHEMA_ID,
2165 database_id: TEST_DATABASE_ID,
2166 name: "source_abort_initial".to_owned(),
2167 owner: TEST_OWNER_ID as _,
2168 ..Default::default()
2169 }))
2170 .exec(&txn)
2171 .await?;
2172
2173 table::ActiveModel {
2174 table_id: Set(obj.oid.as_table_id()),
2175 name: Set("table_abort_initial".to_owned()),
2176 optional_associated_source_id: Set(Some(source_obj.oid.as_source_id())),
2177 table_type: Set(TableType::Table),
2178 belongs_to_job_id: Set(None),
2179 columns: Set(vec![].into()),
2180 pk: Set(vec![].into()),
2181 distribution_key: Set(Vec::<i32>::new().into()),
2182 stream_key: Set(Vec::<i32>::new().into()),
2183 append_only: Set(false),
2184 fragment_id: Set(None),
2185 vnode_col_index: Set(None),
2186 row_id_index: Set(None),
2187 value_indices: Set(Vec::<i32>::new().into()),
2188 definition: Set("CREATE TABLE table_abort_initial (v1 INT)".to_owned()),
2189 handle_pk_conflict_behavior: Set(HandleConflictBehavior::NoCheck),
2190 version_column_indices: Set(None),
2191 read_prefix_len_hint: Set(0),
2192 watermark_indices: Set(Vec::<i32>::new().into()),
2193 dist_key_in_pk: Set(Vec::<i32>::new().into()),
2194 dml_fragment_id: Set(None),
2195 cardinality: Set(None),
2196 cleaned_by_watermark: Set(false),
2197 description: Set(None),
2198 version: Set(None),
2199 retention_seconds: Set(None),
2200 cdc_table_id: Set(None),
2201 vnode_count: Set(1),
2202 webhook_info: Set(None),
2203 engine: Set(None),
2204 clean_watermark_index_in_pk: Set(None),
2205 clean_watermark_indices: Set(None),
2206 refreshable: Set(false),
2207 vector_index_info: Set(None),
2208 cdc_table_type: Set(None),
2209 }
2210 .insert(&txn)
2211 .await?;
2212
2213 let internal_obj = CatalogController::create_object(
2214 &txn,
2215 ObjectType::Table,
2216 TEST_OWNER_ID,
2217 Some(job_id.as_object_id()),
2218 )
2219 .await?;
2220 let internal_table_id = internal_obj.oid.as_table_id();
2221 insert_test_table(
2222 &txn,
2223 internal_table_id,
2224 "__internal_mv_abort_initial",
2225 TableType::Internal,
2226 Some(job_id),
2227 "",
2228 )
2229 .await?;
2230
2231 streaming_job::ActiveModel {
2232 job_id: Set(job_id),
2233 job_status: Set(JobStatus::Creating),
2234 create_type: Set(CreateType::Foreground),
2235 timezone: Set(None),
2236 config_override: Set(None),
2237 adaptive_parallelism_strategy: Set(None),
2238 parallelism: Set(StreamingParallelism::Adaptive),
2239 backfill_parallelism: Set(None),
2240 backfill_adaptive_parallelism_strategy: Set(None),
2241 backfill_orders: Set(None),
2242 max_parallelism: Set(1),
2243 specific_resource_group: Set(None),
2244 is_serverless_backfill: Set(false),
2245 refresh_interval_sec: Set(None),
2246 }
2247 .insert(&txn)
2248 .await?;
2249
2250 let (tx, rx) = oneshot::channel();
2251 inner.register_finish_notifier(TEST_DATABASE_ID, job_id, tx);
2252 txn.commit().await?;
2253 drop(inner);
2254
2255 let abort_result = mgr.try_abort_creating_streaming_job(job_id, true).await?;
2256 assert!(abort_result.aborted);
2257 assert_eq!(abort_result.database_id, Some(TEST_DATABASE_ID));
2258
2259 let err = rx
2260 .await
2261 .expect("finish notifier should be notified")
2262 .expect_err("creating job cancellation should fail the create wait");
2263 assert!(err.contains("cancelled"));
2264
2265 let db = &mgr.inner.read().await.db;
2266 assert!(Object::find_by_id(job_id).one(db).await?.is_none());
2267 assert!(StreamingJob::find_by_id(job_id).one(db).await?.is_none());
2268 assert!(
2269 Table::find_by_id(job_id.as_mv_table_id())
2270 .one(db)
2271 .await?
2272 .is_none()
2273 );
2274 assert!(
2275 Object::find_by_id(internal_table_id)
2276 .one(db)
2277 .await?
2278 .is_none()
2279 );
2280 assert!(
2281 Table::find_by_id(internal_table_id)
2282 .one(db)
2283 .await?
2284 .is_none()
2285 );
2286 assert!(
2287 mgr.inner
2288 .read()
2289 .await
2290 .dropped_tables
2291 .contains_key(&internal_table_id)
2292 );
2293 assert!(
2294 Source::find_by_id(source_obj.oid.as_source_id())
2295 .one(db)
2296 .await?
2297 .is_none()
2298 );
2299
2300 let notification = notification_rx
2301 .recv()
2302 .await
2303 .expect("frontend should receive an abort notification")
2304 .expect("abort notification should be valid");
2305 assert_eq!(notification.operation(), NotificationOperation::Delete);
2306 let object_group = match notification.info {
2307 Some(NotificationInfo::ObjectGroup(object_group)) => object_group,
2308 other => panic!("unexpected notification: {other:?}"),
2309 };
2310 assert!(object_group.objects.iter().any(|object| matches!(
2311 &object.object_info,
2312 Some(PbObjectInfo::Table(table)) if table.id == job_id.as_mv_table_id()
2313 )));
2314 assert!(object_group.objects.iter().any(|object| matches!(
2315 &object.object_info,
2316 Some(PbObjectInfo::Source(source)) if source.id == source_obj.oid.as_source_id()
2317 )));
2318
2319 Ok(())
2320 }
2321
2322 #[tokio::test]
2323 async fn test_failed_foreground_creating_job_is_preserved() -> MetaResult<()> {
2324 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
2325 let (job_id, table_id) = insert_dirty_creating_job_with_fragment(
2326 &mgr,
2327 FragmentId::new(45),
2328 1,
2329 FragmentTypeMask::empty(),
2330 )
2331 .await?;
2332
2333 let abort_result = mgr.try_abort_creating_streaming_job(job_id, false).await?;
2334 assert!(!abort_result.aborted);
2335 assert_eq!(abort_result.database_id, Some(TEST_DATABASE_ID));
2336
2337 let db = &mgr.inner.read().await.db;
2338 assert!(Object::find_by_id(job_id).one(db).await?.is_some());
2339 assert!(StreamingJob::find_by_id(job_id).one(db).await?.is_some());
2340 assert!(Table::find_by_id(table_id).one(db).await?.is_some());
2341
2342 Ok(())
2343 }
2344
2345 #[tokio::test]
2346 async fn test_failed_created_job_is_preserved() -> MetaResult<()> {
2347 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
2348 let (job_id, table_id) = insert_dirty_creating_job_with_fragment(
2349 &mgr,
2350 FragmentId::new(46),
2351 1,
2352 FragmentTypeMask::empty(),
2353 )
2354 .await?;
2355
2356 {
2357 let inner = mgr.inner.read().await;
2358 streaming_job::ActiveModel {
2359 job_id: Set(job_id),
2360 job_status: Set(JobStatus::Created),
2361 ..Default::default()
2362 }
2363 .update(&inner.db)
2364 .await?;
2365 }
2366
2367 let abort_result = mgr.try_abort_creating_streaming_job(job_id, false).await?;
2368 assert!(!abort_result.aborted);
2369 assert_eq!(abort_result.database_id, Some(TEST_DATABASE_ID));
2370 let db = &mgr.inner.read().await.db;
2371 assert!(Object::find_by_id(job_id).one(db).await?.is_some());
2372 assert!(StreamingJob::find_by_id(job_id).one(db).await?.is_some());
2373 assert!(Table::find_by_id(table_id).one(db).await?.is_some());
2374
2375 Ok(())
2376 }
2377
2378 #[tokio::test]
2379 async fn test_clean_dirty_creating_jobs_records_dropped_tables_for_per_db_recovery()
2380 -> MetaResult<()> {
2381 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
2382
2383 let inner = mgr.inner.write().await;
2384 let txn = inner.db.begin().await?;
2385 let mv_obj = CatalogController::create_object(
2386 &txn,
2387 ObjectType::Table,
2388 TEST_OWNER_ID,
2389 Some(TEST_SCHEMA_ID.as_object_id()),
2390 )
2391 .await?;
2392 let job_id = mv_obj.oid.as_job_id();
2393 let mv_table_id = job_id.as_mv_table_id();
2394 insert_test_table(
2395 &txn,
2396 mv_table_id,
2397 "mv_dirty",
2398 TableType::MaterializedView,
2399 None,
2400 "CREATE MATERIALIZED VIEW mv_dirty AS SELECT 1",
2401 )
2402 .await?;
2403
2404 let internal_obj = CatalogController::create_object(
2405 &txn,
2406 ObjectType::Table,
2407 TEST_OWNER_ID,
2408 Some(job_id.as_object_id()),
2409 )
2410 .await?;
2411 let internal_table_id = internal_obj.oid.as_table_id();
2412 insert_test_table(
2413 &txn,
2414 internal_table_id,
2415 "__internal_mv_dirty",
2416 TableType::Internal,
2417 Some(job_id),
2418 "",
2419 )
2420 .await?;
2421
2422 streaming_job::ActiveModel {
2423 job_id: Set(job_id),
2424 job_status: Set(JobStatus::Initial),
2425 create_type: Set(CreateType::Foreground),
2426 timezone: Set(None),
2427 config_override: Set(None),
2428 adaptive_parallelism_strategy: Set(None),
2429 parallelism: Set(StreamingParallelism::Adaptive),
2430 backfill_parallelism: Set(None),
2431 backfill_adaptive_parallelism_strategy: Set(None),
2432 backfill_orders: Set(None),
2433 max_parallelism: Set(1),
2434 specific_resource_group: Set(None),
2435 is_serverless_backfill: Set(false),
2436 refresh_interval_sec: Set(None),
2437 }
2438 .insert(&txn)
2439 .await?;
2440 txn.commit().await?;
2441 drop(inner);
2442
2443 let cleaned = mgr
2444 .clean_dirty_creating_jobs(Some(TEST_DATABASE_ID))
2445 .await?;
2446 assert_eq!(cleaned.streaming_job_ids, vec![job_id]);
2447 assert!(cleaned.source_ids.is_empty());
2448 let mut dropped_table_ids = cleaned.dropped_table_ids;
2449 dropped_table_ids.sort_unstable();
2450 assert_eq!(dropped_table_ids, vec![mv_table_id, internal_table_id]);
2451
2452 let inner = mgr.inner.read().await;
2453 assert!(inner.dropped_tables.contains_key(&mv_table_id));
2454 assert!(inner.dropped_tables.contains_key(&internal_table_id));
2455 assert!(Object::find_by_id(job_id).one(&inner.db).await?.is_none());
2456 assert!(
2457 Object::find_by_id(internal_table_id)
2458 .one(&inner.db)
2459 .await?
2460 .is_none()
2461 );
2462 assert!(
2463 StreamingJob::find_by_id(job_id)
2464 .one(&inner.db)
2465 .await?
2466 .is_none()
2467 );
2468 assert!(
2469 Table::find_by_id(mv_table_id)
2470 .one(&inner.db)
2471 .await?
2472 .is_none()
2473 );
2474 assert!(
2475 Table::find_by_id(internal_table_id)
2476 .one(&inner.db)
2477 .await?
2478 .is_none()
2479 );
2480
2481 Ok(())
2482 }
2483
2484 #[tokio::test]
2485 async fn test_clean_dirty_creating_jobs_notifies_serving_mapping_fragment_delete()
2486 -> MetaResult<()> {
2487 let env = MetaSrvEnv::for_test().await;
2488 let (local_notification_tx, mut local_notification_rx) = mpsc::unbounded_channel();
2489 env.notification_manager()
2490 .insert_local_sender(local_notification_tx);
2491 let mgr = CatalogController::new(env).await?;
2492 let fragment_id = FragmentId::new(3);
2493 let (job_id, mv_table_id) = insert_dirty_creating_job_with_fragment(
2494 &mgr,
2495 fragment_id,
2496 1,
2497 FragmentTypeMask::from(FragmentTypeFlag::Values as u32),
2498 )
2499 .await?;
2500
2501 assert!(
2502 mgr.fragment_serving_infos()
2503 .await?
2504 .contains_key(&fragment_id)
2505 );
2506
2507 let cleaned = mgr
2508 .clean_dirty_creating_jobs(Some(TEST_DATABASE_ID))
2509 .await?;
2510 assert_eq!(cleaned.streaming_job_ids, vec![job_id]);
2511
2512 let inner = mgr.inner.read().await;
2513 assert!(Object::find_by_id(job_id).one(&inner.db).await?.is_none());
2514 assert!(
2515 StreamingJob::find_by_id(job_id)
2516 .one(&inner.db)
2517 .await?
2518 .is_none()
2519 );
2520 assert!(
2521 Table::find_by_id(mv_table_id)
2522 .one(&inner.db)
2523 .await?
2524 .is_none()
2525 );
2526 drop(inner);
2527 assert!(
2528 !mgr.fragment_serving_infos()
2529 .await?
2530 .contains_key(&fragment_id)
2531 );
2532
2533 let notification = local_notification_rx.try_recv().expect(
2534 "dirty-job cleanup must notify the serving mapping worker about deleted fragments",
2535 );
2536 match notification {
2537 LocalNotification::ServingFragmentMappingsDelete(fragment_ids) => {
2538 assert_eq!(fragment_ids, vec![fragment_id]);
2539 }
2540 notification => panic!("unexpected local notification: {notification:?}"),
2541 }
2542
2543 Ok(())
2544 }
2545
2546 #[tokio::test]
2547 async fn test_abort_creating_subscription_commits_delete() -> MetaResult<()> {
2548 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
2549 let pb_view = PbView {
2550 schema_id: TEST_SCHEMA_ID,
2551 database_id: TEST_DATABASE_ID,
2552 name: "subscription_dep_view".to_owned(),
2553 owner: TEST_OWNER_ID as _,
2554 sql: "CREATE VIEW subscription_dep_view AS SELECT 1".to_owned(),
2555 ..Default::default()
2556 };
2557 mgr.create_view(pb_view, HashSet::new()).await?;
2558
2559 let view_id: ViewId = View::find()
2560 .select_only()
2561 .column(view::Column::ViewId)
2562 .filter(view::Column::Name.eq("subscription_dep_view"))
2563 .into_tuple()
2564 .one(&mgr.inner.read().await.db)
2565 .await?
2566 .unwrap();
2567
2568 let mut pb_subscription = PbSubscription {
2569 name: "subscription_to_abort".to_owned(),
2570 definition: "CREATE SUBSCRIPTION subscription_to_abort FROM subscription_dep_view"
2571 .to_owned(),
2572 retention_seconds: 86400,
2573 database_id: TEST_DATABASE_ID,
2574 schema_id: TEST_SCHEMA_ID,
2575 dependent_table_id: view_id.as_object_id().as_table_id(),
2576 owner: TEST_OWNER_ID as _,
2577 subscription_state: SubscriptionState::Init as _,
2578 ..Default::default()
2579 };
2580 mgr.create_subscription_catalog(&mut pb_subscription)
2581 .await?;
2582
2583 mgr.try_abort_creating_subscription(pb_subscription.id)
2584 .await?;
2585
2586 assert!(
2587 Subscription::find_by_id(pb_subscription.id)
2588 .one(&mgr.inner.read().await.db)
2589 .await?
2590 .is_none()
2591 );
2592
2593 Ok(())
2594 }
2595
2596 #[tokio::test]
2597 async fn test_drop_table_cascade_drops_dependent_subscription() -> MetaResult<()> {
2598 let mgr = CatalogController::new(MetaSrvEnv::for_test().await).await?;
2599
2600 let inner = mgr.inner.write().await;
2601 let txn = inner.db.begin().await?;
2602 let table_obj = CatalogController::create_object(
2603 &txn,
2604 ObjectType::Table,
2605 TEST_OWNER_ID,
2606 Some(TEST_SCHEMA_ID.as_object_id()),
2607 )
2608 .await?;
2609 let table_id = table_obj.oid.as_table_id();
2610 insert_test_table(
2611 &txn,
2612 table_id,
2613 "subscription_dep_table",
2614 TableType::Table,
2615 None,
2616 "CREATE TABLE subscription_dep_table (v1 INT)",
2617 )
2618 .await?;
2619 txn.commit().await?;
2620 drop(inner);
2621
2622 let mut pb_subscription = PbSubscription {
2623 name: "subscription_to_drop_with_table".to_owned(),
2624 definition:
2625 "CREATE SUBSCRIPTION subscription_to_drop_with_table FROM subscription_dep_table"
2626 .to_owned(),
2627 retention_seconds: 86400,
2628 database_id: TEST_DATABASE_ID,
2629 schema_id: TEST_SCHEMA_ID,
2630 dependent_table_id: table_id,
2631 owner: TEST_OWNER_ID as _,
2632 subscription_state: SubscriptionState::Created as _,
2633 ..Default::default()
2634 };
2635 mgr.create_subscription_catalog(&mut pb_subscription)
2636 .await?;
2637
2638 mgr.drop_object(ObjectType::Table, table_id, DropMode::Cascade)
2639 .await?;
2640
2641 let db = &mgr.inner.read().await.db;
2642 assert!(Table::find_by_id(table_id).one(db).await?.is_none());
2643 assert!(
2644 Object::find_by_id(table_id.as_object_id())
2645 .one(db)
2646 .await?
2647 .is_none()
2648 );
2649 assert!(
2650 Subscription::find_by_id(pb_subscription.id)
2651 .one(db)
2652 .await?
2653 .is_none()
2654 );
2655 assert!(
2656 Object::find_by_id(pb_subscription.id.as_object_id())
2657 .one(db)
2658 .await?
2659 .is_none()
2660 );
2661
2662 Ok(())
2663 }
2664}