record dependencies in Request message

Signed-off-by: Richard Chien <[email protected]>

proto/catalog.proto

@@ -342,7 +342,7 @@ message Table {
   repeated plan_common.ColumnCatalog columns = 5;
   repeated common.ColumnOrder pk = 6;
   // For cdc table created from a cdc source, here records the source id.
-  repeated uint32 dependent_relations = 8;
+  repeated uint32 dependent_relations = 8; // TODO(rc): deprecate this by passing dependencies via `Request` message
   oneof optional_associated_source_id {
     uint32 associated_source_id = 9;
   }

src/meta/service/src/ddl_service.rs

@@ -241,6 +241,7 @@ impl DdlService for DdlServiceImpl {
                         fragment_graph,
                         CreateType::Foreground,
                         None,
+                        HashSet::new(), // TODO(rc): pass dependencies through this field instead of `PbSource`
                     ))
                     .await?;
                 Ok(Response::new(CreateSourceResponse {
@@ -295,6 +296,7 @@ impl DdlService for DdlServiceImpl {
             fragment_graph,
             CreateType::Foreground,
             affected_table_change.map(Self::extract_replace_table_info),
+            HashSet::new(), // TODO(rc): pass dependencies through this field instead of `PbSink`
         );
 
         let version = self.ddl_controller.run_command(command).await?;
@@ -380,6 +382,7 @@ impl DdlService for DdlServiceImpl {
         let mview = req.get_materialized_view()?.clone();
         let create_type = mview.get_create_type().unwrap_or(CreateType::Foreground);
         let fragment_graph = req.get_fragment_graph()?.clone();
+        let dependencies = req.get_dependencies().iter().map(|id| *id as i32).collect();
 
         let stream_job = StreamingJob::MaterializedView(mview);
         let version = self
@@ -389,6 +392,7 @@ impl DdlService for DdlServiceImpl {
                 fragment_graph,
                 create_type,
                 None,
+                dependencies,
             ))
             .await?;
 
@@ -442,6 +446,7 @@ impl DdlService for DdlServiceImpl {
                 fragment_graph,
                 CreateType::Foreground,
                 None,
+                HashSet::new(),
             ))
             .await?;
 
@@ -528,6 +533,7 @@ impl DdlService for DdlServiceImpl {
                 fragment_graph,
                 CreateType::Foreground,
                 None,
+                HashSet::new(), // TODO(rc): pass dependencies through this field instead of `PbTable`
             ))
             .await?;
 

src/meta/src/controller/streaming_job.rs

@@ -112,6 +112,7 @@ impl CatalogController {
         ctx: &StreamContext,
         parallelism: &Option<Parallelism>,
         max_parallelism: usize,
+        mut dependencies: HashSet<ObjectId>,
     ) -> MetaResult<()> {
         let inner = self.inner.write().await;
         let txn = inner.db.begin().await?;
@@ -133,9 +134,16 @@ impl CatalogController {
         )
         .await?;
 
-        // check if any dependent relation is in altering status.
-        let dependent_relations = streaming_job.dependent_relations();
-        if !dependent_relations.is_empty() {
+        // TODO(rc): pass all dependencies uniformly, deprecate `dependent_relations` and `dependent_secret_ids`.
+        dependencies.extend(
+            streaming_job
+                .dependent_relations()
+                .into_iter()
+                .map(|id| id as ObjectId),
+        );
+
+        // check if any dependency is in altering status.
+        if !dependencies.is_empty() {
             let altering_cnt = ObjectDependency::find()
                 .join(
                     JoinType::InnerJoin,
@@ -144,7 +152,7 @@ impl CatalogController {
                 .join(JoinType::InnerJoin, object::Relation::StreamingJob.def())
                 .filter(
                     object_dependency::Column::Oid
-                        .is_in(dependent_relations.iter().map(|id| *id as ObjectId))
+                        .is_in(dependencies.clone())
                         .and(object::Column::ObjType.eq(ObjectType::Table))
                         .and(streaming_job::Column::JobStatus.ne(JobStatus::Created))
                         .and(
@@ -308,17 +316,19 @@ impl CatalogController {
             }
         }
 
-        // get dependent secrets.
-        let dependent_secret_ids = streaming_job.dependent_secret_ids()?;
+        // collect dependent secrets.
+        dependencies.extend(
+            streaming_job
+                .dependent_secret_ids()?
+                .into_iter()
+                .map(|secret_id| secret_id as ObjectId),
+        );
 
-        let dependent_objs = dependent_relations
-            .iter()
-            .chain(dependent_secret_ids.iter());
         // record object dependency.
-        if !dependent_secret_ids.is_empty() || !dependent_relations.is_empty() {
-            ObjectDependency::insert_many(dependent_objs.map(|id| {
+        if !dependencies.is_empty() {
+            ObjectDependency::insert_many(dependencies.into_iter().map(|oid| {
                 object_dependency::ActiveModel {
-                    oid: Set(*id as _),
+                    oid: Set(oid),
                     used_by: Set(streaming_job.id() as _),
                     ..Default::default()
                 }

src/meta/src/manager/streaming_job.rs

@@ -299,7 +299,7 @@ impl StreamingJob {
         }
     }
 
-    // TODO: record all objects instead.
+    // TODO: to be removed, pass all objects uniformly through `dependencies` field instead.
     pub fn dependent_relations(&self) -> Vec<u32> {
         match self {
             StreamingJob::MaterializedView(table) => table.dependent_relations.clone(),

src/meta/src/rpc/ddl_controller.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use std::cmp::Ordering;
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::num::NonZeroUsize;
 use std::sync::Arc;
 use std::time::Duration;
@@ -36,9 +36,10 @@ use risingwave_connector::source::{
 };
 use risingwave_connector::{dispatch_source_prop, WithOptionsSecResolved};
 use risingwave_meta_model::object::ObjectType;
+use risingwave_meta_model::prelude::ObjectDependency;
 use risingwave_meta_model::{
-    ConnectionId, DatabaseId, FunctionId, IndexId, ObjectId, SchemaId, SecretId, SinkId, SourceId,
-    SubscriptionId, TableId, UserId, ViewId,
+    object_dependency, ConnectionId, DatabaseId, FunctionId, IndexId, ObjectId, SchemaId, SecretId,
+    SinkId, SourceId, SubscriptionId, TableId, UserId, ViewId,
 };
 use risingwave_pb::catalog::{
     Comment, Connection, CreateType, Database, Function, PbSink, Schema, Secret, Sink, Source,
@@ -58,6 +59,7 @@ use risingwave_pb::stream_plan::{
     Dispatcher, DispatcherType, FragmentTypeFlag, MergeNode, PbStreamFragmentGraph,
     StreamFragmentGraph as StreamFragmentGraphProto,
 };
+use sea_orm::{EntityTrait, Set};
 use thiserror_ext::AsReport;
 use tokio::sync::Semaphore;
 use tokio::time::sleep;
@@ -137,6 +139,7 @@ pub enum DdlCommand {
         StreamFragmentGraphProto,
         CreateType,
         Option<ReplaceTableInfo>,
+        HashSet<ObjectId>,
     ),
     DropStreamingJob(StreamingJobId, DropMode, Option<ReplaceTableInfo>),
     AlterName(alter_name_request::Object, String),
@@ -177,7 +180,7 @@ impl DdlCommand {
             | DdlCommand::CommentOn(_)
             | DdlCommand::CreateSecret(_)
             | DdlCommand::AlterSwapRename(_) => true,
-            DdlCommand::CreateStreamingJob(_, _, _, _)
+            DdlCommand::CreateStreamingJob(_, _, _, _, _)
             | DdlCommand::CreateSourceWithoutStreamingJob(_)
             | DdlCommand::ReplaceTable(_)
             | DdlCommand::AlterSourceColumn(_)
@@ -310,11 +313,13 @@ impl DdlController {
                     fragment_graph,
                     _create_type,
                     affected_table_replace_info,
+                    dependencies,
                 ) => {
                     ctrl.create_streaming_job(
                         stream_job,
                         fragment_graph,
                         affected_table_replace_info,
+                        dependencies,
                     )
                     .await
                 }
@@ -920,6 +925,7 @@ impl DdlController {
         mut streaming_job: StreamingJob,
         fragment_graph: StreamFragmentGraphProto,
         affected_table_replace_info: Option<ReplaceTableInfo>,
+        dependencies: HashSet<ObjectId>,
     ) -> MetaResult<NotificationVersion> {
         let ctx = StreamContext::from_protobuf(fragment_graph.get_ctx().unwrap());
         self.metadata_manager
@@ -929,6 +935,7 @@ impl DdlController {
                 &ctx,
                 &fragment_graph.parallelism,
                 fragment_graph.max_parallelism as _,
+                dependencies,
             )
             .await?;
         let job_id = streaming_job.id();
@@ -947,7 +954,6 @@ impl DdlController {
             .unwrap();
         let _reschedule_job_lock = self.stream_manager.reschedule_lock_read_guard().await;
 
-        let id = streaming_job.id();
         let name = streaming_job.name();
         let definition = streaming_job.definition();
         let source_id = match &streaming_job {
@@ -969,7 +975,7 @@ impl DdlController {
             Err(err) => {
                 tracing::error!(id = job_id, error = %err.as_report(), "failed to create streaming job");
                 let event = risingwave_pb::meta::event_log::EventCreateStreamJobFail {
-                    id,
+                    id: job_id,
                     name,
                     definition,
                     error: err.as_report().to_string(),