Merge branch 'main' into yiming/single-database-command

.config/nextest.toml

@@ -34,7 +34,3 @@ test-group = 'failpoint-limited'
 [[profile.ci-sim.overrides]]
 filter = 'test(failpoint_limited::)'
 test-group = 'failpoint-limited'
-
-[[profile.ci.junit.overrides]]
-filter = 'test(failpoint_limited::)'
-test-group = 'failpoint-limited'

Cargo.toml

@@ -161,6 +161,9 @@ deltalake = { version = "0.20.1", features = [
 itertools = "0.13.0"
 jsonbb = "0.1.4"
 lru = { git = "https://github.com/risingwavelabs/lru-rs.git", rev = "2682b85" }
+mysql_async = { version = "0.34", default-features = false, features = [
+    "default",
+] }
 parquet = { version = "53", features = ["async"] }
 thiserror-ext = "0.1.2"
 tikv-jemalloc-ctl = { git = "https://github.com/risingwavelabs/jemallocator.git", rev = "64a2d9" }
@@ -285,6 +288,7 @@ new_without_default = "allow"
 # duplicated_attributes = "allow"
 # TODO: remove later https://github.com/rust-lang/rust-clippy/issues/12436
 mixed_attributes_style = "allow"
+too_long_first_doc_paragraph = "allow"
 
 [workspace.lints.rustdoc]
 private_intra_doc_links = "allow"

ci/Dockerfile

@@ -77,7 +77,7 @@ RUN cargo binstall -y --locked --no-symlinks cargo-llvm-cov cargo-nextest cargo-
     && rm -rf "/root/.cargo/registry/cache" \
     && rm -rf "/root/.cargo/git/db" \
     && cargo uninstall cargo-cache
-RUN cargo install cargo-dylint@3.1.0 dylint-link@3.1.0
+RUN cargo install cargo-dylint@3.2.1 dylint-link@3.2.1
 
 # install risedev
 COPY <<EOF /root/.cargo/bin/risedev

ci/build-ci-image.sh

@@ -10,7 +10,7 @@ cat ../rust-toolchain
 # shellcheck disable=SC2155
 
 # REMEMBER TO ALSO UPDATE ci/docker-compose.yml
-export BUILD_ENV_VERSION=v20240911
+export BUILD_ENV_VERSION=v20241030
 
 export BUILD_TAG="public.ecr.aws/w1p7b4n3/rw-build-env:${BUILD_ENV_VERSION}"
 

ci/docker-compose.yml

@@ -72,7 +72,7 @@ services:
       retries: 5
 
   source-test-env:
-    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240911
+    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20241030
     depends_on:
       - mysql
       - sqlserver-server
@@ -86,7 +86,7 @@ services:
       - ..:/risingwave
 
   sink-test-env:
-    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240911
+    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20241030
     depends_on:
       - mysql
       - db
@@ -109,7 +109,7 @@ services:
 
 
   rw-build-env:
-    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240911
+    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20241030
     volumes:
       - ..:/risingwave
 
@@ -122,7 +122,7 @@ services:
       - ..:/risingwave
 
   ci-flamegraph-env:
-    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240911
+    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20241030
     # NOTE(kwannoel): This is used in order to permit
     # syscalls for `nperf` (perf_event_open),
     # so it can do CPU profiling.
@@ -133,7 +133,7 @@ services:
       - ..:/risingwave
 
   regress-test-env:
-    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240911
+    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20241030
     depends_on:
       db:
         condition: service_healthy

ci/rust-toolchain

@@ -4,4 +4,4 @@
 # 3. (optional) **follow the instructions in lints/README.md** to update the toolchain and dependencies for lints
 
 [toolchain]
-channel = "nightly-2024-07-19"
+channel = "nightly-2024-10-11"

ci/scripts/build.sh

@@ -27,17 +27,6 @@ if [[ "$profile" != "ci-dev" ]] && [[ "$profile" != "ci-release" ]]; then
     exit 1
 fi
 
-echo "--- Rust cargo-sort check"
-cargo sort --check --workspace --grouped
-
-# Disable hakari until we make sure it's useful
-# echo "--- Rust cargo-hakari check"
-# cargo hakari generate --diff
-# cargo hakari verify
-
-echo "--- Rust format check"
-cargo fmt --all -- --check
-
 echo "--- Build Rust components"
 
 if [[ "$profile" == "ci-dev" ]]; then

ci/scripts/check.sh

@@ -23,6 +23,17 @@ configure_static_openssl
 echo "--- Run trailing spaces check"
 scripts/check/check-trailing-spaces.sh
 
+echo "--- Rust cargo-sort check"
+cargo sort --check --workspace --grouped
+
+# Disable hakari until we make sure it's useful
+# echo "--- Rust cargo-hakari check"
+# cargo hakari generate --diff
+# cargo hakari verify
+
+echo "--- Rust format check"
+cargo fmt --all -- --check
+
 echo "--- Run clippy check (dev, all features)"
 cargo clippy --all-targets --all-features --locked -- -D warnings
 

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