feat(dashboard): list streaming job from streaming_job table & show info

dashboard/components/Relations.tsx

@@ -35,7 +35,7 @@ import Title from "../components/Title"
 import useFetch from "../lib/api/fetch"
 import {
   Relation,
-  StreamingJob,
+  StreamingRelation,
   getDatabases,
   getSchemas,
   getUsers,
@@ -73,7 +73,7 @@ export const dependentsColumn: Column<Relation> = {
   ),
 }
 
-export const fragmentsColumn: Column<StreamingJob> = {
+export const fragmentsColumn: Column<StreamingRelation> = {
   name: "Fragments",
   width: 1,
   content: (r) => (

dashboard/lib/api/streaming.ts

@@ -15,6 +15,7 @@
  *
  */
 
+import { Expose, plainToInstance } from "class-transformer"
 import _ from "lodash"
 import sortBy from "lodash/sortBy"
 import {
@@ -53,14 +54,6 @@ export async function getRelationIdInfos(): Promise<RelationIdInfos> {
   return fragmentIds
 }
 
-export async function getFragments(): Promise<TableFragments[]> {
-  let fragmentList: TableFragments[] = (await api.get("/fragments2")).map(
-    TableFragments.fromJSON
-  )
-  fragmentList = sortBy(fragmentList, (x) => x.tableId)
-  return fragmentList
-}
-
 export interface Relation {
   id: number
   name: string
@@ -75,7 +68,43 @@ export interface Relation {
   databaseName?: string
 }
 
-export interface StreamingJob extends Relation {
+export class StreamingJob {
+  @Expose({ name: "jobId" })
+  id!: number
+  @Expose({ name: "objType" })
+  _objType!: string
+  name!: string
+  jobStatus!: string
+  @Expose({ name: "parallelism" })
+  _parallelism!: any
+  maxParallelism!: number
+
+  get parallelism() {
+    const parallelism = this._parallelism
+    if (typeof parallelism === "string") {
+      // `Adaptive`
+      return parallelism
+    } else if (typeof parallelism === "object") {
+      // `Fixed (64)`
+      let key = Object.keys(parallelism)[0]
+      let value = parallelism[key]
+      return `${key} (${value})`
+    } else {
+      // fallback
+      return JSON.stringify(parallelism)
+    }
+  }
+
+  get type() {
+    if (this._objType == "Table") {
+      return "Table / MV"
+    } else {
+      return this._objType
+    }
+  }
+}
+
+export interface StreamingRelation extends Relation {
   dependentRelations: number[]
 }
 
@@ -98,17 +127,15 @@ export function relationTypeTitleCase(x: Relation) {
   return _.startCase(_.toLower(relationType(x)))
 }
 
-export function relationIsStreamingJob(x: Relation): x is StreamingJob {
+export function relationIsStreamingJob(x: Relation): x is StreamingRelation {
   const type = relationType(x)
   return type !== "UNKNOWN" && type !== "SOURCE" && type !== "INTERNAL"
 }
 
 export async function getStreamingJobs() {
-  let jobs = _.concat<StreamingJob>(
-    await getMaterializedViews(),
-    await getTables(),
-    await getIndexes(),
-    await getSinks()
+  let jobs = plainToInstance(
+    StreamingJob,
+    (await api.get("/streaming_jobs")) as any[]
   )
   jobs = sortBy(jobs, (x) => x.id)
   return jobs

dashboard/package.json

@@ -24,6 +24,7 @@
     "@uidotdev/usehooks": "^2.4.1",
     "base64url": "^3.0.1",
     "bootstrap-icons": "^1.9.1",
+    "class-transformer": "^0.5.1",
     "d3": "^7.6.1",
     "d3-axis": "^3.0.0",
     "d3-dag": "^0.11.4",
@@ -42,6 +43,7 @@
     "react-json-view": "^1.21.3",
     "react-syntax-highlighter": "^15.5.0",
     "recharts": "^2.3.2",
+    "reflect-metadata": "^0.2.2",
     "styled-components": "5.3.0",
     "ts-proto": "^1.169.1"
   },

dashboard/pages/_app.tsx

@@ -15,6 +15,7 @@
  *
  */
 import "bootstrap-icons/font/bootstrap-icons.css"
+import "reflect-metadata"
 import "../styles/global.css"
 
 import { ChakraProvider } from "@chakra-ui/react"

dashboard/pages/fragment_graph.tsx

@@ -24,7 +24,12 @@ import {
   HStack,
   Input,
   Select,
+  Table,
+  TableContainer,
+  Tbody,
+  Td,
   Text,
+  Tr,
   VStack,
 } from "@chakra-ui/react"
 import * as d3 from "d3"
@@ -184,7 +189,7 @@ function buildFragmentDependencyAsEdges(
   return nodes
 }
 
-const SIDEBAR_WIDTH = 200
+const SIDEBAR_WIDTH = 225
 
 type BackPressureDataSource = "Embedded" | "Prometheus"
 const backPressureDataSources: BackPressureDataSource[] = [
@@ -202,23 +207,28 @@ interface EmbeddedBackPressureInfo {
 }
 
 export default function Streaming() {
-  const { response: relationList } = useFetch(getStreamingJobs)
+  const { response: streamingJobList } = useFetch(getStreamingJobs)
   const { response: relationIdInfos } = useFetch(getRelationIdInfos)
 
-  const [relationId, setRelationId] = useQueryState("id", parseAsInteger)
+  const [jobId, setJobId] = useQueryState("id", parseAsInteger)
   const [selectedFragmentId, setSelectedFragmentId] = useState<number>()
   const [tableFragments, setTableFragments] = useState<TableFragments>()
 
+  const job = useMemo(
+    () => streamingJobList?.find((j) => j.id === jobId),
+    [streamingJobList, jobId]
+  )
+
   const toast = useErrorToast()
 
   useEffect(() => {
-    if (relationId) {
+    if (jobId) {
       setTableFragments(undefined)
-      getFragmentsByJobId(relationId).then((tf) => {
+      getFragmentsByJobId(jobId).then((tf) => {
         setTableFragments(tf)
       })
     }
-  }, [relationId])
+  }, [jobId])
 
   const fragmentDependencyCallback = useCallback(() => {
     if (tableFragments) {
@@ -232,14 +242,14 @@ export default function Streaming() {
   }, [tableFragments])
 
   useEffect(() => {
-    if (relationList) {
-      if (!relationId) {
-        if (relationList.length > 0) {
-          setRelationId(relationList[0].id)
+    if (streamingJobList) {
+      if (!jobId) {
+        if (streamingJobList.length > 0) {
+          setJobId(streamingJobList[0].id)
         }
       }
     }
-  }, [relationId, relationList, setRelationId])
+  }, [jobId, streamingJobList, setJobId])
 
   // The table fragments of the selected fragment id
   const fragmentDependency = fragmentDependencyCallback()?.fragmentDep
@@ -276,7 +286,7 @@ export default function Streaming() {
         const fragmentIdToRelationId = map[relationId].map
         for (const fragmentId in fragmentIdToRelationId) {
           if (parseInt(fragmentId) == searchFragIdInt) {
-            setRelationId(parseInt(relationId))
+            setJobId(parseInt(relationId))
             setSelectedFragmentId(searchFragIdInt)
             return
           }
@@ -295,7 +305,7 @@ export default function Streaming() {
         for (const fragmentId in fragmentIdToRelationId) {
           let actorIds = fragmentIdToRelationId[fragmentId].ids
           if (actorIds.includes(searchActorIdInt)) {
-            setRelationId(parseInt(relationId))
+            setJobId(parseInt(relationId))
             setSelectedFragmentId(parseInt(fragmentId))
             return
           }
@@ -406,41 +416,70 @@ export default function Streaming() {
           height="full"
         >
           <FormControl>
-            <FormLabel>Relations</FormLabel>
+            <FormLabel>Streaming Jobs</FormLabel>
             <Input
               list="relationList"
               spellCheck={false}
               onChange={(event) => {
-                const id = relationList?.find(
+                const id = streamingJobList?.find(
                   (x) => x.name == event.target.value
                 )?.id
                 if (id) {
-                  setRelationId(id)
+                  setJobId(id)
                 }
               }}
               placeholder="Search..."
               mb={2}
             ></Input>
             <datalist id="relationList">
-              {relationList &&
-                relationList.map((r) => (
+              {streamingJobList &&
+                streamingJobList.map((r) => (
                   <option value={r.name} key={r.id}>
                     ({r.id}) {r.name}
                   </option>
                 ))}
             </datalist>
             <Select
-              value={relationId ?? undefined}
-              onChange={(event) => setRelationId(parseInt(event.target.value))}
+              value={jobId ?? undefined}
+              onChange={(event) => setJobId(parseInt(event.target.value))}
             >
-              {relationList &&
-                relationList.map((r) => (
+              {streamingJobList &&
+                streamingJobList.map((r) => (
                   <option value={r.id} key={r.name}>
                     ({r.id}) {r.name}
                   </option>
                 ))}
             </Select>
           </FormControl>
+          {job && (
+            <FormControl>
+              <FormLabel>Information</FormLabel>
+              <TableContainer>
+                <Table size="sm">
+                  <Tbody>
+                    <Tr>
+                      <Td fontWeight="medium">Type</Td>
+                      <Td isNumeric>{job.type}</Td>
+                    </Tr>
+                    <Tr>
+                      <Td fontWeight="medium">Status</Td>
+                      <Td isNumeric>{job.jobStatus}</Td>
+                    </Tr>
+                    <Tr>
+                      <Td fontWeight="medium">Parallelism</Td>
+                      <Td isNumeric>{job.parallelism}</Td>
+                    </Tr>
+                    <Tr>
+                      <Td fontWeight="medium" paddingEnd={0}>
+                        Max Parallelism
+                      </Td>
+                      <Td isNumeric>{job.maxParallelism}</Td>
+                    </Tr>
+                  </Tbody>
+                </Table>
+              </TableContainer>
+            </FormControl>
+          )}
           <FormControl>
             <FormLabel>Goto</FormLabel>
             <VStack spacing={2}>

dashboard/tsconfig.json

@@ -18,6 +18,8 @@
     "isolatedModules": true,
     "jsx": "preserve",
     "incremental": true,
+    "experimentalDecorators": true,
+    "emitDecoratorMetadata": true,
   },
   "include": [
     "next-env.d.ts",