add graph

Author: xxhZs

dashboard/lib/api/streaming.ts

@@ -67,7 +67,10 @@ export function relationType(x: Relation) {
     return "SINK"
   } else if ((x as Source).info !== undefined) {
     return "SOURCE"
-  } else {
+  }  else if ((x as Subscription).dependentTableId !== undefined) {
+    return "SUBSCRIPTION"
+  }
+  else {
     return "UNKNOWN"
   }
 }

src/meta/src/controller/catalog.rs

@@ -581,6 +581,29 @@ impl CatalogController {
             }
         }));
 
+        let subscription_dependencies: Vec<(SubscriptionId, TableId)> = Sink::find()
+            .select_only()
+            .columns([
+                subscription::Column::SubscriptionId,
+                subscription::Column::DependentTableId,
+            ])
+            .join(JoinType::InnerJoin, sink::Relation::Object.def())
+            .filter(
+                subscription::Column::SubscriptionState
+                    .eq(Into::<i32>::into(SubscriptionState::Created))
+                    .and(subscription::Column::DependentTableId.is_not_null()),
+            )
+            .into_tuple()
+            .all(&inner.db)
+            .await?;
+
+        obj_dependencies.extend(subscription_dependencies.into_iter().map(
+            |(sink_id, table_id)| PbObjectDependencies {
+                object_id: table_id as _,
+                referenced_object_id: sink_id as _,
+            },
+        ));
+
         Ok(obj_dependencies)
     }
 

src/meta/src/manager/catalog/mod.rs

@@ -3698,6 +3698,13 @@ impl CatalogManager {
             }
         }
 
+        for subscription in core.subscriptions.values() {
+            dependencies.push(PbObjectDependencies {
+                object_id: subscription.id,
+                referenced_object_id: subscription.dependent_table_id,
+            });
+        }
+
         dependencies
     }