add spill

dashboard/README.md

@@ -33,8 +33,10 @@ For example:
 
 ```bash
 ./risedev d
-sqllogictest -p 4566 -d dev './e2e_test/nexmark/create_tables.slt.part'
-sqllogictest -p 4566 -d dev './e2e_test/streaming/nexmark/create_views.slt.part'
+./risedev slt  e2e_test/nexmark/create_sources.slt.part
+./risedev psql -c 'CREATE TABLE dimension (v1 int);'
+./risedev psql -c 'CREATE MATERIALIZED VIEW mv AS SELECT auction.* FROM dimension join auction on auction.id-auction.id = dimension.v1;'
+./risedev psql -c 'INSERT INTO dimension select 0 from generate_series(1, 50);'
 ```
 
 Install dependencies and start the development server.

dashboard/lib/api/metric.ts

@@ -112,23 +112,55 @@ function convertToBackPressureMetrics(
   return bpMetrics
 }
 
+export function calculateSpill(
+  current: BackPressureInfo[],
+  newBP: BackPressureInfo[],
+  mapSpilled: Map<string, number>,
+  intervalMs: number
+): Map<string, number> {
+  let mapOld = convertToMapAndAgg(current)
+  let mapNew = convertToMapAndAgg(newBP)
+  let allKeys = new Set([
+    ...mapNew.keys(),
+    ...mapOld.keys(),
+    ...mapSpilled.keys(),
+  ])
+  let newSpilled = new Map<string, number>()
+  allKeys.forEach((key) => {
+    if (mapOld.has(key)) {
+      let newValue = mapNew.get(key) || 0
+      let oldValue = mapOld.get(key) || 0
+      let oldSpill = mapSpilled.get(key) || 0
+      let intervalNs = intervalMs * 1000000
+      let rawSpill = newValue - oldValue - intervalNs + oldSpill
+      let spill = rawSpill < 0 ? 0 : rawSpill
+      newSpilled.set(key, spill)
+    } else {
+      newSpilled.set(key, 0)
+    }
+  })
+  return newSpilled
+}
+
 export function calculateBPRate(
   backPressureNew: BackPressureInfo[],
   backPressureOld: BackPressureInfo[],
+  mapSpill: Map<string, number>,
   intervalMs: number
 ): BackPressuresMetrics {
   let mapNew = convertToMapAndAgg(backPressureNew)
   let mapOld = convertToMapAndAgg(backPressureOld)
   let result = new Map<string, number>()
-  mapNew.forEach((value, key) => {
+  mapNew.forEach((newValue, key) => {
     if (mapOld.has(key)) {
-      result.set(
-        key,
-        // The *100 in end of the formular is to convert the BP rate to the value used in web UI drawing
-        ((value - (mapOld.get(key) || 0)) /
-          ((intervalMs / 1000) * 1000000000)) *
-          100
-      )
+      let oldValue = mapOld.get(key) || 0
+      let intervalNs = intervalMs * 1000000
+      let spill = mapSpill.get(key) || 0
+      let backpressureRate = (newValue - oldValue + spill) / intervalNs
+      let scale = 100 // Convert BP rate from decimal to %.
+      let scaledBackpressureRate = backpressureRate * scale
+
+      result.set(key, scaledBackpressureRate)
     } else {
       result.set(key, 0)
     }

dashboard/pages/fragment_graph.tsx

@@ -41,6 +41,7 @@ import useFetch from "../lib/api/fetch"
 import {
   BackPressureInfo,
   calculateBPRate,
+  calculateSpill,
   fetchEmbeddedBackPressure,
   fetchPrometheusBackPressure,
 } from "../lib/api/metric"
@@ -187,6 +188,8 @@ const backPressureDataSources: BackPressureDataSource[] = [
 interface EmbeddedBackPressureInfo {
   previous: BackPressureInfo[]
   current: BackPressureInfo[]
+  // Extra duration longer than the polling interval will be spilled.
+  spilled: Map<string, number>
 }
 
 export default function Streaming() {
@@ -307,17 +310,27 @@ export default function Streaming() {
         fetchEmbeddedBackPressure().then(
           (newBP) => {
             console.log(newBP)
-            setEmbeddedBackPressureInfo((prev) =>
-              prev
-                ? {
-                    previous: prev.current,
-                    current: newBP,
-                  }
-                : {
-                    previous: newBP, // Use current value to show zero rate, but it's fine
-                    current: newBP,
-                  }
-            )
+            setEmbeddedBackPressureInfo((prev) => {
+              if (prev) {
+                let spilled = calculateSpill(
+                  prev.current,
+                  newBP,
+                  prev.spilled,
+                  INTERVAL
+                )
+                return {
+                  previous: prev.current,
+                  current: newBP,
+                  spilled: spilled,
+                }
+              } else {
+                return {
+                  previous: newBP, // Use current value to show zero rate, but it's fine
+                  current: newBP,
+                  spilled: new Map<string, number>(),
+                }
+              }
+            })
           },
           (e) => {
             console.error(e)
@@ -339,6 +352,7 @@ export default function Streaming() {
         const metrics = calculateBPRate(
           embeddedBackPressureInfo.current,
           embeddedBackPressureInfo.previous,
+          embeddedBackPressureInfo.spilled,
           INTERVAL
         )
         for (const m of metrics.outputBufferBlockingDuration) {