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