(
{extraColumns.map((c) => (
| {c.content(r)} |
))}
-
- {r.columns
- .filter((col) => ("isHidden" in col ? !col.isHidden : true))
- .map((col) => extractColumnInfo(col))
- .join(", ")}
- |
+ {r.columns && r.columns.length > 0 && (
+
+ {r.columns
+ .filter((col) =>
+ "isHidden" in col ? !col.isHidden : true
+ )
+ .map((col) => extractColumnInfo(col))
+ .join(", ")}
+ |
+ )}
))}
diff --git a/dashboard/lib/api/streaming.ts b/dashboard/lib/api/streaming.ts
index 1a8e97081caa..948cd567d3f2 100644
--- a/dashboard/lib/api/streaming.ts
+++ b/dashboard/lib/api/streaming.ts
@@ -22,6 +22,7 @@ import {
Schema,
Sink,
Source,
+ Subscription,
Table,
View,
} from "../../proto/gen/catalog"
@@ -47,9 +48,9 @@ export interface Relation {
owner: number
schemaId: number
databaseId: number
- columns: (ColumnCatalog | Field)[]
// For display
+ columns?: (ColumnCatalog | Field)[]
ownerName?: string
schemaName?: string
databaseName?: string
@@ -66,6 +67,8 @@ export function relationType(x: Relation) {
return "SINK"
} else if ((x as Source).info !== undefined) {
return "SOURCE"
+ } else if ((x as Subscription).dependentTableId !== undefined) {
+ return "SUBSCRIPTION"
} else {
return "UNKNOWN"
}
@@ -98,7 +101,8 @@ export async function getRelations() {
await getTables(),
await getIndexes(),
await getSinks(),
- await getSources()
+ await getSources(),
+ await getSubscriptions()
)
relations = sortBy(relations, (x) => x.id)
return relations
@@ -150,6 +154,14 @@ export async function getViews() {
return views
}
+export async function getSubscriptions() {
+ let subscriptions: Subscription[] = (await api.get("/subscriptions")).map(
+ Subscription.fromJSON
+ )
+ subscriptions = sortBy(subscriptions, (x) => x.id)
+ return subscriptions
+}
+
export async function getUsers() {
let users: UserInfo[] = (await api.get("/users")).map(UserInfo.fromJSON)
users = sortBy(users, (x) => x.id)
diff --git a/dashboard/mock-server.js b/dashboard/mock-server.js
index 2db52df788e2..50c55e12686b 100644
--- a/dashboard/mock-server.js
+++ b/dashboard/mock-server.js
@@ -45,6 +45,10 @@ app.get("/indexes", (req, res, next) => {
res.json(require("./mock/indexes.json"))
})
+app.get("/indexes", (req, res, next) => {
+ res.json(require("./mock/indexes.json"))
+})
+
app.get("/internal_tables", (req, res, next) => {
res.json(require("./mock/internal_tables.json"))
})
diff --git a/dashboard/package-lock.json b/dashboard/package-lock.json
index 064357bf3f1f..5bf9ae127252 100644
--- a/dashboard/package-lock.json
+++ b/dashboard/package-lock.json
@@ -6,6 +6,7 @@
"": {
"hasInstallScript": true,
"dependencies": {
+ "16": "^0.0.2",
"@chakra-ui/react": "^2.3.1",
"@emotion/react": "^11.10.4",
"@emotion/styled": "^11.10.4",
@@ -3350,6 +3351,14 @@
"resolved": "https://registry.npmjs.org/@zag-js/focus-visible/-/focus-visible-0.1.0.tgz",
"integrity": "sha512-PeaBcTmdZWcFf7n1aM+oiOdZc+sy14qi0emPIeUuGMTjbP0xLGrZu43kdpHnWSXy7/r4Ubp/vlg50MCV8+9Isg=="
},
+ "node_modules/16": {
+ "version": "0.0.2",
+ "resolved": "https://registry.npmjs.org/16/-/16-0.0.2.tgz",
+ "integrity": "sha512-AhG4lpdn+/it+U5Xl1bm5SbaHYTH5NfU/vXZkP7E7CHjtVtITuFVZKa3AZP3gN38RDJHYYtEqWmqzCutlXaR7w==",
+ "dependencies": {
+ "numeric": "^1.2.6"
+ }
+ },
"node_modules/abab": {
"version": "2.0.6",
"resolved": "https://registry.npmjs.org/abab/-/abab-2.0.6.tgz",
@@ -8680,6 +8689,11 @@
"set-blocking": "^2.0.0"
}
},
+ "node_modules/numeric": {
+ "version": "1.2.6",
+ "resolved": "https://registry.npmjs.org/numeric/-/numeric-1.2.6.tgz",
+ "integrity": "sha512-avBiDAP8siMa7AfJgYyuxw1oyII4z2sswS23+O+ZfV28KrtNzy0wxUFwi4f3RyM4eeeXNs1CThxR7pb5QQcMiw=="
+ },
"node_modules/nuqs": {
"version": "1.14.1",
"resolved": "https://registry.npmjs.org/nuqs/-/nuqs-1.14.1.tgz",
@@ -11586,6 +11600,14 @@
}
},
"dependencies": {
+ "16": {
+ "version": "0.0.2",
+ "resolved": "https://registry.npmjs.org/16/-/16-0.0.2.tgz",
+ "integrity": "sha512-AhG4lpdn+/it+U5Xl1bm5SbaHYTH5NfU/vXZkP7E7CHjtVtITuFVZKa3AZP3gN38RDJHYYtEqWmqzCutlXaR7w==",
+ "requires": {
+ "numeric": "^1.2.6"
+ }
+ },
"@aashutoshrathi/word-wrap": {
"version": "1.2.6",
"resolved": "https://registry.npmjs.org/@aashutoshrathi/word-wrap/-/word-wrap-1.2.6.tgz",
@@ -18034,6 +18056,11 @@
"set-blocking": "^2.0.0"
}
},
+ "numeric": {
+ "version": "1.2.6",
+ "resolved": "https://registry.npmjs.org/numeric/-/numeric-1.2.6.tgz",
+ "integrity": "sha512-avBiDAP8siMa7AfJgYyuxw1oyII4z2sswS23+O+ZfV28KrtNzy0wxUFwi4f3RyM4eeeXNs1CThxR7pb5QQcMiw=="
+ },
"nuqs": {
"version": "1.14.1",
"resolved": "https://registry.npmjs.org/nuqs/-/nuqs-1.14.1.tgz",
diff --git a/dashboard/package.json b/dashboard/package.json
index f3c1436aca98..a3716f7802cc 100644
--- a/dashboard/package.json
+++ b/dashboard/package.json
@@ -13,6 +13,7 @@
"clean": "rm -rf .next/ && rm -rf out/"
},
"dependencies": {
+ "16": "^0.0.2",
"@chakra-ui/react": "^2.3.1",
"@emotion/react": "^11.10.4",
"@emotion/styled": "^11.10.4",
diff --git a/dashboard/pages/subscriptions.tsx b/dashboard/pages/subscriptions.tsx
new file mode 100644
index 000000000000..b2daa38c3f95
--- /dev/null
+++ b/dashboard/pages/subscriptions.tsx
@@ -0,0 +1,38 @@
+/*
+ * Copyright 2024 RisingWave Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+import { Column, Relations } from "../components/Relations"
+import { getSubscriptions } from "../lib/api/streaming"
+import { Subscription as RwSubscription } from "../proto/gen/catalog"
+
+export default function Subscriptions() {
+ const subscriptionRetentionSeconds: Column = {
+ name: "Retention Seconds",
+ width: 3,
+ content: (r) => r.retentionSeconds ?? "unknown",
+ }
+
+ const subscriptionDependentTableId: Column = {
+ name: "Dependent Table Id",
+ width: 3,
+ content: (r) => r.dependentTableId ?? "unknown",
+ }
+ return Relations("Subscriptions", getSubscriptions, [
+ subscriptionRetentionSeconds,
+ subscriptionDependentTableId,
+ ])
+}
diff --git a/src/meta/src/controller/catalog.rs b/src/meta/src/controller/catalog.rs
index b8d88fabad22..409672138d79 100644
--- a/src/meta/src/controller/catalog.rs
+++ b/src/meta/src/controller/catalog.rs
@@ -581,6 +581,29 @@ impl CatalogController {
}
}));
+ let subscription_dependencies: Vec<(SubscriptionId, TableId)> = Subscription::find()
+ .select_only()
+ .columns([
+ subscription::Column::SubscriptionId,
+ subscription::Column::DependentTableId,
+ ])
+ .join(JoinType::InnerJoin, subscription::Relation::Object.def())
+ .filter(
+ subscription::Column::SubscriptionState
+ .eq(Into::::into(SubscriptionState::Created))
+ .and(subscription::Column::DependentTableId.is_not_null()),
+ )
+ .into_tuple()
+ .all(&inner.db)
+ .await?;
+
+ obj_dependencies.extend(subscription_dependencies.into_iter().map(
+ |(subscription_id, table_id)| PbObjectDependencies {
+ object_id: subscription_id as _,
+ referenced_object_id: table_id as _,
+ },
+ ));
+
Ok(obj_dependencies)
}
diff --git a/src/meta/src/dashboard/mod.rs b/src/meta/src/dashboard/mod.rs
index ec52c3a3ee53..122955403261 100644
--- a/src/meta/src/dashboard/mod.rs
+++ b/src/meta/src/dashboard/mod.rs
@@ -55,7 +55,7 @@ pub(super) mod handlers {
use itertools::Itertools;
use risingwave_common_heap_profiling::COLLAPSED_SUFFIX;
use risingwave_pb::catalog::table::TableType;
- use risingwave_pb::catalog::{PbDatabase, PbSchema, Sink, Source, Table, View};
+ use risingwave_pb::catalog::{PbDatabase, PbSchema, Sink, Source, Subscription, Table, View};
use risingwave_pb::common::{WorkerNode, WorkerType};
use risingwave_pb::meta::list_object_dependencies_response::PbObjectDependencies;
use risingwave_pb::meta::PbTableFragments;
@@ -141,6 +141,21 @@ pub(super) mod handlers {
list_table_catalogs_inner(&srv.metadata_manager, TableType::Index).await
}
+ pub async fn list_subscription(
+ Extension(srv): Extension,
+ ) -> Result>> {
+ let subscriptions = match &srv.metadata_manager {
+ MetadataManager::V1(mgr) => mgr.catalog_manager.list_subscriptions().await,
+ MetadataManager::V2(mgr) => mgr
+ .catalog_controller
+ .list_subscriptions()
+ .await
+ .map_err(err)?,
+ };
+
+ Ok(Json(subscriptions))
+ }
+
pub async fn list_internal_tables(
Extension(srv): Extension,
) -> Result>> {
@@ -417,6 +432,7 @@ impl DashboardService {
.route("/materialized_views", get(list_materialized_views))
.route("/tables", get(list_tables))
.route("/indexes", get(list_indexes))
+ .route("/subscriptions", get(list_subscription))
.route("/internal_tables", get(list_internal_tables))
.route("/sources", get(list_sources))
.route("/sinks", get(list_sinks))
diff --git a/src/meta/src/manager/catalog/mod.rs b/src/meta/src/manager/catalog/mod.rs
index ab452dfe05b2..b600a4a16c54 100644
--- a/src/meta/src/manager/catalog/mod.rs
+++ b/src/meta/src/manager/catalog/mod.rs
@@ -3702,6 +3702,13 @@ impl CatalogManager {
}
}
+ for subscription in core.subscriptions.values() {
+ dependencies.push(PbObjectDependencies {
+ object_id: subscription.id,
+ referenced_object_id: subscription.dependent_table_id,
+ });
+ }
+
dependencies
}