feat: add database and schema in dashboard (#16610)

dashboard/components/Relations.tsx

@@ -33,7 +33,13 @@ import Link from "next/link"
 import { Fragment } from "react"
 import Title from "../components/Title"
 import useFetch from "../lib/api/fetch"
-import { Relation, StreamingJob } from "../lib/api/streaming"
+import {
+  Relation,
+  StreamingJob,
+  getDatabases,
+  getSchemas,
+  getUsers,
+} from "../lib/api/streaming"
 import extractColumnInfo from "../lib/extractInfo"
 import {
   Sink as RwSink,
@@ -114,7 +120,22 @@ export function Relations<R extends Relation>(
   getRelations: () => Promise<R[]>,
   extraColumns: Column<R>[]
 ) {
-  const { response: relationList } = useFetch(getRelations)
+  const { response: relationList } = useFetch(async () => {
+    const relations = await getRelations()
+    const users = await getUsers()
+    const databases = await getDatabases()
+    const schemas = await getSchemas()
+    return relations.map((r) => {
+      // Add owner, schema, and database names. It's linear search but the list is small.
+      const owner = users.find((u) => u.id === r.owner)
+      const ownerName = owner?.name
+      const schema = schemas.find((s) => s.id === r.schemaId)
+      const schemaName = schema?.name
+      const database = databases.find((d) => d.id === r.databaseId)
+      const databaseName = database?.name
+      return { ...r, ownerName, schemaName, databaseName }
+    })
+  })
   const [modalData, setModalId] = useCatalogModal(relationList)
 
   const modal = (
@@ -129,6 +150,8 @@ export function Relations<R extends Relation>(
           <Thead>
             <Tr>
               <Th width={3}>Id</Th>
+              <Th width={5}>Database</Th>
+              <Th width={5}>Schema</Th>
               <Th width={5}>Name</Th>
               <Th width={3}>Owner</Th>
               {extraColumns.map((c) => (
@@ -153,8 +176,10 @@ export function Relations<R extends Relation>(
                     {r.id}
                   </Button>
                 </Td>
+                <Td>{r.databaseName}</Td>
+                <Td>{r.schemaName}</Td>
                 <Td>{r.name}</Td>
-                <Td>{r.owner}</Td>
+                <Td>{r.ownerName}</Td>
                 {extraColumns.map((c) => (
                   <Td key={c.name}>{c.content(r)}</Td>
                 ))}

dashboard/lib/api/api.ts

@@ -26,7 +26,9 @@ export const PREDEFINED_API_ENDPOINTS = [
 ]
 
 export const DEFAULT_API_ENDPOINT: string =
-  process.env.NODE_ENV === "production" ? PROD_API_ENDPOINT : MOCK_API_ENDPOINT
+  process.env.NODE_ENV === "production"
+    ? PROD_API_ENDPOINT
+    : MOCK_API_ENDPOINT; // EXTERNAL_META_NODE_API_ENDPOINT to debug with RisingWave servers
 
 export const API_ENDPOINT_KEY = "risingwave.dashboard.api.endpoint"
 

dashboard/lib/api/fetch.ts

@@ -33,6 +33,7 @@ export default function useFetch<T>(
   const [response, setResponse] = useState<T>()
   const toast = useErrorToast()
 
+  // NOTE(eric): Don't put `fetchFn` in the dependency array. It might be a lambda function
   useEffect(() => {
     const fetchData = async () => {
       if (when) {
@@ -52,7 +53,7 @@ export default function useFetch<T>(
 
     const timer = setInterval(fetchData, intervalMs)
     return () => clearInterval(timer)
-  }, [toast, fetchFn, intervalMs, when])
+  }, [toast, intervalMs, when])
 
   return { response }
 }

dashboard/lib/api/streaming.ts

@@ -17,12 +17,20 @@
 
 import _ from "lodash"
 import sortBy from "lodash/sortBy"
-import { Sink, Source, Table, View } from "../../proto/gen/catalog"
+import {
+  Database,
+  Schema,
+  Sink,
+  Source,
+  Table,
+  View,
+} from "../../proto/gen/catalog"
 import {
   ListObjectDependenciesResponse_ObjectDependencies as ObjectDependencies,
   TableFragments,
 } from "../../proto/gen/meta"
 import { ColumnCatalog, Field } from "../../proto/gen/plan_common"
+import { UserInfo } from "../../proto/gen/user"
 import api from "./api"
 
 export async function getFragments(): Promise<TableFragments[]> {
@@ -37,7 +45,14 @@ export interface Relation {
   id: number
   name: string
   owner: number
+  schemaId: number
+  databaseId: number
   columns: (ColumnCatalog | Field)[]
+
+  // For display
+  ownerName?: string
+  schemaName?: string
+  databaseName?: string
 }
 
 export interface StreamingJob extends Relation {
@@ -135,6 +150,26 @@ export async function getViews() {
   return views
 }
 
+export async function getUsers() {
+  let users: UserInfo[] = (await api.get("/users")).map(UserInfo.fromJSON)
+  users = sortBy(users, (x) => x.id)
+  return users
+}
+
+export async function getDatabases() {
+  let databases: Database[] = (await api.get("/databases")).map(
+    Database.fromJSON
+  )
+  databases = sortBy(databases, (x) => x.id)
+  return databases
+}
+
+export async function getSchemas() {
+  let schemas: Schema[] = (await api.get("/schemas")).map(Schema.fromJSON)
+  schemas = sortBy(schemas, (x) => x.id)
+  return schemas
+}
+
 export async function getObjectDependencies() {
   let objDependencies: ObjectDependencies[] = (
     await api.get("/object_dependencies")

src/meta/src/controller/catalog.rs

@@ -2522,6 +2522,11 @@ impl CatalogController {
         inner.list_databases().await
     }
 
+    pub async fn list_schemas(&self) -> MetaResult<Vec<PbSchema>> {
+        let inner = self.inner.read().await;
+        inner.list_schemas().await
+    }
+
     pub async fn list_all_state_tables(&self) -> MetaResult<Vec<PbTable>> {
         let inner = self.inner.read().await;
         inner.list_all_state_tables().await
@@ -2612,6 +2617,11 @@ impl CatalogController {
         inner.list_views().await
     }
 
+    pub async fn list_users(&self) -> MetaResult<Vec<PbUserInfo>> {
+        let inner = self.inner.read().await;
+        inner.list_users().await
+    }
+
     pub async fn get_table_by_name(
         &self,
         database_name: &str,

src/meta/src/dashboard/mod.rs

@@ -55,14 +55,15 @@ pub(super) mod handlers {
     use itertools::Itertools;
     use risingwave_common_heap_profiling::COLLAPSED_SUFFIX;
     use risingwave_pb::catalog::table::TableType;
-    use risingwave_pb::catalog::{Sink, Source, Table, View};
+    use risingwave_pb::catalog::{PbDatabase, PbSchema, Sink, Source, Table, View};
     use risingwave_pb::common::{WorkerNode, WorkerType};
     use risingwave_pb::meta::list_object_dependencies_response::PbObjectDependencies;
     use risingwave_pb::meta::PbTableFragments;
     use risingwave_pb::monitor_service::{
         GetBackPressureResponse, HeapProfilingResponse, ListHeapProfilingResponse,
         StackTraceResponse,
     };
+    use risingwave_pb::user::PbUserInfo;
     use serde_json::json;
     use thiserror_ext::AsReport;
 
@@ -195,6 +196,37 @@ pub(super) mod handlers {
         Ok(Json(table_fragments))
     }
 
+    pub async fn list_users(Extension(srv): Extension<Service>) -> Result<Json<Vec<PbUserInfo>>> {
+        let users = match &srv.metadata_manager {
+            MetadataManager::V1(mgr) => mgr.catalog_manager.list_users().await,
+            MetadataManager::V2(mgr) => mgr.catalog_controller.list_users().await.map_err(err)?,
+        };
+
+        Ok(Json(users))
+    }
+
+    pub async fn list_databases(
+        Extension(srv): Extension<Service>,
+    ) -> Result<Json<Vec<PbDatabase>>> {
+        let databases = match &srv.metadata_manager {
+            MetadataManager::V1(mgr) => mgr.catalog_manager.list_databases().await,
+            MetadataManager::V2(mgr) => {
+                mgr.catalog_controller.list_databases().await.map_err(err)?
+            }
+        };
+
+        Ok(Json(databases))
+    }
+
+    pub async fn list_schemas(Extension(srv): Extension<Service>) -> Result<Json<Vec<PbSchema>>> {
+        let schemas = match &srv.metadata_manager {
+            MetadataManager::V1(mgr) => mgr.catalog_manager.list_schemas().await,
+            MetadataManager::V2(mgr) => mgr.catalog_controller.list_schemas().await.map_err(err)?,
+        };
+
+        Ok(Json(schemas))
+    }
+
     pub async fn list_object_dependencies(
         Extension(srv): Extension<Service>,
     ) -> Result<Json<Vec<PbObjectDependencies>>> {
@@ -388,6 +420,9 @@ impl DashboardService {
             .route("/internal_tables", get(list_internal_tables))
             .route("/sources", get(list_sources))
             .route("/sinks", get(list_sinks))
+            .route("/users", get(list_users))
+            .route("/databases", get(list_databases))
+            .route("/schemas", get(list_schemas))
             .route("/object_dependencies", get(list_object_dependencies))
             .route("/metrics/cluster", get(prometheus::list_prometheus_cluster))
             .route(

src/meta/src/manager/catalog/database.rs

@@ -296,6 +296,10 @@ impl DatabaseManager {
         self.databases.values().cloned().collect_vec()
     }
 
+    pub fn list_schemas(&self) -> Vec<Schema> {
+        self.schemas.values().cloned().collect_vec()
+    }
+
     pub fn list_creating_background_mvs(&self) -> Vec<Table> {
         self.tables
             .values()

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

@@ -3515,6 +3515,10 @@ impl CatalogManager {
         self.core.lock().await.database.list_databases()
     }
 
+    pub async fn list_schemas(&self) -> Vec<Schema> {
+        self.core.lock().await.database.list_schemas()
+    }
+
     pub async fn list_tables(&self) -> Vec<Table> {
         self.core.lock().await.database.list_tables()
     }
@@ -3866,7 +3870,6 @@ impl CatalogManager {
         Ok(())
     }
 
-    #[cfg(test)]
     pub async fn list_users(&self) -> Vec<UserInfo> {
         self.core.lock().await.user.list_users()
     }