refactor: rename FlowWorkerManager to FlowStreamingEngine

chore: better error variant
chore: use better error variant
2025-12-27 08:29:59 +00:00 · 2025-04-23 11:36:30 +08:00 · 2025-04-22 19:52:19 +08:00 · 2025-04-22 17:30:24 +08:00 · 2025-04-22 17:09:42 +08:00 · 2025-04-22 17:09:42 +08:00
349 changed files with 8316 additions and 15429 deletions
--- a/.coderabbit.yaml
+++ b/.coderabbit.yaml
@@ -0,0 +1,15 @@
+# yaml-language-server: $schema=https://coderabbit.ai/integrations/schema.v2.json
+language: "en-US"
+early_access: false
+reviews:
+  profile: "chill"
+  request_changes_workflow: false
+  high_level_summary: true
+  poem: true
+  review_status: true
+  collapse_walkthrough: false
+  auto_review:
+    enabled: false
+    drafts: false
+chat:
+  auto_reply: true
--- a/.github/scripts/create-version.sh
+++ b/.github/scripts/create-version.sh
@@ -10,17 +10,17 @@ set -e
 function create_version() {
  # Read from envrionment variables.
  if [ -z "$GITHUB_EVENT_NAME" ]; then
-      echo "GITHUB_EVENT_NAME is empty" >&2
+      echo "GITHUB_EVENT_NAME is empty"
      exit 1
  fi

  if [ -z "$NEXT_RELEASE_VERSION" ]; then
-      echo "NEXT_RELEASE_VERSION is empty, use version from Cargo.toml" >&2
-      export NEXT_RELEASE_VERSION=$(grep '^version = ' Cargo.toml | cut -d '"' -f 2 | head -n 1)
+      echo "NEXT_RELEASE_VERSION is empty"
+      exit 1
  fi

  if [ -z "$NIGHTLY_RELEASE_PREFIX" ]; then
-      echo "NIGHTLY_RELEASE_PREFIX is empty" >&2
+      echo "NIGHTLY_RELEASE_PREFIX is empty"
      exit 1
  fi

@@ -35,7 +35,7 @@ function create_version() {
  # It will be like 'dev-2023080819-f0e7216c'.
  if [ "$NEXT_RELEASE_VERSION" = dev ]; then
    if [ -z "$COMMIT_SHA" ]; then
-      echo "COMMIT_SHA is empty in dev build" >&2
+      echo "COMMIT_SHA is empty in dev build"
      exit 1
    fi
    echo "dev-$(date "+%Y%m%d-%s")-$(echo "$COMMIT_SHA" | cut -c1-8)"
@@ -45,7 +45,7 @@ function create_version() {
  # Note: Only output 'version=xxx' to stdout when everything is ok, so that it can be used in GitHub Actions Outputs.
  if [ "$GITHUB_EVENT_NAME" = push ]; then
    if [ -z "$GITHUB_REF_NAME" ]; then
-      echo "GITHUB_REF_NAME is empty in push event" >&2
+      echo "GITHUB_REF_NAME is empty in push event"
      exit 1
    fi
    echo "$GITHUB_REF_NAME"
@@ -54,7 +54,7 @@ function create_version() {
  elif [ "$GITHUB_EVENT_NAME" = schedule ]; then
    echo "$NEXT_RELEASE_VERSION-$NIGHTLY_RELEASE_PREFIX-$(date "+%Y%m%d")"
  else
-    echo "Unsupported GITHUB_EVENT_NAME: $GITHUB_EVENT_NAME" >&2
+    echo "Unsupported GITHUB_EVENT_NAME: $GITHUB_EVENT_NAME"
    exit 1
  fi
 }
--- a/.github/scripts/update-dev-builder-version.sh
+++ b/.github/scripts/update-dev-builder-version.sh
@@ -1,37 +0,0 @@
-#!/bin/bash
-
-DEV_BUILDER_IMAGE_TAG=$1
-
-update_dev_builder_version() {
-  if [ -z "$DEV_BUILDER_IMAGE_TAG" ]; then
-    echo "Error: Should specify the dev-builder image tag" 
-    exit 1
-  fi
-
-  # Configure Git configs.
-  git config --global user.email greptimedb-ci@greptime.com
-  git config --global user.name greptimedb-ci
-
-  # Checkout a new branch.
-  BRANCH_NAME="ci/update-dev-builder-$(date +%Y%m%d%H%M%S)"
-  git checkout -b $BRANCH_NAME
-
-  # Update the dev-builder image tag in the Makefile.
-  gsed -i "s/DEV_BUILDER_IMAGE_TAG ?=.*/DEV_BUILDER_IMAGE_TAG ?= ${DEV_BUILDER_IMAGE_TAG}/g" Makefile
-
-  # Commit the changes.
-  git add Makefile
-  git commit -m "ci: update dev-builder image tag"
-  git push origin $BRANCH_NAME
-
-  # Create a Pull Request.
-  gh pr create \
-      --title "ci: update dev-builder image tag" \
-      --body "This PR updates the dev-builder image tag" \
-      --base main \
-      --head $BRANCH_NAME \
-      --reviewer zyy17 \
-      --reviewer daviderli614
-}
-
-update_dev_builder_version
--- a/.github/workflows/develop.yml
+++ b/.github/workflows/develop.yml
@@ -22,7 +22,6 @@ concurrency:

 jobs:
  check-typos-and-docs:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Check typos and docs
    runs-on: ubuntu-latest
    steps:
@@ -37,7 +36,6 @@ jobs:
          || (echo "'config/config.md' is not up-to-date, please run 'make config-docs'." && exit 1)

  license-header-check:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    runs-on: ubuntu-latest
    name: Check License Header
    steps:
@@ -47,7 +45,6 @@ jobs:
      - uses: korandoru/hawkeye@v5

  check:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Check
    runs-on: ${{ matrix.os }}
    strategy:
@@ -74,7 +71,6 @@ jobs:
        run: cargo check --locked --workspace --all-targets

  toml:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Toml Check
    runs-on: ubuntu-latest
    timeout-minutes: 60
@@ -89,7 +85,6 @@ jobs:
        run: taplo format --check

  build:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Build GreptimeDB binaries
    runs-on: ${{ matrix.os }}
    strategy:
@@ -132,7 +127,6 @@ jobs:
          version: current

  fuzztest:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Fuzz Test
    needs: build
    runs-on: ubuntu-latest
@@ -189,7 +183,6 @@ jobs:
          max-total-time: 120

  unstable-fuzztest:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Unstable Fuzz Test
    needs: build-greptime-ci
    runs-on: ubuntu-latest
@@ -251,7 +244,6 @@ jobs:
          retention-days: 3

  build-greptime-ci:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Build GreptimeDB binary (profile-CI)
    runs-on: ${{ matrix.os }}
    strategy:
@@ -293,7 +285,6 @@ jobs:
          version: current

  distributed-fuzztest:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Fuzz Test (Distributed, ${{ matrix.mode.name }}, ${{ matrix.target }})
    runs-on: ubuntu-latest
    needs:  build-greptime-ci
@@ -425,7 +416,6 @@ jobs:
          docker system prune -f

  distributed-fuzztest-with-chaos:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Fuzz Test with Chaos (Distributed, ${{ matrix.mode.name }}, ${{ matrix.target }})
    runs-on: ubuntu-latest
    needs:  build-greptime-ci
@@ -573,7 +563,6 @@ jobs:
          docker system prune -f

  sqlness:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Sqlness Test (${{ matrix.mode.name }})
    needs: build
    runs-on: ${{ matrix.os }}
@@ -620,7 +609,6 @@ jobs:
          retention-days: 3

  fmt:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Rustfmt
    runs-on: ubuntu-latest
    timeout-minutes: 60
@@ -638,7 +626,6 @@ jobs:
        run: make fmt-check

  clippy:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Clippy
    runs-on: ubuntu-latest
    timeout-minutes: 60
@@ -664,7 +651,6 @@ jobs:
        run: make clippy

  conflict-check:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Check for conflict
    runs-on: ubuntu-latest
    steps:
@@ -675,7 +661,7 @@ jobs:
        uses: olivernybroe/action-conflict-finder@v4.0

  test:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' && github.event_name != 'merge_group' }}
+    if: github.event_name != 'merge_group'
    runs-on: ubuntu-22.04-arm
    timeout-minutes: 60
    needs:  [conflict-check, clippy, fmt]
@@ -727,7 +713,7 @@ jobs:
          UNITTEST_LOG_DIR: "__unittest_logs"

  coverage:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' && github.event_name == 'merge_group' }}
+    if: github.event_name == 'merge_group'
    runs-on: ubuntu-22.04-8-cores
    timeout-minutes: 60
    steps:
@@ -787,7 +773,6 @@ jobs:
          verbose: true

  # compat:
-  #   if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
  #   name: Compatibility Test
  #   needs: build
  #   runs-on: ubuntu-22.04
--- a/.github/workflows/nightly-ci.yml
+++ b/.github/workflows/nightly-ci.yml
@@ -117,16 +117,16 @@ jobs:
    name: Run clean build on Linux
    runs-on: ubuntu-latest
    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
-    timeout-minutes: 45
+    timeout-minutes: 60
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0
          persist-credentials: false
-      - uses: cachix/install-nix-action@v31
+      - uses: cachix/install-nix-action@v27
        with:
          nix_path: nixpkgs=channel:nixos-24.11
-      - run: nix develop --command cargo build --bin greptime
+      - run: nix develop --command cargo build

  check-status:
    name: Check status
--- a/.github/workflows/release-dev-builder-images.yaml
+++ b/.github/workflows/release-dev-builder-images.yaml
@@ -24,19 +24,11 @@ on:
        description: Release dev-builder-android image
        required: false
        default: false
-      update_dev_builder_image_tag:
-        type: boolean
-        description: Update the DEV_BUILDER_IMAGE_TAG in Makefile and create a PR
-        required: false
-        default: false

 jobs:
  release-dev-builder-images:
    name: Release dev builder images
-    # The jobs are triggered by the following events:
-    # 1. Manually triggered workflow_dispatch event
-    # 2. Push event when the PR that modifies the `rust-toolchain.toml` or `docker/dev-builder/**` is merged to main
-    if: ${{ github.event_name == 'push' || inputs.release_dev_builder_ubuntu_image || inputs.release_dev_builder_centos_image || inputs.release_dev_builder_android_image }}
+    if: ${{ inputs.release_dev_builder_ubuntu_image || inputs.release_dev_builder_centos_image || inputs.release_dev_builder_android_image }} # Only manually trigger this job.
    runs-on: ubuntu-latest
    outputs:
      version: ${{ steps.set-version.outputs.version }}
@@ -65,9 +57,9 @@ jobs:
          version: ${{ env.VERSION }}
          dockerhub-image-registry-username: ${{ secrets.DOCKERHUB_USERNAME }}
          dockerhub-image-registry-token: ${{ secrets.DOCKERHUB_TOKEN }}
-          build-dev-builder-ubuntu: ${{ inputs.release_dev_builder_ubuntu_image || github.event_name == 'push' }}
-          build-dev-builder-centos: ${{ inputs.release_dev_builder_centos_image || github.event_name == 'push' }}
-          build-dev-builder-android: ${{ inputs.release_dev_builder_android_image || github.event_name == 'push' }}
+          build-dev-builder-ubuntu: ${{ inputs.release_dev_builder_ubuntu_image }}
+          build-dev-builder-centos: ${{ inputs.release_dev_builder_centos_image }}
+          build-dev-builder-android: ${{ inputs.release_dev_builder_android_image }}

  release-dev-builder-images-ecr:
    name: Release dev builder images to AWS ECR
@@ -93,7 +85,7 @@ jobs:

      - name: Push dev-builder-ubuntu image
        shell: bash
-        if: ${{ inputs.release_dev_builder_ubuntu_image || github.event_name == 'push' }}
+        if: ${{ inputs.release_dev_builder_ubuntu_image }}
        env:
          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -114,7 +106,7 @@ jobs:

      - name: Push dev-builder-centos image
        shell: bash
-        if: ${{ inputs.release_dev_builder_centos_image || github.event_name == 'push' }}
+        if: ${{ inputs.release_dev_builder_centos_image }}
        env:
          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -135,7 +127,7 @@ jobs:

      - name: Push dev-builder-android image
        shell: bash
-        if: ${{ inputs.release_dev_builder_android_image || github.event_name == 'push' }}
+        if: ${{ inputs.release_dev_builder_android_image }}
        env:
          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -170,7 +162,7 @@ jobs:

      - name: Push dev-builder-ubuntu image
        shell: bash
-        if: ${{ inputs.release_dev_builder_ubuntu_image || github.event_name == 'push' }}
+        if: ${{ inputs.release_dev_builder_ubuntu_image }}
        env:
          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -184,7 +176,7 @@ jobs:

      - name: Push dev-builder-centos image
        shell: bash
-        if: ${{ inputs.release_dev_builder_centos_image || github.event_name == 'push' }}
+        if: ${{ inputs.release_dev_builder_centos_image }}
        env:
          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -198,7 +190,7 @@ jobs:

      - name: Push dev-builder-android image
        shell: bash
-        if: ${{ inputs.release_dev_builder_android_image || github.event_name == 'push' }}
+        if: ${{ inputs.release_dev_builder_android_image }}
        env:
          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -209,24 +201,3 @@ jobs:
            quay.io/skopeo/stable:latest \
            copy -a docker://docker.io/$IMAGE_NAMESPACE/dev-builder-android:$IMAGE_VERSION \
            docker://$ACR_IMAGE_REGISTRY/$IMAGE_NAMESPACE/dev-builder-android:$IMAGE_VERSION
-  
-  update-dev-builder-image-tag:
-    name: Update dev-builder image tag
-    runs-on: ubuntu-latest
-    permissions:
-      contents: write
-      pull-requests: write
-    if: ${{ github.event_name == 'push' || inputs.update_dev_builder_image_tag }}
-    needs: [
-      release-dev-builder-images
-    ]
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@v4
-
-      - name: Update dev-builder image tag
-        shell: bash
-        env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-        run: |
-          ./.github/scripts/update-dev-builder-version.sh ${{ needs.release-dev-builder-images.outputs.version }}
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -90,6 +90,8 @@ env:

  # The scheduled version is '${{ env.NEXT_RELEASE_VERSION }}-nightly-YYYYMMDD', like v0.2.0-nigthly-20230313;
  NIGHTLY_RELEASE_PREFIX: nightly
+  # Note: The NEXT_RELEASE_VERSION should be modified manually by every formal release.
+  NEXT_RELEASE_VERSION: v0.14.0

 jobs:
  allocate-runners:
@@ -133,6 +135,7 @@ jobs:
        env:
          GITHUB_EVENT_NAME: ${{ github.event_name }}
          GITHUB_REF_NAME: ${{ github.ref_name }}
+          NEXT_RELEASE_VERSION: ${{ env.NEXT_RELEASE_VERSION }}
          NIGHTLY_RELEASE_PREFIX: ${{ env.NIGHTLY_RELEASE_PREFIX }}

      - name: Allocate linux-amd64 runner
@@ -464,29 +467,6 @@ jobs:
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
          DOCS_REPO_TOKEN: ${{ secrets.DOCS_REPO_TOKEN }}

-  bump-website-version:
-    name: Bump website version
-    if: ${{ github.event_name == 'push' || github.event_name == 'schedule' }}
-    needs: [allocate-runners]
-    runs-on: ubuntu-latest
-    # Permission reference: https://docs.github.com/en/actions/using-jobs/assigning-permissions-to-jobs
-    permissions:
-      issues: write # Allows the action to create issues for cyborg.
-      contents: write # Allows the action to create a release.
-    steps:
-      - uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-          persist-credentials: false
-      - uses: ./.github/actions/setup-cyborg
-      - name: Bump website version
-        working-directory: cyborg
-        run: pnpm tsx bin/bump-website-version.ts
-        env:
-          VERSION: ${{ needs.allocate-runners.outputs.version }}
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-          WEBSITE_REPO_TOKEN: ${{ secrets.WEBSITE_REPO_TOKEN }}
-
  notification:
    if: ${{ github.repository == 'GreptimeTeam/greptimedb' && (github.event_name == 'push' || github.event_name == 'schedule') && always() }}
    name: Send notification to Greptime team
--- a/.gitignore
+++ b/.gitignore
@@ -28,7 +28,6 @@ debug/
 # Logs
 **/__unittest_logs
 logs/
-!grafana/dashboards/logs/

 # cpython's generated python byte code
 **/__pycache__/
--- a/Cargo.lock
+++ b/Cargo.lock
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -68,16 +68,15 @@ members = [
 resolver = "2"

 [workspace.package]
-version = "0.15.0"
+version = "0.14.0"
 edition = "2021"
 license = "Apache-2.0"

 [workspace.lints]
+clippy.print_stdout = "warn"
+clippy.print_stderr = "warn"
 clippy.dbg_macro = "warn"
 clippy.implicit_clone = "warn"
-clippy.result_large_err = "allow"
-clippy.large_enum_variant = "allow"
-clippy.doc_overindented_list_items = "allow"
 rust.unknown_lints = "deny"
 rust.unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tokio_unstable)'] }

@@ -113,15 +112,15 @@ clap = { version = "4.4", features = ["derive"] }
 config = "0.13.0"
 crossbeam-utils = "0.8"
 dashmap = "6.1"
-datafusion = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
-datafusion-common = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
-datafusion-expr = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
-datafusion-functions = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
-datafusion-optimizer = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
-datafusion-physical-expr = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
-datafusion-physical-plan = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
-datafusion-sql = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
-datafusion-substrait = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
+datafusion = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "5bbedc6704162afb03478f56ffb629405a4e1220" }
+datafusion-common = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "5bbedc6704162afb03478f56ffb629405a4e1220" }
+datafusion-expr = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "5bbedc6704162afb03478f56ffb629405a4e1220" }
+datafusion-functions = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "5bbedc6704162afb03478f56ffb629405a4e1220" }
+datafusion-optimizer = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "5bbedc6704162afb03478f56ffb629405a4e1220" }
+datafusion-physical-expr = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "5bbedc6704162afb03478f56ffb629405a4e1220" }
+datafusion-physical-plan = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "5bbedc6704162afb03478f56ffb629405a4e1220" }
+datafusion-sql = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "5bbedc6704162afb03478f56ffb629405a4e1220" }
+datafusion-substrait = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "5bbedc6704162afb03478f56ffb629405a4e1220" }
 deadpool = "0.12"
 deadpool-postgres = "0.14"
 derive_builder = "0.20"
@@ -130,7 +129,7 @@ etcd-client = "0.14"
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "17a3550751c8b1e02ec16be40101d5f24dc255c3" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "e82b0158cd38d4021edb4e4c0ae77f999051e62f" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"
@@ -163,7 +162,7 @@ paste = "1.0"
 pin-project = "1.0"
 prometheus = { version = "0.13.3", features = ["process"] }
 promql-parser = { version = "0.5.1", features = ["ser"] }
-prost = { version = "0.13", features = ["no-recursion-limit"] }
+prost = "0.13"
 raft-engine = { version = "0.4.1", default-features = false }
 rand = "0.9"
 ratelimit = "0.10"
@@ -192,7 +191,7 @@ simd-json = "0.15"
 similar-asserts = "1.6.0"
 smallvec = { version = "1", features = ["serde"] }
 snafu = "0.8"
-sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "0cf6c04490d59435ee965edd2078e8855bd8471e", features = [
+sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "e98e6b322426a9d397a71efef17075966223c089", features = [
    "visitor",
    "serde",
 ] } # branch = "v0.54.x"
--- a/2
+++ b/2
@@ -8,7 +8,7 @@ CARGO_BUILD_OPTS := --locked
 IMAGE_REGISTRY ?= docker.io
 IMAGE_NAMESPACE ?= greptime
 IMAGE_TAG ?= latest
-DEV_BUILDER_IMAGE_TAG ?= 2025-04-15-1a517ec8-20250428023155
+DEV_BUILDER_IMAGE_TAG ?= 2024-12-25-a71b93dd-20250305072908
 BUILDX_MULTI_PLATFORM_BUILD ?= false
 BUILDX_BUILDER_NAME ?= gtbuilder
 BASE_IMAGE ?= ubuntu
--- a/README.md
+++ b/README.md
@@ -8,8 +8,6 @@

 <h2 align="center">Real-Time & Cloud-Native Observability  Database<br/>for metrics, logs, and traces</h2>

->  Delivers sub-second querying at PB scale and exceptional cost efficiency from edge to cloud.
-
 <div align="center">
 <h3 align="center">
  <a href="https://greptime.com/product/cloud">GreptimeCloud</a> |
@@ -51,77 +49,74 @@
 </div>

 - [Introduction](#introduction)
- [⭐ Key Features](#features)
- [Quick Comparison](#quick-comparison)
- [Architecture](#architecture)
- [Try GreptimeDB](#try-greptimedb)
+- [**Features: Why GreptimeDB**](#why-greptimedb)
+- [Architecture](https://docs.greptime.com/contributor-guide/overview/#architecture)
+- [Try it for free](#try-greptimedb)
 - [Getting Started](#getting-started)
- [Build From Source](#build-from-source)
- [Tools & Extensions](#tools--extensions)
 - [Project Status](#project-status)
- [Community](#community)
+- [Join the community](#community)
+  - [Contributing](#contributing)
+- [Tools & Extensions](#tools--extensions)
 - [License](#license)
- [Commercial Support](#commercial-support)
- [Contributing](#contributing)
 - [Acknowledgement](#acknowledgement)

 ## Introduction

-**GreptimeDB** is an open-source, cloud-native database purpose-built for the unified collection and analysis of observability data (metrics, logs, and traces). Whether you’re operating on the edge, in the cloud, or across hybrid environments, GreptimeDB empowers real-time insights at massive scale — all in one system.
+**GreptimeDB** is an open-source, cloud-native, unified & cost-effective observability database for **Metrics**, **Logs**, and **Traces**. You can gain real-time insights from Edge to Cloud at Any Scale.

-## Features
+## News

-|   Feature  | Description |
-| --------- | ----------- |
-| [Unified Observability Data](https://docs.greptime.com/user-guide/concepts/why-greptimedb) | Store metrics, logs, and traces as timestamped, contextual wide events. Query via [SQL](https://docs.greptime.com/user-guide/query-data/sql), [PromQL](https://docs.greptime.com/user-guide/query-data/promql), and [streaming](https://docs.greptime.com/user-guide/flow-computation/overview). |
-| [High Performance & Cost Effective](https://docs.greptime.com/user-guide/manage-data/data-index) | Written in Rust, with a distributed query engine, [rich indexing](https://docs.greptime.com/user-guide/manage-data/data-index), and optimized columnar storage, delivering sub-second responses at PB scale. |
-| [Cloud-Native Architecture](https://docs.greptime.com/user-guide/concepts/architecture) | Designed for [Kubernetes](https://docs.greptime.com/user-guide/deployments/deploy-on-kubernetes/greptimedb-operator-management), with compute/storage separation, native object storage (AWS S3, Azure Blob, etc.) and seamless cross-cloud access. |
-| [Developer-Friendly](https://docs.greptime.com/user-guide/protocols/overview) | Access via SQL/PromQL interfaces, REST API, MySQL/PostgreSQL protocols, and popular ingestion [protocols](https://docs.greptime.com/user-guide/protocols/overview). |
-| [Flexible Deployment](https://docs.greptime.com/user-guide/deployments/overview) | Deploy anywhere: edge (including ARM/[Android](https://docs.greptime.com/user-guide/deployments/run-on-android)) or cloud, with unified APIs and efficient data sync. |
+**[GreptimeDB tops JSONBench's billion-record cold run test!](https://greptime.com/blogs/2025-03-18-jsonbench-greptimedb-performance)**

-Learn more in [Why GreptimeDB](https://docs.greptime.com/user-guide/concepts/why-greptimedb) and [Observability 2.0 and the Database for It](https://greptime.com/blogs/2025-04-25-greptimedb-observability2-new-database).
+## Why GreptimeDB

-## Quick Comparison
+Our core developers have been building observability data platforms for years. Based on our best practices, GreptimeDB was born to give you:

-| Feature                         | GreptimeDB            | Traditional TSDB   | Log Stores      |
-|----------------------------------|-----------------------|--------------------|-----------------|
-| Data Types                      | Metrics, Logs, Traces | Metrics only       | Logs only       |
-| Query Language                  | SQL, PromQL, Streaming|  Custom/PromQL     | Custom/DSL      |
-| Deployment                      | Edge + Cloud          | Cloud/On-prem      | Mostly central  |
-| Indexing & Performance          | PB-Scale, Sub-second  | Varies             | Varies          |
-| Integration                     | REST, SQL, Common protocols | Varies     | Varies          |
+* **Unified Processing of Observability Data**

-**Performance:**
-* [GreptimeDB tops JSONBench's billion-record cold run test!](https://greptime.com/blogs/2025-03-18-jsonbench-greptimedb-performance)
-* [TSBS Benchmark](https://github.com/GreptimeTeam/greptimedb/tree/main/docs/benchmarks/tsbs)
+  A unified database that treats metrics, logs, and traces as timestamped wide events with context, supporting [SQL](https://docs.greptime.com/user-guide/query-data/sql)/[PromQL](https://docs.greptime.com/user-guide/query-data/promql) queries and [stream processing](https://docs.greptime.com/user-guide/flow-computation/overview) to simplify complex data stacks.

-Read [more benchmark reports](https://docs.greptime.com/user-guide/concepts/features-that-you-concern#how-is-greptimedbs-performance-compared-to-other-solutions).
+* **High Performance and Cost-effective**

-## Architecture
+   Written in Rust, combines a distributed query engine with [rich indexing](https://docs.greptime.com/user-guide/manage-data/data-index) (inverted, fulltext, skip data, and vector) and optimized columnar storage to deliver sub-second responses on petabyte-scale data and high-cost efficiency.

-* Read the [architecture](https://docs.greptime.com/contributor-guide/overview/#architecture) document.
-* [DeepWiki](https://deepwiki.com/GreptimeTeam/greptimedb/1-overview) provides an in-depth look at GreptimeDB:
-  <img alt="GreptimeDB System Overview" src="docs/architecture.png">
+* **Cloud-native Distributed Database**
+
+  Built for [Kubernetes](https://docs.greptime.com/user-guide/deployments/deploy-on-kubernetes/greptimedb-operator-management). GreptimeDB achieves seamless scalability with its [cloud-native architecture](https://docs.greptime.com/user-guide/concepts/architecture) of separated compute and storage, built on object storage (AWS S3, Azure Blob Storage, etc.) while enabling cross-cloud deployment through a unified data access layer.
+
+* **Developer-Friendly**
+
+  Access standardized SQL/PromQL interfaces through built-in web dashboard, REST API, and MySQL/PostgreSQL protocols. Supports widely adopted data ingestion [protocols](https://docs.greptime.com/user-guide/protocols/overview) for seamless migration and integration.
+
+* **Flexible Deployment Options**
+
+  Deploy GreptimeDB anywhere from ARM-based edge devices to cloud environments with unified APIs and bandwidth-efficient data synchronization. Query edge and cloud data seamlessly through identical APIs. [Learn how to run on Android](https://docs.greptime.com/user-guide/deployments/run-on-android/).
+
+For more detailed info please read  [Why GreptimeDB](https://docs.greptime.com/user-guide/concepts/why-greptimedb).

 ## Try GreptimeDB

 ### 1. [Live Demo](https://greptime.com/playground)

-Experience GreptimeDB directly in your browser.
+Try out the features of GreptimeDB right from your browser.

 ### 2. [GreptimeCloud](https://console.greptime.cloud/)

 Start instantly with a free cluster.

-### 3. Docker (Local Quickstart)
+### 3. Docker Image
+
+To install GreptimeDB locally, the recommended way is via Docker:

 ```shell
 docker pull greptime/greptimedb
 ```

+Start a GreptimeDB container with:
+
 ```shell
 docker run -p 127.0.0.1:4000-4003:4000-4003 \
-  -v "$(pwd)/greptimedb:/greptimedb_data" \
+  -v "$(pwd)/greptimedb:./greptimedb_data" \
  --name greptime --rm \
  greptime/greptimedb:latest standalone start \
  --http-addr 0.0.0.0:4000 \
@@ -129,90 +124,114 @@ docker run -p 127.0.0.1:4000-4003:4000-4003 \
  --mysql-addr 0.0.0.0:4002 \
  --postgres-addr 0.0.0.0:4003
 ```
-Dashboard: [http://localhost:4000/dashboard](http://localhost:4000/dashboard)  
-[Full Install Guide](https://docs.greptime.com/getting-started/installation/overview)

-**Troubleshooting:**
-* Cannot connect to the database? Ensure that ports `4000`, `4001`, `4002`, and `4003` are not blocked by a firewall or used by other services.
-* Failed to start? Check the container logs with `docker logs greptime` for further details.
+Access the dashboard via `http://localhost:4000/dashboard`.
+
+Read more about [Installation](https://docs.greptime.com/getting-started/installation/overview) on docs.

 ## Getting Started

- [Quickstart](https://docs.greptime.com/getting-started/quick-start)
- [User Guide](https://docs.greptime.com/user-guide/overview)
- [Demo Scenes](https://github.com/GreptimeTeam/demo-scene)
- [FAQ](https://docs.greptime.com/faq-and-others/faq)
+* [Quickstart](https://docs.greptime.com/getting-started/quick-start)
+* [User Guide](https://docs.greptime.com/user-guide/overview)
+* [Demos](https://github.com/GreptimeTeam/demo-scene)
+* [FAQ](https://docs.greptime.com/faq-and-others/faq)

-## Build From Source
+## Build
+
+Check the prerequisite:

-**Prerequisites:**
 * [Rust toolchain](https://www.rust-lang.org/tools/install) (nightly)
 * [Protobuf compiler](https://grpc.io/docs/protoc-installation/) (>= 3.15)
 * C/C++ building essentials, including `gcc`/`g++`/`autoconf` and glibc library (eg. `libc6-dev` on Ubuntu and `glibc-devel` on Fedora)
 * Python toolchain (optional): Required only if using some test scripts.

-**Build and Run:**
-```bash
+Build GreptimeDB binary:
+
+```shell
 make
+```
+
+Run a standalone server:
+
+```shell
 cargo run -- standalone start
 ```

 ## Tools & Extensions

- **Kubernetes:** [GreptimeDB Operator](https://github.com/GrepTimeTeam/greptimedb-operator)
- **Helm Charts:** [Greptime Helm Charts](https://github.com/GreptimeTeam/helm-charts)
- **Dashboard:** [Web UI](https://github.com/GreptimeTeam/dashboard)
- **SDKs/Ingester:** [Go](https://github.com/GreptimeTeam/greptimedb-ingester-go), [Java](https://github.com/GreptimeTeam/greptimedb-ingester-java), [C++](https://github.com/GreptimeTeam/greptimedb-ingester-cpp), [Erlang](https://github.com/GreptimeTeam/greptimedb-ingester-erl), [Rust](https://github.com/GreptimeTeam/greptimedb-ingester-rust), [JS](https://github.com/GreptimeTeam/greptimedb-ingester-js)
- **Grafana**: [Official Dashboard](https://github.com/GreptimeTeam/greptimedb/blob/main/grafana/README.md)
+### Kubernetes
+
+- [GreptimeDB Operator](https://github.com/GrepTimeTeam/greptimedb-operator)
+
+### Dashboard
+
+- [The dashboard UI for GreptimeDB](https://github.com/GreptimeTeam/dashboard)
+
+### SDK
+
+- [GreptimeDB Go Ingester](https://github.com/GreptimeTeam/greptimedb-ingester-go)
+- [GreptimeDB Java Ingester](https://github.com/GreptimeTeam/greptimedb-ingester-java)
+- [GreptimeDB C++ Ingester](https://github.com/GreptimeTeam/greptimedb-ingester-cpp)
+- [GreptimeDB Erlang Ingester](https://github.com/GreptimeTeam/greptimedb-ingester-erl)
+- [GreptimeDB Rust Ingester](https://github.com/GreptimeTeam/greptimedb-ingester-rust)
+- [GreptimeDB JavaScript Ingester](https://github.com/GreptimeTeam/greptimedb-ingester-js)
+
+### Grafana Dashboard
+
+Our official Grafana dashboard for monitoring GreptimeDB is available at [grafana](grafana/README.md) directory.

 ## Project Status

-> **Status:** Beta.  
-> **GA (v1.0):** Targeted for mid 2025.
+GreptimeDB is currently in Beta. We are targeting GA (General Availability) with v1.0 release by Early 2025.

- Being used in production by early adopters
- Stable, actively maintained, with regular releases ([version info](https://docs.greptime.com/nightly/reference/about-greptimedb-version))
- Suitable for evaluation and pilot deployments
+While in Beta, GreptimeDB is already:
+
+* Being used in production by early adopters
+* Actively maintained with regular releases, [about version number](https://docs.greptime.com/nightly/reference/about-greptimedb-version)
+* Suitable for testing and evaluation

 For production use, we recommend using the latest stable release.
-[![Star History Chart](https://api.star-history.com/svg?repos=GreptimeTeam/GreptimeDB&type=Date)](https://www.star-history.com/#GreptimeTeam/GreptimeDB&Date)
-
-If you find this project useful, a ⭐ would mean a lot to us!
-<img alt="Known Users" src="https://greptime.com/logo/img/users.png"/>

 ## Community

-We invite you to engage and contribute!
+Our core team is thrilled to see you participate in any ways you like. When you are stuck, try to
+ask for help by filling an issue with a detailed description of what you were trying to do
+and what went wrong. If you have any questions or if you would like to get involved in our
+community, please check out:

- [Slack](https://greptime.com/slack)
- [Discussions](https://github.com/GreptimeTeam/greptimedb/discussions)
- [Official Website](https://greptime.com/)
- [Blog](https://greptime.com/blogs/)
- [LinkedIn](https://www.linkedin.com/company/greptime/)
- [Twitter](https://twitter.com/greptime)
+- GreptimeDB Community on [Slack](https://greptime.com/slack)
+- GreptimeDB [GitHub Discussions forum](https://github.com/GreptimeTeam/greptimedb/discussions)
+- Greptime official [website](https://greptime.com)

-## License
+In addition, you may:

-GreptimeDB is licensed under the [Apache License 2.0](https://apache.org/licenses/LICENSE-2.0.txt).
+- View our official [Blog](https://greptime.com/blogs/)
+- Connect us with [Linkedin](https://www.linkedin.com/company/greptime/)
+- Follow us on [Twitter](https://twitter.com/greptime)

 ## Commercial Support

-Running GreptimeDB in your organization?  
-We offer enterprise add-ons, services, training, and consulting.  
-[Contact us](https://greptime.com/contactus) for details.
+If you are running GreptimeDB OSS in your organization, we offer additional
+enterprise add-ons, installation services, training, and consulting. [Contact
+us](https://greptime.com/contactus) and we will reach out to you with more
+detail of our commercial license.
+
+## License
+
+GreptimeDB uses the [Apache License 2.0](https://apache.org/licenses/LICENSE-2.0.txt) to strike a balance between
+open contributions and allowing you to use the software however you want.

 ## Contributing

- Read our [Contribution Guidelines](https://github.com/GreptimeTeam/greptimedb/blob/main/CONTRIBUTING.md).
- Explore [Internal Concepts](https://docs.greptime.com/contributor-guide/overview.html) and [DeepWiki](https://deepwiki.com/GreptimeTeam/greptimedb).
- Pick up a [good first issue](https://github.com/GreptimeTeam/greptimedb/issues?q=is%3Aissue+is%3Aopen+label%3A%22good+first+issue%22) and join the #contributors [Slack](https://greptime.com/slack) channel.
+Please refer to [contribution guidelines](CONTRIBUTING.md) and [internal concepts docs](https://docs.greptime.com/contributor-guide/overview.html) for more information.

 ## Acknowledgement

-Special thanks to all contributors! See [AUTHORS.md](https://github.com/GreptimeTeam/greptimedb/blob/main/AUTHOR.md).
+Special thanks to all the contributors who have propelled GreptimeDB forward. For a complete list of contributors, please refer to [AUTHOR.md](AUTHOR.md).

- Uses [Apache Arrow™](https://arrow.apache.org/) (memory model)
- [Apache Parquet™](https://parquet.apache.org/) (file storage)
- [Apache Arrow DataFusion™](https://arrow.apache.org/datafusion/) (query engine)
- [Apache OpenDAL™](https://opendal.apache.org/) (data access abstraction)
- [etcd](https://etcd.io/) (meta service)
+- GreptimeDB uses [Apache Arrow™](https://arrow.apache.org/) as the memory model and [Apache Parquet™](https://parquet.apache.org/) as the persistent file format.
+- GreptimeDB's query engine is powered by [Apache Arrow DataFusion™](https://arrow.apache.org/datafusion/).
+- [Apache OpenDAL™](https://opendal.apache.org) gives GreptimeDB a very general and elegant data access abstraction layer.
+- GreptimeDB's meta service is based on [etcd](https://etcd.io/).
+
+<img alt="Known Users" src="https://greptime.com/logo/img/users.png"/>
--- a/config/config.md
+++ b/config/config.md
@@ -319,7 +319,6 @@
 | `selector` | String | `round_robin` | Datanode selector type.<br/>- `round_robin` (default value)<br/>- `lease_based`<br/>- `load_based`<br/>For details, please see "https://docs.greptime.com/developer-guide/metasrv/selector". |
 | `use_memory_store` | Bool | `false` | Store data in memory. |
 | `enable_region_failover` | Bool | `false` | Whether to enable region failover.<br/>This feature is only available on GreptimeDB running on cluster mode and<br/>- Using Remote WAL<br/>- Using shared storage (e.g., s3). |
-| `allow_region_failover_on_local_wal` | Bool | `false` | Whether to allow region failover on local WAL.<br/>**This option is not recommended to be set to true, because it may lead to data loss during failover.** |
 | `node_max_idle_time` | String | `24hours` | Max allowed idle time before removing node info from metasrv memory. |
 | `enable_telemetry` | Bool | `true` | Whether to enable greptimedb telemetry. Enabled by default. |
 | `runtime` | -- | -- | The runtime options. |
--- a/config/metasrv.example.toml
+++ b/config/metasrv.example.toml
@@ -50,10 +50,6 @@ use_memory_store = false
 ## - Using shared storage (e.g., s3).
 enable_region_failover = false

-## Whether to allow region failover on local WAL.
-## **This option is not recommended to be set to true, because it may lead to data loss during failover.**
-allow_region_failover_on_local_wal = false
-
 ## Max allowed idle time before removing node info from metasrv memory.
 node_max_idle_time = "24hours"

--- a/cyborg/bin/bump-website-version.ts
+++ b/cyborg/bin/bump-website-version.ts
@@ -1,57 +0,0 @@
-/*
- * Copyright 2023 Greptime Team
- *
- * 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 * as core from "@actions/core";
-import {obtainClient} from "@/common";
-
-async function triggerWorkflow(workflowId: string, version: string) {
-  const websiteClient = obtainClient("WEBSITE_REPO_TOKEN")
-  try {
-    await websiteClient.rest.actions.createWorkflowDispatch({
-      owner: "GreptimeTeam",
-      repo: "website",
-      workflow_id: workflowId,
-      ref: "main",
-      inputs: {
-        version,
-      },
-    });
-    console.log(`Successfully triggered ${workflowId} workflow with version ${version}`);
-  } catch (error) {
-    core.setFailed(`Failed to trigger workflow: ${error.message}`);
-  }
-}
-
-const version = process.env.VERSION;
-if (!version) {
-  core.setFailed("VERSION environment variable is required");
-  process.exit(1);
-}
-
-// Remove 'v' prefix if exists
-const cleanVersion = version.startsWith('v') ? version.slice(1) : version;
-
-if (cleanVersion.includes('nightly')) {
-  console.log('Nightly version detected, skipping workflow trigger.');
-  process.exit(0);
-}
-
-try {
-  triggerWorkflow('bump-patch-version.yml', cleanVersion);
-} catch (error) {
-  core.setFailed(`Error processing version: ${error.message}`);
-  process.exit(1);
-}
--- a/docs/architecture.png
+++ b/docs/architecture.png
--- a/flake.lock
+++ b/flake.lock
@@ -8,11 +8,11 @@
        "rust-analyzer-src": "rust-analyzer-src"
      },
      "locked": {
-        "lastModified": 1745735608,
-        "narHash": "sha256-L0jzm815XBFfF2wCFmR+M1CF+beIEFj6SxlqVKF59Ec=",
+        "lastModified": 1737613896,
+        "narHash": "sha256-ldqXIglq74C7yKMFUzrS9xMT/EVs26vZpOD68Sh7OcU=",
        "owner": "nix-community",
        "repo": "fenix",
-        "rev": "c39a78eba6ed2a022cc3218db90d485077101496",
+        "rev": "303a062fdd8e89f233db05868468975d17855d80",
        "type": "github"
      },
      "original": {
@@ -41,11 +41,11 @@
    },
    "nixpkgs": {
      "locked": {
-        "lastModified": 1745487689,
-        "narHash": "sha256-FQoi3R0NjQeBAsEOo49b5tbDPcJSMWc3QhhaIi9eddw=",
+        "lastModified": 1737569578,
+        "narHash": "sha256-6qY0pk2QmUtBT9Mywdvif0i/CLVgpCjMUn6g9vB+f3M=",
        "owner": "NixOS",
        "repo": "nixpkgs",
-        "rev": "5630cf13cceac06cefe9fc607e8dfa8fb342dde3",
+        "rev": "47addd76727f42d351590c905d9d1905ca895b82",
        "type": "github"
      },
      "original": {
@@ -65,11 +65,11 @@
    "rust-analyzer-src": {
      "flake": false,
      "locked": {
-        "lastModified": 1745694049,
-        "narHash": "sha256-fxvRYH/tS7hGQeg9zCVh5RBcSWT+JGJet7RA8Ss+rC0=",
+        "lastModified": 1737581772,
+        "narHash": "sha256-t1P2Pe3FAX9TlJsCZbmJ3wn+C4qr6aSMypAOu8WNsN0=",
        "owner": "rust-lang",
        "repo": "rust-analyzer",
-        "rev": "d8887c0758bbd2d5f752d5bd405d4491e90e7ed6",
+        "rev": "582af7ee9c8d84f5d534272fc7de9f292bd849be",
        "type": "github"
      },
      "original": {
--- a/flake.nix
+++ b/flake.nix
@@ -21,7 +21,7 @@
        lib = nixpkgs.lib;
        rustToolchain = fenix.packages.${system}.fromToolchainName {
          name = (lib.importTOML ./rust-toolchain.toml).toolchain.channel;
-          sha256 = "sha256-arzEYlWLGGYeOhECHpBxQd2joZ4rPKV3qLNnZ+eql6A=";
+          sha256 = "sha256-f/CVA1EC61EWbh0SjaRNhLL0Ypx2ObupbzigZp8NmL4=";
        };
      in
      {
--- a/grafana/README.md
+++ b/grafana/README.md
@@ -2,63 +2,24 @@

 ## Overview

-This repository contains Grafana dashboards for visualizing metrics and logs of GreptimeDB instances running in either cluster or standalone mode. **The Grafana version should be greater than 9.0**.
+This repository maintains the Grafana dashboards for GreptimeDB. It has two types of dashboards:

-We highly recommend using the self-monitoring feature provided by [GreptimeDB Operator](https://github.com/GrepTimeTeam/greptimedb-operator) to automatically collect metrics and logs from your GreptimeDB instances and store them in a dedicated GreptimeDB instance.
+- `cluster/`: The dashboard for the GreptimeDB cluster. Read the [dashboard.md](./dashboards/cluster/dashboard.md) for more details.
+- `standalone/`: The dashboard for the standalone GreptimeDB instance. Read the [dashboard.md](./dashboards/standalone/dashboard.md) for more details.

- **Metrics Dashboards**
+As the rapid development of GreptimeDB, the metrics may be changed, and please feel free to submit your feedback and/or contribution to this dashboard 🤗

-  - `dashboards/metrics/cluster/dashboard.json`: The Grafana dashboard for the GreptimeDB cluster. Read the [dashboard.md](./dashboards/metrics/cluster/dashboard.md) for more details.
-  
-  - `dashboards/metrics/standalone/dashboard.json`: The Grafana dashboard for the standalone GreptimeDB instance. **It's generated from the `cluster/dashboard.json` by removing the instance filter through the `make dashboards` command**. Read the [dashboard.md](./dashboards/metrics/standalone/dashboard.md) for more details.
+To maintain the dashboards, we use the [`dac`](https://github.com/zyy17/dac) tool to generate the intermediate dashboards and markdown documents:

- **Logs Dashboard**
-
-  The `dashboards/logs/dashboard.json` provides a comprehensive Grafana dashboard for visualizing GreptimeDB logs. To utilize this dashboard effectively, you need to collect logs in JSON format from your GreptimeDB instances and store them in a dedicated GreptimeDB instance.
-
-  For proper integration, the logs table must adhere to the following schema design with the table name `_gt_logs`:
-
-  ```sql
-  CREATE TABLE IF NOT EXISTS `_gt_logs` (
-    `pod_ip` STRING NULL,
-    `namespace` STRING NULL,
-    `cluster` STRING NULL,
-    `file` STRING NULL,
-    `module_path` STRING NULL,
-    `level` STRING NULL,
-    `target` STRING NULL,
-    `role` STRING NULL,
-    `pod` STRING NULL SKIPPING INDEX WITH(granularity = '10240', type = 'BLOOM'),
-    `message` STRING NULL FULLTEXT INDEX WITH(analyzer = 'English', backend = 'bloom', case_sensitive = 'false'),
-    `err` STRING NULL FULLTEXT INDEX WITH(analyzer = 'English', backend = 'bloom', case_sensitive = 'false'),
-    `timestamp` TIMESTAMP(9) NOT NULL,
-    TIME INDEX (`timestamp`),
-    PRIMARY KEY (`level`, `target`, `role`)
-  )
-    ENGINE=mito
-  WITH (
-    append_mode = 'true'
-  )
-  ```
-
-## Development
-
-As GreptimeDB evolves rapidly, metrics may change over time. We welcome your feedback and contributions to improve these dashboards 🤗
-
-To modify the metrics dashboards, simply edit the `dashboards/metrics/cluster/dashboard.json` file and run the `make dashboards` command. This will automatically generate the updated `dashboards/metrics/standalone/dashboard.json` and other related files.
-
-For easier dashboard maintenance, we utilize the [`dac`](https://github.com/zyy17/dac) tool to generate human-readable intermediate dashboards and documentation:
-
- `dashboards/metrics/cluster/dashboard.yaml`: The intermediate dashboard file for the GreptimeDB cluster.
- `dashboards/metrics/standalone/dashboard.yaml`: The intermediate dashboard file for standalone GreptimeDB instances.
+- `cluster/dashboard.yaml`: The intermediate dashboard for the GreptimeDB cluster.
+- `standalone/dashboard.yaml`: The intermediatedashboard for the standalone GreptimeDB instance.

 ## Data Sources

-The following data sources are used to fetch metrics and logs:
+There are two data sources for the dashboards to fetch the metrics:

- **`${metrics}`**: Prometheus data source for providing the GreptimeDB metrics.
- **`${logs}`**: MySQL data source for providing the GreptimeDB logs.
- **`${information_schema}`**: MySQL data source for providing the information schema of the current instance and used for the `overview` panel. It is the MySQL port of the current monitored instance.
+- **Prometheus**: Expose the metrics of GreptimeDB.
+- **Information Schema**: It is the MySQL port of the current monitored instance. The `overview` dashboard will use this datasource to show the information schema of the current instance.

 ## Instance Filters

@@ -76,9 +37,9 @@ And the legend will be like: `[{{instance}}]-[{{ pod }}]`.

 ## Deployment

-### (Recommended) Helm Chart
+### Helm

-If you use the [Helm Chart](https://github.com/GreptimeTeam/helm-charts) to deploy a GreptimeDB cluster, you can enable self-monitoring by setting the following values in your Helm chart:
+If you use the Helm [chart](https://github.com/GreptimeTeam/helm-charts) to deploy a GreptimeDB cluster, you can enable self-monitoring by setting the following values in your Helm chart:

 - `monitoring.enabled=true`: Deploys a standalone GreptimeDB instance dedicated to monitoring the cluster;
 - `grafana.enabled=true`: Deploys Grafana and automatically imports the monitoring dashboard;
@@ -118,5 +79,5 @@ The standalone GreptimeDB instance will collect metrics from your cluster, and t

 3. **Import the dashboards based on your deployment scenario**

-   - **Cluster**: Import the `dashboards/metrics/cluster/dashboard.json` dashboard.
-   - **Standalone**: Import the `dashboards/metrics/standalone/dashboard.json` dashboard.
+   - **Cluster**: Import the `cluster/dashboard.json` dashboard.
+   - **Standalone**: Import the `standalone/dashboard.json` dashboard.
--- a/grafana/dashboards/metrics/cluster/dashboard.json
+++ b/grafana/dashboards/metrics/cluster/dashboard.json
--- a/grafana/dashboards/metrics/cluster/dashboard.md
+++ b/grafana/dashboards/metrics/cluster/dashboard.md
@@ -1,97 +1,96 @@
 # Overview
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Uptime | `time() - process_start_time_seconds` | `stat` | The start time of GreptimeDB. | `prometheus` | `s` | `__auto` |
-| Version | `SELECT pkg_version FROM information_schema.build_info` | `stat` | GreptimeDB version. | `mysql` | -- | -- |
-| Total Ingestion Rate | `sum(rate(greptime_table_operator_ingest_rows[$__rate_interval]))` | `stat` | Total ingestion rate. | `prometheus` | `rowsps` | `__auto` |
-| Total Storage Size | `select SUM(disk_size) from information_schema.region_statistics;` | `stat` | Total number of data file size. | `mysql` | `decbytes` | -- |
-| Total Rows | `select SUM(region_rows) from information_schema.region_statistics;` | `stat` | Total number of data rows in the cluster. Calculated by sum of rows from each region. | `mysql` | `sishort` | -- |
-| Deployment | `SELECT count(*) as datanode FROM information_schema.cluster_info WHERE peer_type = 'DATANODE';`<br/>`SELECT count(*) as frontend FROM information_schema.cluster_info WHERE peer_type = 'FRONTEND';`<br/>`SELECT count(*) as metasrv FROM information_schema.cluster_info WHERE peer_type = 'METASRV';`<br/>`SELECT count(*) as flownode FROM information_schema.cluster_info WHERE peer_type = 'FLOWNODE';` | `stat` | The deployment topology of GreptimeDB. | `mysql` | -- | -- |
-| Database Resources | `SELECT COUNT(*) as databases FROM information_schema.schemata WHERE schema_name NOT IN ('greptime_private', 'information_schema')`<br/>`SELECT COUNT(*) as tables FROM information_schema.tables WHERE table_schema != 'information_schema'`<br/>`SELECT COUNT(region_id) as regions FROM information_schema.region_peers`<br/>`SELECT COUNT(*) as flows FROM information_schema.flows` | `stat` | The number of the key resources in GreptimeDB. | `mysql` | -- | -- |
-| Data Size | `SELECT SUM(memtable_size) * 0.42825 as WAL FROM information_schema.region_statistics;`<br/>`SELECT SUM(index_size) as index FROM information_schema.region_statistics;`<br/>`SELECT SUM(manifest_size) as manifest FROM information_schema.region_statistics;` | `stat` | The data size of wal/index/manifest in the GreptimeDB. | `mysql` | `decbytes` | -- |
+| Uptime | `time() - process_start_time_seconds` | `stat` | The start time of GreptimeDB. | `s` | `prometheus` | `__auto` |
+| Version | `SELECT pkg_version FROM information_schema.build_info` | `stat` | GreptimeDB version. | -- | `mysql` | -- |
+| Total Ingestion Rate | `sum(rate(greptime_table_operator_ingest_rows[$__rate_interval]))` | `stat` | Total ingestion rate. | `rowsps` | `prometheus` | `__auto` |
+| Total Storage Size | `select SUM(disk_size) from information_schema.region_statistics;` | `stat` | Total number of data file size. | `decbytes` | `mysql` | -- |
+| Total Rows | `select SUM(region_rows) from information_schema.region_statistics;` | `stat` | Total number of data rows in the cluster. Calculated by sum of rows from each region. | `sishort` | `mysql` | -- |
+| Deployment | `SELECT count(*) as datanode FROM information_schema.cluster_info WHERE peer_type = 'DATANODE';`<br/>`SELECT count(*) as frontend FROM information_schema.cluster_info WHERE peer_type = 'FRONTEND';`<br/>`SELECT count(*) as metasrv FROM information_schema.cluster_info WHERE peer_type = 'METASRV';`<br/>`SELECT count(*) as flownode FROM information_schema.cluster_info WHERE peer_type = 'FLOWNODE';` | `stat` | The deployment topology of GreptimeDB. | -- | `mysql` | -- |
+| Database Resources | `SELECT COUNT(*) as databases FROM information_schema.schemata WHERE schema_name NOT IN ('greptime_private', 'information_schema')`<br/>`SELECT COUNT(*) as tables FROM information_schema.tables WHERE table_schema != 'information_schema'`<br/>`SELECT COUNT(region_id) as regions FROM information_schema.region_peers`<br/>`SELECT COUNT(*) as flows FROM information_schema.flows` | `stat` | The number of the key resources in GreptimeDB. | -- | `mysql` | -- |
+| Data Size | `SELECT SUM(memtable_size) * 0.42825 as WAL FROM information_schema.region_statistics;`<br/>`SELECT SUM(index_size) as index FROM information_schema.region_statistics;`<br/>`SELECT SUM(manifest_size) as manifest FROM information_schema.region_statistics;` | `stat` | The data size of wal/index/manifest in the GreptimeDB. | `decbytes` | `mysql` | -- |
 # Ingestion
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Total Ingestion Rate | `sum(rate(greptime_table_operator_ingest_rows{instance=~"$frontend"}[$__rate_interval]))` | `timeseries` | Total ingestion rate.<br/><br/>Here we listed 3 primary protocols:<br/><br/>- Prometheus remote write<br/>- Greptime's gRPC API (when using our ingest SDK)<br/>- Log ingestion http API<br/> | `prometheus` | `rowsps` | `ingestion` |
-| Ingestion Rate by Type | `sum(rate(greptime_servers_http_logs_ingestion_counter[$__rate_interval]))`<br/>`sum(rate(greptime_servers_prometheus_remote_write_samples[$__rate_interval]))` | `timeseries` | Total ingestion rate.<br/><br/>Here we listed 3 primary protocols:<br/><br/>- Prometheus remote write<br/>- Greptime's gRPC API (when using our ingest SDK)<br/>- Log ingestion http API<br/> | `prometheus` | `rowsps` | `http-logs` |
+| Total Ingestion Rate | `sum(rate(greptime_table_operator_ingest_rows{instance=~"$frontend"}[$__rate_interval]))` | `timeseries` | Total ingestion rate.<br/><br/>Here we listed 3 primary protocols:<br/><br/>- Prometheus remote write<br/>- Greptime's gRPC API (when using our ingest SDK)<br/>- Log ingestion http API<br/> | `rowsps` | `prometheus` | `ingestion` |
+| Ingestion Rate by Type | `sum(rate(greptime_servers_http_logs_ingestion_counter[$__rate_interval]))`<br/>`sum(rate(greptime_servers_prometheus_remote_write_samples[$__rate_interval]))` | `timeseries` | Total ingestion rate.<br/><br/>Here we listed 3 primary protocols:<br/><br/>- Prometheus remote write<br/>- Greptime's gRPC API (when using our ingest SDK)<br/>- Log ingestion http API<br/> | `rowsps` | `prometheus` | `http-logs` |
 # Queries
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Total Query Rate | `sum (rate(greptime_servers_mysql_query_elapsed_count{instance=~"$frontend"}[$__rate_interval]))`<br/>`sum (rate(greptime_servers_postgres_query_elapsed_count{instance=~"$frontend"}[$__rate_interval]))`<br/>`sum (rate(greptime_servers_http_promql_elapsed_counte{instance=~"$frontend"}[$__rate_interval]))` | `timeseries` | Total rate of query API calls by protocol. This metric is collected from frontends.<br/><br/>Here we listed 3 main protocols:<br/>- MySQL<br/>- Postgres<br/>- Prometheus API<br/><br/>Note that there are some other minor query APIs like /sql are not included | `prometheus` | `reqps` | `mysql` |
+| Total Query Rate | `sum (rate(greptime_servers_mysql_query_elapsed_count{instance=~"$frontend"}[$__rate_interval]))`<br/>`sum (rate(greptime_servers_postgres_query_elapsed_count{instance=~"$frontend"}[$__rate_interval]))`<br/>`sum (rate(greptime_servers_http_promql_elapsed_counte{instance=~"$frontend"}[$__rate_interval]))` | `timeseries` | Total rate of query API calls by protocol. This metric is collected from frontends.<br/><br/>Here we listed 3 main protocols:<br/>- MySQL<br/>- Postgres<br/>- Prometheus API<br/><br/>Note that there are some other minor query APIs like /sql are not included | `reqps` | `prometheus` | `mysql` |
 # Resources
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Datanode Memory per Instance | `sum(process_resident_memory_bytes{instance=~"$datanode"}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `prometheus` | `decbytes` | `[{{instance}}]-[{{ pod }}]` |
-| Datanode CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{instance=~"$datanode"}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
-| Frontend Memory per Instance | `sum(process_resident_memory_bytes{instance=~"$frontend"}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `prometheus` | `decbytes` | `[{{ instance }}]-[{{ pod }}]` |
-| Frontend CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{instance=~"$frontend"}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]-cpu` |
-| Metasrv Memory per Instance | `sum(process_resident_memory_bytes{instance=~"$metasrv"}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `prometheus` | `decbytes` | `[{{ instance }}]-[{{ pod }}]-resident` |
-| Metasrv CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{instance=~"$metasrv"}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
-| Flownode Memory per Instance | `sum(process_resident_memory_bytes{instance=~"$flownode"}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `prometheus` | `decbytes` | `[{{ instance }}]-[{{ pod }}]` |
-| Flownode CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{instance=~"$flownode"}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
+| Datanode Memory per Instance | `sum(process_resident_memory_bytes{instance=~"$datanode"}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `decbytes` | `prometheus` | `[{{instance}}]-[{{ pod }}]` |
+| Datanode CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{instance=~"$datanode"}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `none` | `prometheus` | `[{{ instance }}]-[{{ pod }}]` |
+| Frontend Memory per Instance | `sum(process_resident_memory_bytes{instance=~"$frontend"}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `decbytes` | `prometheus` | `[{{ instance }}]-[{{ pod }}]` |
+| Frontend CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{instance=~"$frontend"}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `none` | `prometheus` | `[{{ instance }}]-[{{ pod }}]-cpu` |
+| Metasrv Memory per Instance | `sum(process_resident_memory_bytes{instance=~"$metasrv"}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `decbytes` | `prometheus` | `[{{ instance }}]-[{{ pod }}]-resident` |
+| Metasrv CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{instance=~"$metasrv"}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `none` | `prometheus` | `[{{ instance }}]-[{{ pod }}]` |
+| Flownode Memory per Instance | `sum(process_resident_memory_bytes{instance=~"$flownode"}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `decbytes` | `prometheus` | `[{{ instance }}]-[{{ pod }}]` |
+| Flownode CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{instance=~"$flownode"}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `none` | `prometheus` | `[{{ instance }}]-[{{ pod }}]` |
 # Frontend Requests
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| HTTP QPS per Instance | `sum by(instance, pod, path, method, code) (rate(greptime_servers_http_requests_elapsed_count{instance=~"$frontend",path!~"/health\|/metrics"}[$__rate_interval]))` | `timeseries` | HTTP QPS per Instance. | `prometheus` | `reqps` | `[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]` |
-| HTTP P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, path, method, code) (rate(greptime_servers_http_requests_elapsed_bucket{instance=~"$frontend",path!~"/health\|/metrics"}[$__rate_interval])))` | `timeseries` | HTTP P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]-p99` |
-| gRPC QPS per Instance | `sum by(instance, pod, path, code) (rate(greptime_servers_grpc_requests_elapsed_count{instance=~"$frontend"}[$__rate_interval]))` | `timeseries` | gRPC QPS per Instance. | `prometheus` | `reqps` | `[{{instance}}]-[{{pod}}]-[{{path}}]-[{{code}}]` |
-| gRPC P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, path, code) (rate(greptime_servers_grpc_requests_elapsed_bucket{instance=~"$frontend"}[$__rate_interval])))` | `timeseries` | gRPC P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]-p99` |
-| MySQL QPS per Instance | `sum by(pod, instance)(rate(greptime_servers_mysql_query_elapsed_count{instance=~"$frontend"}[$__rate_interval]))` | `timeseries` | MySQL QPS per Instance. | `prometheus` | `reqps` | `[{{instance}}]-[{{pod}}]` |
-| MySQL P99 per Instance | `histogram_quantile(0.99, sum by(pod, instance, le) (rate(greptime_servers_mysql_query_elapsed_bucket{instance=~"$frontend"}[$__rate_interval])))` | `timeseries` | MySQL P99 per Instance. | `prometheus` | `s` | `[{{ instance }}]-[{{ pod }}]-p99` |
-| PostgreSQL QPS per Instance | `sum by(pod, instance)(rate(greptime_servers_postgres_query_elapsed_count{instance=~"$frontend"}[$__rate_interval]))` | `timeseries` | PostgreSQL QPS per Instance. | `prometheus` | `reqps` | `[{{instance}}]-[{{pod}}]` |
-| PostgreSQL P99 per Instance | `histogram_quantile(0.99, sum by(pod,instance,le) (rate(greptime_servers_postgres_query_elapsed_bucket{instance=~"$frontend"}[$__rate_interval])))` | `timeseries` | PostgreSQL P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-p99` |
+| HTTP QPS per Instance | `sum by(instance, pod, path, method, code) (rate(greptime_servers_http_requests_elapsed_count{instance=~"$frontend",path!~"/health\|/metrics"}[$__rate_interval]))` | `timeseries` | HTTP QPS per Instance. | `reqps` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]` |
+| HTTP P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, path, method, code) (rate(greptime_servers_http_requests_elapsed_bucket{instance=~"$frontend",path!~"/health\|/metrics"}[$__rate_interval])))` | `timeseries` | HTTP P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]-p99` |
+| gRPC QPS per Instance | `sum by(instance, pod, path, code) (rate(greptime_servers_grpc_requests_elapsed_count{instance=~"$frontend"}[$__rate_interval]))` | `timeseries` | gRPC QPS per Instance. | `reqps` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{path}}]-[{{code}}]` |
+| gRPC P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, path, code) (rate(greptime_servers_grpc_requests_elapsed_bucket{instance=~"$frontend"}[$__rate_interval])))` | `timeseries` | gRPC P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]-p99` |
+| MySQL QPS per Instance | `sum by(pod, instance)(rate(greptime_servers_mysql_query_elapsed_count{instance=~"$frontend"}[$__rate_interval]))` | `timeseries` | MySQL QPS per Instance. | `reqps` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
+| MySQL P99 per Instance | `histogram_quantile(0.99, sum by(pod, instance, le) (rate(greptime_servers_mysql_query_elapsed_bucket{instance=~"$frontend"}[$__rate_interval])))` | `timeseries` | MySQL P99 per Instance. | `s` | `prometheus` | `[{{ instance }}]-[{{ pod }}]-p99` |
+| PostgreSQL QPS per Instance | `sum by(pod, instance)(rate(greptime_servers_postgres_query_elapsed_count{instance=~"$frontend"}[$__rate_interval]))` | `timeseries` | PostgreSQL QPS per Instance. | `reqps` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
+| PostgreSQL P99 per Instance | `histogram_quantile(0.99, sum by(pod,instance,le) (rate(greptime_servers_postgres_query_elapsed_bucket{instance=~"$frontend"}[$__rate_interval])))` | `timeseries` | PostgreSQL P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-p99` |
 # Frontend to Datanode
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Ingest Rows per Instance | `sum by(instance, pod)(rate(greptime_table_operator_ingest_rows{instance=~"$frontend"}[$__rate_interval]))` | `timeseries` | Ingestion rate by row as in each frontend | `prometheus` | `rowsps` | `[{{instance}}]-[{{pod}}]` |
-| Region Call QPS per Instance | `sum by(instance, pod, request_type) (rate(greptime_grpc_region_request_count{instance=~"$frontend"}[$__rate_interval]))` | `timeseries` | Region Call QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{request_type}}]` |
-| Region Call P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, request_type) (rate(greptime_grpc_region_request_bucket{instance=~"$frontend"}[$__rate_interval])))` | `timeseries` | Region Call P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{request_type}}]` |
+| Ingest Rows per Instance | `sum by(instance, pod)(rate(greptime_table_operator_ingest_rows{instance=~"$frontend"}[$__rate_interval]))` | `timeseries` | Ingestion rate by row as in each frontend | `rowsps` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
+| Region Call QPS per Instance | `sum by(instance, pod, request_type) (rate(greptime_grpc_region_request_count{instance=~"$frontend"}[$__rate_interval]))` | `timeseries` | Region Call QPS per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{request_type}}]` |
+| Region Call P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, request_type) (rate(greptime_grpc_region_request_bucket{instance=~"$frontend"}[$__rate_interval])))` | `timeseries` | Region Call P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{request_type}}]` |
 # Mito Engine
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Request OPS per Instance | `sum by(instance, pod, type) (rate(greptime_mito_handle_request_elapsed_count{instance=~"$datanode"}[$__rate_interval]))` | `timeseries` | Request QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{type}}]` |
-| Request P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, type) (rate(greptime_mito_handle_request_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))` | `timeseries` | Request P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{type}}]` |
-| Write Buffer per Instance | `greptime_mito_write_buffer_bytes{instance=~"$datanode"}` | `timeseries` | Write Buffer per Instance. | `prometheus` | `decbytes` | `[{{instance}}]-[{{pod}}]` |
-| Write Rows per Instance | `sum by (instance, pod) (rate(greptime_mito_write_rows_total{instance=~"$datanode"}[$__rate_interval]))` | `timeseries` | Ingestion size by row counts. | `prometheus` | `rowsps` | `[{{instance}}]-[{{pod}}]` |
-| Flush OPS per Instance | `sum by(instance, pod, reason) (rate(greptime_mito_flush_requests_total{instance=~"$datanode"}[$__rate_interval]))` | `timeseries` | Flush QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{reason}}]` |
-| Write Stall per Instance | `sum by(instance, pod) (greptime_mito_write_stall_total{instance=~"$datanode"})` | `timeseries` | Write Stall per Instance. | `prometheus` | -- | `[{{instance}}]-[{{pod}}]` |
-| Read Stage OPS per Instance | `sum by(instance, pod) (rate(greptime_mito_read_stage_elapsed_count{instance=~"$datanode", stage="total"}[$__rate_interval]))` | `timeseries` | Read Stage OPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]` |
-| Read Stage P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_read_stage_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))` | `timeseries` | Read Stage P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]` |
-| Write Stage P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_write_stage_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))` | `timeseries` | Write Stage P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]` |
-| Compaction OPS per Instance | `sum by(instance, pod) (rate(greptime_mito_compaction_total_elapsed_count{instance=~"$datanode"}[$__rate_interval]))` | `timeseries` | Compaction OPS per Instance. | `prometheus` | `ops` | `[{{ instance }}]-[{{pod}}]` |
-| Compaction P99 per Instance by Stage | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))` | `timeseries` | Compaction latency by stage | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-p99` |
-| Compaction P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le,stage) (rate(greptime_mito_compaction_total_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))` | `timeseries` | Compaction P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-compaction` |
-| WAL write size | `histogram_quantile(0.95, sum by(le,instance, pod) (rate(raft_engine_write_size_bucket[$__rate_interval])))`<br/>`histogram_quantile(0.99, sum by(le,instance,pod) (rate(raft_engine_write_size_bucket[$__rate_interval])))`<br/>`sum by (instance, pod)(rate(raft_engine_write_size_sum[$__rate_interval]))` | `timeseries` | Write-ahead logs write size as bytes. This chart includes stats of p95 and p99 size by instance, total WAL write rate. | `prometheus` | `bytes` | `[{{instance}}]-[{{pod}}]-req-size-p95` |
-| Cached Bytes per Instance | `greptime_mito_cache_bytes{instance=~"$datanode"}` | `timeseries` | Cached Bytes per Instance. | `prometheus` | `decbytes` | `[{{instance}}]-[{{pod}}]-[{{type}}]` |
-| Inflight Compaction | `greptime_mito_inflight_compaction_count` | `timeseries` | Ongoing compaction task count | `prometheus` | `none` | `[{{instance}}]-[{{pod}}]` |
-| WAL sync duration seconds | `histogram_quantile(0.99, sum by(le, type, node, instance, pod) (rate(raft_engine_sync_log_duration_seconds_bucket[$__rate_interval])))` | `timeseries` | Raft engine (local disk) log store sync latency, p99 | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-p99` |
-| Log Store op duration seconds | `histogram_quantile(0.99, sum by(le,logstore,optype,instance, pod) (rate(greptime_logstore_op_elapsed_bucket[$__rate_interval])))` | `timeseries` | Write-ahead log operations latency at p99 | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{logstore}}]-[{{optype}}]-p99` |
-| Inflight Flush | `greptime_mito_inflight_flush_count` | `timeseries` | Ongoing flush task count | `prometheus` | `none` | `[{{instance}}]-[{{pod}}]` |
+| Request OPS per Instance | `sum by(instance, pod, type) (rate(greptime_mito_handle_request_elapsed_count{instance=~"$datanode"}[$__rate_interval]))` | `timeseries` | Request QPS per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{type}}]` |
+| Request P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, type) (rate(greptime_mito_handle_request_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))` | `timeseries` | Request P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{type}}]` |
+| Write Buffer per Instance | `greptime_mito_write_buffer_bytes{instance=~"$datanode"}` | `timeseries` | Write Buffer per Instance. | `decbytes` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
+| Write Rows per Instance | `sum by (instance, pod) (rate(greptime_mito_write_rows_total{instance=~"$datanode"}[$__rate_interval]))` | `timeseries` | Ingestion size by row counts. | `rowsps` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
+| Flush OPS per Instance | `sum by(instance, pod, reason) (rate(greptime_mito_flush_requests_total{instance=~"$datanode"}[$__rate_interval]))` | `timeseries` | Flush QPS per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{reason}}]` |
+| Write Stall per Instance | `sum by(instance, pod) (greptime_mito_write_stall_total{instance=~"$datanode"})` | `timeseries` | Write Stall per Instance. | `decbytes` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
+| Read Stage OPS per Instance | `sum by(instance, pod) (rate(greptime_mito_read_stage_elapsed_count{instance=~"$datanode", stage="total"}[$__rate_interval]))` | `timeseries` | Read Stage OPS per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
+| Read Stage P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_read_stage_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))` | `timeseries` | Read Stage P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{stage}}]` |
+| Write Stage P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_write_stage_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))` | `timeseries` | Write Stage P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{stage}}]` |
+| Compaction OPS per Instance | `sum by(instance, pod) (rate(greptime_mito_compaction_total_elapsed_count{instance=~"$datanode"}[$__rate_interval]))` | `timeseries` | Compaction OPS per Instance. | `ops` | `prometheus` | `[{{ instance }}]-[{{pod}}]` |
+| Compaction P99 per Instance by Stage | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))` | `timeseries` | Compaction latency by stage | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-p99` |
+| Compaction P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le,stage) (rate(greptime_mito_compaction_total_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))` | `timeseries` | Compaction P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-compaction` |
+| WAL write size | `histogram_quantile(0.95, sum by(le,instance, pod) (rate(raft_engine_write_size_bucket[$__rate_interval])))`<br/>`histogram_quantile(0.99, sum by(le,instance,pod) (rate(raft_engine_write_size_bucket[$__rate_interval])))`<br/>`sum by (instance, pod)(rate(raft_engine_write_size_sum[$__rate_interval]))` | `timeseries` | Write-ahead logs write size as bytes. This chart includes stats of p95 and p99 size by instance, total WAL write rate. | `bytes` | `prometheus` | `[{{instance}}]-[{{pod}}]-req-size-p95` |
+| Cached Bytes per Instance | `greptime_mito_cache_bytes{instance=~"$datanode"}` | `timeseries` | Cached Bytes per Instance. | `decbytes` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{type}}]` |
+| Inflight Compaction | `greptime_mito_inflight_compaction_count` | `timeseries` | Ongoing compaction task count | `none` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
+| WAL sync duration seconds | `histogram_quantile(0.99, sum by(le, type, node, instance, pod) (rate(raft_engine_sync_log_duration_seconds_bucket[$__rate_interval])))` | `timeseries` | Raft engine (local disk) log store sync latency, p99 | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-p99` |
+| Log Store op duration seconds | `histogram_quantile(0.99, sum by(le,logstore,optype,instance, pod) (rate(greptime_logstore_op_elapsed_bucket[$__rate_interval])))` | `timeseries` | Write-ahead log operations latency at p99 | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{logstore}}]-[{{optype}}]-p99` |
+| Inflight Flush | `greptime_mito_inflight_flush_count` | `timeseries` | Ongoing flush task count | `none` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
 # OpenDAL
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| QPS per Instance | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode"}[$__rate_interval]))` | `timeseries` | QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
-| Read QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation="read"}[$__rate_interval]))` | `timeseries` | Read QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
-| Read P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode",operation="read"}[$__rate_interval])))` | `timeseries` | Read P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-{{scheme}}` |
-| Write QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation="write"}[$__rate_interval]))` | `timeseries` | Write QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-{{scheme}}` |
-| Write P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode", operation="write"}[$__rate_interval])))` | `timeseries` | Write P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
-| List QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation="list"}[$__rate_interval]))` | `timeseries` | List QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
-| List P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode", operation="list"}[$__rate_interval])))` | `timeseries` | List P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
-| Other Requests per Instance | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode",operation!~"read\|write\|list\|stat"}[$__rate_interval]))` | `timeseries` | Other Requests per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
-| Other Request P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode", operation!~"read\|write\|list"}[$__rate_interval])))` | `timeseries` | Other Request P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
-| Opendal traffic | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_bytes_sum{instance=~"$datanode"}[$__rate_interval]))` | `timeseries` | Total traffic as in bytes by instance and operation | `prometheus` | `decbytes` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
-| OpenDAL errors per Instance | `sum by(instance, pod, scheme, operation, error) (rate(opendal_operation_errors_total{instance=~"$datanode", error!="NotFound"}[$__rate_interval]))` | `timeseries` | OpenDAL error counts per Instance. | `prometheus` | -- | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]-[{{error}}]` |
+| QPS per Instance | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode"}[$__rate_interval]))` | `timeseries` | QPS per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
+| Read QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation="read"}[$__rate_interval]))` | `timeseries` | Read QPS per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
+| Read P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode",operation="read"}[$__rate_interval])))` | `timeseries` | Read P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-{{scheme}}` |
+| Write QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation="write"}[$__rate_interval]))` | `timeseries` | Write QPS per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-{{scheme}}` |
+| Write P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode", operation="write"}[$__rate_interval])))` | `timeseries` | Write P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
+| List QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation="list"}[$__rate_interval]))` | `timeseries` | List QPS per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
+| List P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode", operation="list"}[$__rate_interval])))` | `timeseries` | List P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
+| Other Requests per Instance | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode",operation!~"read\|write\|list\|stat"}[$__rate_interval]))` | `timeseries` | Other Requests per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
+| Other Request P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode", operation!~"read\|write\|list"}[$__rate_interval])))` | `timeseries` | Other Request P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
+| Opendal traffic | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_bytes_sum{instance=~"$datanode"}[$__rate_interval]))` | `timeseries` | Total traffic as in bytes by instance and operation | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
 # Metasrv
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Region migration datanode | `greptime_meta_region_migration_stat{datanode_type="src"}`<br/>`greptime_meta_region_migration_stat{datanode_type="desc"}` | `state-timeline` | Counter of region migration by source and destination | `prometheus` | `none` | `from-datanode-{{datanode_id}}` |
-| Region migration error | `greptime_meta_region_migration_error` | `timeseries` | Counter of region migration error | `prometheus` | `none` | `__auto` |
-| Datanode load | `greptime_datanode_load` | `timeseries` | Gauge of load information of each datanode, collected via heartbeat between datanode and metasrv. This information is for metasrv to schedule workloads. | `prometheus` | `none` | `__auto` |
+| Region migration datanode | `greptime_meta_region_migration_stat{datanode_type="src"}`<br/>`greptime_meta_region_migration_stat{datanode_type="desc"}` | `state-timeline` | Counter of region migration by source and destination | `none` | `prometheus` | `from-datanode-{{datanode_id}}` |
+| Region migration error | `greptime_meta_region_migration_error` | `timeseries` | Counter of region migration error | `none` | `prometheus` | `__auto` |
+| Datanode load | `greptime_datanode_load` | `timeseries` | Gauge of load information of each datanode, collected via heartbeat between datanode and metasrv. This information is for metasrv to schedule workloads. | `none` | `prometheus` | `__auto` |
 # Flownode
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Flow Ingest / Output Rate | `sum by(instance, pod, direction) (rate(greptime_flow_processed_rows[$__rate_interval]))` | `timeseries` | Flow Ingest / Output Rate. | `prometheus` | -- | `[{{pod}}]-[{{instance}}]-[{{direction}}]` |
-| Flow Ingest Latency | `histogram_quantile(0.95, sum(rate(greptime_flow_insert_elapsed_bucket[$__rate_interval])) by (le, instance, pod))`<br/>`histogram_quantile(0.99, sum(rate(greptime_flow_insert_elapsed_bucket[$__rate_interval])) by (le, instance, pod))` | `timeseries` | Flow Ingest Latency. | `prometheus` | -- | `[{{instance}}]-[{{pod}}]-p95` |
-| Flow Operation Latency | `histogram_quantile(0.95, sum(rate(greptime_flow_processing_time_bucket[$__rate_interval])) by (le,instance,pod,type))`<br/>`histogram_quantile(0.99, sum(rate(greptime_flow_processing_time_bucket[$__rate_interval])) by (le,instance,pod,type))` | `timeseries` | Flow Operation Latency. | `prometheus` | -- | `[{{instance}}]-[{{pod}}]-[{{type}}]-p95` |
-| Flow Buffer Size per Instance | `greptime_flow_input_buf_size` | `timeseries` | Flow Buffer Size per Instance. | `prometheus` | -- | `[{{instance}}]-[{{pod}]` |
-| Flow Processing Error per Instance | `sum by(instance,pod,code) (rate(greptime_flow_errors[$__rate_interval]))` | `timeseries` | Flow Processing Error per Instance. | `prometheus` | -- | `[{{instance}}]-[{{pod}}]-[{{code}}]` |
+| Flow Ingest / Output Rate | `sum by(instance, pod, direction) (rate(greptime_flow_processed_rows[$__rate_interval]))` | `timeseries` | Flow Ingest / Output Rate. | -- | `prometheus` | `[{{pod}}]-[{{instance}}]-[{{direction}}]` |
+| Flow Ingest Latency | `histogram_quantile(0.95, sum(rate(greptime_flow_insert_elapsed_bucket[$__rate_interval])) by (le, instance, pod))`<br/>`histogram_quantile(0.99, sum(rate(greptime_flow_insert_elapsed_bucket[$__rate_interval])) by (le, instance, pod))` | `timeseries` | Flow Ingest Latency. | -- | `prometheus` | `[{{instance}}]-[{{pod}}]-p95` |
+| Flow Operation Latency | `histogram_quantile(0.95, sum(rate(greptime_flow_processing_time_bucket[$__rate_interval])) by (le,instance,pod,type))`<br/>`histogram_quantile(0.99, sum(rate(greptime_flow_processing_time_bucket[$__rate_interval])) by (le,instance,pod,type))` | `timeseries` | Flow Operation Latency. | -- | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{type}}]-p95` |
+| Flow Buffer Size per Instance | `greptime_flow_input_buf_size` | `timeseries` | Flow Buffer Size per Instance. | -- | `prometheus` | `[{{instance}}]-[{{pod}]` |
+| Flow Processing Error per Instance | `sum by(instance,pod,code) (rate(greptime_flow_errors[$__rate_interval]))` | `timeseries` | Flow Processing Error per Instance. | -- | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{code}}]` |
--- a/grafana/dashboards/metrics/cluster/dashboard.yaml
+++ b/grafana/dashboards/metrics/cluster/dashboard.yaml
@@ -426,6 +426,7 @@ groups:
        - title: Write Stall per Instance
          type: timeseries
          description: Write Stall per Instance.
+          unit: decbytes
          queries:
            - expr: sum by(instance, pod) (greptime_mito_write_stall_total{instance=~"$datanode"})
              datasource:
@@ -657,22 +658,13 @@ groups:
        - title: Opendal traffic
          type: timeseries
          description: Total traffic as in bytes by instance and operation
-          unit: decbytes
+          unit: ops
          queries:
            - expr: sum by(instance, pod, scheme, operation) (rate(opendal_operation_bytes_sum{instance=~"$datanode"}[$__rate_interval]))
              datasource:
                type: prometheus
                uid: ${metrics}
              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]'
-        - title: OpenDAL errors per Instance
-          type: timeseries
-          description: OpenDAL error counts per Instance.
-          queries:
-            - expr: sum by(instance, pod, scheme, operation, error) (rate(opendal_operation_errors_total{instance=~"$datanode", error!="NotFound"}[$__rate_interval]))
-              datasource:
-                type: prometheus
-                uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]-[{{error}}]'
    - title: Metasrv
      panels:
        - title: Region migration datanode
--- a/grafana/dashboards/logs/dashboard.json
+++ b/grafana/dashboards/logs/dashboard.json
@@ -1,292 +0,0 @@
-{
-  "annotations": {
-    "list": [
-      {
-        "builtIn": 1,
-        "datasource": {
-          "type": "grafana",
-          "uid": "-- Grafana --"
-        },
-        "enable": true,
-        "hide": true,
-        "iconColor": "rgba(0, 211, 255, 1)",
-        "name": "Annotations & Alerts",
-        "type": "dashboard"
-      }
-    ]
-  },
-  "editable": true,
-  "fiscalYearStartMonth": 0,
-  "graphTooltip": 0,
-  "id": 12,
-  "links": [],
-  "panels": [
-    {
-      "datasource": {
-        "default": false,
-        "type": "mysql",
-        "uid": "${datasource}"
-      },
-      "fieldConfig": {
-        "defaults": {},
-        "overrides": []
-      },
-      "gridPos": {
-        "h": 20,
-        "w": 24,
-        "x": 0,
-        "y": 0
-      },
-      "id": 1,
-      "options": {
-        "dedupStrategy": "none",
-        "enableInfiniteScrolling": true,
-        "enableLogDetails": true,
-        "prettifyLogMessage": false,
-        "showCommonLabels": false,
-        "showLabels": false,
-        "showTime": true,
-        "sortOrder": "Descending",
-        "wrapLogMessage": false
-      },
-      "pluginVersion": "11.6.0",
-      "targets": [
-        {
-          "dataset": "greptime_private",
-          "datasource": {
-            "type": "mysql",
-            "uid": "${datasource}"
-          },
-          "editorMode": "code",
-          "format": "table",
-          "rawQuery": true,
-          "rawSql": "SELECT `timestamp`, CONCAT('[', `level`, ']', ' ', '<', `target`, '>', ' ', `message`),\n  `role`,\n  `pod`,\n  `pod_ip`,\n  `namespace`,\n  `cluster`,\n  `err`,\n  `file`,\n  `module_path`\nFROM\n  `_gt_logs`\nWHERE\n  (\n    \"$level\" = \"'all'\"\n    OR `level` IN ($level)\n  ) \n  AND (\n    \"$role\" = \"'all'\"\n    OR `role` IN ($role)\n  )\n  AND (\n    \"$pod\" = \"\"\n    OR `pod` = '$pod'\n  )\n  AND (\n    \"$target\" = \"\"\n    OR `target` = '$target'\n  )\n  AND (\n    \"$search\" = \"\"\n    OR matches_term(`message`, '$search')\n  )\n  AND (\n    \"$exclude\" = \"\"\n    OR NOT matches_term(`message`, '$exclude')\n  )\n  AND $__timeFilter(`timestamp`)\nORDER BY `timestamp` DESC\nLIMIT $limit;\n",
-          "refId": "A",
-          "sql": {
-            "columns": [
-              {
-                "parameters": [],
-                "type": "function"
-              }
-            ],
-            "groupBy": [
-              {
-                "property": {
-                  "type": "string"
-                },
-                "type": "groupBy"
-              }
-            ],
-            "limit": 50
-          }
-        }
-      ],
-      "title": "Logs",
-      "type": "logs"
-    }
-  ],
-  "preload": false,
-  "refresh": "",
-  "schemaVersion": 41,
-  "tags": [],
-  "templating": {
-    "list": [
-      {
-        "current": {
-          "text": "logs",
-          "value": "P98F38F12DB221A8C"
-        },
-        "includeAll": false,
-        "name": "datasource",
-        "options": [],
-        "query": "mysql",
-        "refresh": 1,
-        "regex": "",
-        "type": "datasource"
-      },
-      {
-        "allValue": "'all'",
-        "current": {
-          "text": [
-            "$__all"
-          ],
-          "value": [
-            "$__all"
-          ]
-        },
-        "includeAll": true,
-        "label": "level",
-        "multi": true,
-        "name": "level",
-        "options": [
-          {
-            "selected": false,
-            "text": "INFO",
-            "value": "INFO"
-          },
-          {
-            "selected": false,
-            "text": "ERROR",
-            "value": "ERROR"
-          },
-          {
-            "selected": false,
-            "text": "WARN",
-            "value": "WARN"
-          },
-          {
-            "selected": false,
-            "text": "DEBUG",
-            "value": "DEBUG"
-          },
-          {
-            "selected": false,
-            "text": "TRACE",
-            "value": "TRACE"
-          }
-        ],
-        "query": "INFO,ERROR,WARN,DEBUG,TRACE",
-        "type": "custom"
-      },
-      {
-        "allValue": "'all'",
-        "current": {
-          "text": [
-            "$__all"
-          ],
-          "value": [
-            "$__all"
-          ]
-        },
-        "includeAll": true,
-        "label": "role",
-        "multi": true,
-        "name": "role",
-        "options": [
-          {
-            "selected": false,
-            "text": "datanode",
-            "value": "datanode"
-          },
-          {
-            "selected": false,
-            "text": "frontend",
-            "value": "frontend"
-          },
-          {
-            "selected": false,
-            "text": "meta",
-            "value": "meta"
-          }
-        ],
-        "query": "datanode,frontend,meta",
-        "type": "custom"
-      },
-      {
-        "current": {
-          "text": "",
-          "value": ""
-        },
-        "label": "pod",
-        "name": "pod",
-        "options": [
-          {
-            "selected": true,
-            "text": "",
-            "value": ""
-          }
-        ],
-        "query": "",
-        "type": "textbox"
-      },
-      {
-        "current": {
-          "text": "",
-          "value": ""
-        },
-        "label": "target",
-        "name": "target",
-        "options": [
-          {
-            "selected": true,
-            "text": "",
-            "value": ""
-          }
-        ],
-        "query": "",
-        "type": "textbox"
-      },
-      {
-        "current": {
-          "text": "",
-          "value": ""
-        },
-        "label": "search",
-        "name": "search",
-        "options": [
-          {
-            "selected": true,
-            "text": "",
-            "value": ""
-          }
-        ],
-        "query": "",
-        "type": "textbox"
-      },
-      {
-        "current": {
-          "text": "",
-          "value": ""
-        },
-        "label": "exclude",
-        "name": "exclude",
-        "options": [
-          {
-            "selected": true,
-            "text": "",
-            "value": ""
-          }
-        ],
-        "query": "",
-        "type": "textbox"
-      },
-      {
-        "current": {
-          "text": "2000",
-          "value": "2000"
-        },
-        "includeAll": false,
-        "label": "limit",
-        "name": "limit",
-        "options": [
-          {
-            "selected": true,
-            "text": "2000",
-            "value": "2000"
-          },
-          {
-            "selected": false,
-            "text": "5000",
-            "value": "5000"
-          },
-          {
-            "selected": false,
-            "text": "8000",
-            "value": "8000"
-          }
-        ],
-        "query": "2000,5000,8000",
-        "type": "custom"
-      }
-    ]
-  },
-  "time": {
-    "from": "now-6h",
-    "to": "now"
-  },
-  "timepicker": {},
-  "timezone": "browser",
-  "title": "GreptimeDB Logs",
-  "uid": "edx5veo4rd3wge2",
-  "version": 1
-}
--- a/grafana/dashboards/metrics/standalone/dashboard.json
+++ b/grafana/dashboards/metrics/standalone/dashboard.json
--- a/grafana/dashboards/metrics/standalone/dashboard.md
+++ b/grafana/dashboards/metrics/standalone/dashboard.md
@@ -1,97 +1,96 @@
 # Overview
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Uptime | `time() - process_start_time_seconds` | `stat` | The start time of GreptimeDB. | `prometheus` | `s` | `__auto` |
-| Version | `SELECT pkg_version FROM information_schema.build_info` | `stat` | GreptimeDB version. | `mysql` | -- | -- |
-| Total Ingestion Rate | `sum(rate(greptime_table_operator_ingest_rows[$__rate_interval]))` | `stat` | Total ingestion rate. | `prometheus` | `rowsps` | `__auto` |
-| Total Storage Size | `select SUM(disk_size) from information_schema.region_statistics;` | `stat` | Total number of data file size. | `mysql` | `decbytes` | -- |
-| Total Rows | `select SUM(region_rows) from information_schema.region_statistics;` | `stat` | Total number of data rows in the cluster. Calculated by sum of rows from each region. | `mysql` | `sishort` | -- |
-| Deployment | `SELECT count(*) as datanode FROM information_schema.cluster_info WHERE peer_type = 'DATANODE';`<br/>`SELECT count(*) as frontend FROM information_schema.cluster_info WHERE peer_type = 'FRONTEND';`<br/>`SELECT count(*) as metasrv FROM information_schema.cluster_info WHERE peer_type = 'METASRV';`<br/>`SELECT count(*) as flownode FROM information_schema.cluster_info WHERE peer_type = 'FLOWNODE';` | `stat` | The deployment topology of GreptimeDB. | `mysql` | -- | -- |
-| Database Resources | `SELECT COUNT(*) as databases FROM information_schema.schemata WHERE schema_name NOT IN ('greptime_private', 'information_schema')`<br/>`SELECT COUNT(*) as tables FROM information_schema.tables WHERE table_schema != 'information_schema'`<br/>`SELECT COUNT(region_id) as regions FROM information_schema.region_peers`<br/>`SELECT COUNT(*) as flows FROM information_schema.flows` | `stat` | The number of the key resources in GreptimeDB. | `mysql` | -- | -- |
-| Data Size | `SELECT SUM(memtable_size) * 0.42825 as WAL FROM information_schema.region_statistics;`<br/>`SELECT SUM(index_size) as index FROM information_schema.region_statistics;`<br/>`SELECT SUM(manifest_size) as manifest FROM information_schema.region_statistics;` | `stat` | The data size of wal/index/manifest in the GreptimeDB. | `mysql` | `decbytes` | -- |
+| Uptime | `time() - process_start_time_seconds` | `stat` | The start time of GreptimeDB. | `s` | `prometheus` | `__auto` |
+| Version | `SELECT pkg_version FROM information_schema.build_info` | `stat` | GreptimeDB version. | -- | `mysql` | -- |
+| Total Ingestion Rate | `sum(rate(greptime_table_operator_ingest_rows[$__rate_interval]))` | `stat` | Total ingestion rate. | `rowsps` | `prometheus` | `__auto` |
+| Total Storage Size | `select SUM(disk_size) from information_schema.region_statistics;` | `stat` | Total number of data file size. | `decbytes` | `mysql` | -- |
+| Total Rows | `select SUM(region_rows) from information_schema.region_statistics;` | `stat` | Total number of data rows in the cluster. Calculated by sum of rows from each region. | `sishort` | `mysql` | -- |
+| Deployment | `SELECT count(*) as datanode FROM information_schema.cluster_info WHERE peer_type = 'DATANODE';`<br/>`SELECT count(*) as frontend FROM information_schema.cluster_info WHERE peer_type = 'FRONTEND';`<br/>`SELECT count(*) as metasrv FROM information_schema.cluster_info WHERE peer_type = 'METASRV';`<br/>`SELECT count(*) as flownode FROM information_schema.cluster_info WHERE peer_type = 'FLOWNODE';` | `stat` | The deployment topology of GreptimeDB. | -- | `mysql` | -- |
+| Database Resources | `SELECT COUNT(*) as databases FROM information_schema.schemata WHERE schema_name NOT IN ('greptime_private', 'information_schema')`<br/>`SELECT COUNT(*) as tables FROM information_schema.tables WHERE table_schema != 'information_schema'`<br/>`SELECT COUNT(region_id) as regions FROM information_schema.region_peers`<br/>`SELECT COUNT(*) as flows FROM information_schema.flows` | `stat` | The number of the key resources in GreptimeDB. | -- | `mysql` | -- |
+| Data Size | `SELECT SUM(memtable_size) * 0.42825 as WAL FROM information_schema.region_statistics;`<br/>`SELECT SUM(index_size) as index FROM information_schema.region_statistics;`<br/>`SELECT SUM(manifest_size) as manifest FROM information_schema.region_statistics;` | `stat` | The data size of wal/index/manifest in the GreptimeDB. | `decbytes` | `mysql` | -- |
 # Ingestion
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Total Ingestion Rate | `sum(rate(greptime_table_operator_ingest_rows{}[$__rate_interval]))` | `timeseries` | Total ingestion rate.<br/><br/>Here we listed 3 primary protocols:<br/><br/>- Prometheus remote write<br/>- Greptime's gRPC API (when using our ingest SDK)<br/>- Log ingestion http API<br/> | `prometheus` | `rowsps` | `ingestion` |
-| Ingestion Rate by Type | `sum(rate(greptime_servers_http_logs_ingestion_counter[$__rate_interval]))`<br/>`sum(rate(greptime_servers_prometheus_remote_write_samples[$__rate_interval]))` | `timeseries` | Total ingestion rate.<br/><br/>Here we listed 3 primary protocols:<br/><br/>- Prometheus remote write<br/>- Greptime's gRPC API (when using our ingest SDK)<br/>- Log ingestion http API<br/> | `prometheus` | `rowsps` | `http-logs` |
+| Total Ingestion Rate | `sum(rate(greptime_table_operator_ingest_rows{}[$__rate_interval]))` | `timeseries` | Total ingestion rate.<br/><br/>Here we listed 3 primary protocols:<br/><br/>- Prometheus remote write<br/>- Greptime's gRPC API (when using our ingest SDK)<br/>- Log ingestion http API<br/> | `rowsps` | `prometheus` | `ingestion` |
+| Ingestion Rate by Type | `sum(rate(greptime_servers_http_logs_ingestion_counter[$__rate_interval]))`<br/>`sum(rate(greptime_servers_prometheus_remote_write_samples[$__rate_interval]))` | `timeseries` | Total ingestion rate.<br/><br/>Here we listed 3 primary protocols:<br/><br/>- Prometheus remote write<br/>- Greptime's gRPC API (when using our ingest SDK)<br/>- Log ingestion http API<br/> | `rowsps` | `prometheus` | `http-logs` |
 # Queries
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Total Query Rate | `sum (rate(greptime_servers_mysql_query_elapsed_count{}[$__rate_interval]))`<br/>`sum (rate(greptime_servers_postgres_query_elapsed_count{}[$__rate_interval]))`<br/>`sum (rate(greptime_servers_http_promql_elapsed_counte{}[$__rate_interval]))` | `timeseries` | Total rate of query API calls by protocol. This metric is collected from frontends.<br/><br/>Here we listed 3 main protocols:<br/>- MySQL<br/>- Postgres<br/>- Prometheus API<br/><br/>Note that there are some other minor query APIs like /sql are not included | `prometheus` | `reqps` | `mysql` |
+| Total Query Rate | `sum (rate(greptime_servers_mysql_query_elapsed_count{}[$__rate_interval]))`<br/>`sum (rate(greptime_servers_postgres_query_elapsed_count{}[$__rate_interval]))`<br/>`sum (rate(greptime_servers_http_promql_elapsed_counte{}[$__rate_interval]))` | `timeseries` | Total rate of query API calls by protocol. This metric is collected from frontends.<br/><br/>Here we listed 3 main protocols:<br/>- MySQL<br/>- Postgres<br/>- Prometheus API<br/><br/>Note that there are some other minor query APIs like /sql are not included | `reqps` | `prometheus` | `mysql` |
 # Resources
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Datanode Memory per Instance | `sum(process_resident_memory_bytes{}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `prometheus` | `decbytes` | `[{{instance}}]-[{{ pod }}]` |
-| Datanode CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
-| Frontend Memory per Instance | `sum(process_resident_memory_bytes{}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `prometheus` | `decbytes` | `[{{ instance }}]-[{{ pod }}]` |
-| Frontend CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]-cpu` |
-| Metasrv Memory per Instance | `sum(process_resident_memory_bytes{}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `prometheus` | `decbytes` | `[{{ instance }}]-[{{ pod }}]-resident` |
-| Metasrv CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
-| Flownode Memory per Instance | `sum(process_resident_memory_bytes{}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `prometheus` | `decbytes` | `[{{ instance }}]-[{{ pod }}]` |
-| Flownode CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
+| Datanode Memory per Instance | `sum(process_resident_memory_bytes{}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `decbytes` | `prometheus` | `[{{instance}}]-[{{ pod }}]` |
+| Datanode CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `none` | `prometheus` | `[{{ instance }}]-[{{ pod }}]` |
+| Frontend Memory per Instance | `sum(process_resident_memory_bytes{}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `decbytes` | `prometheus` | `[{{ instance }}]-[{{ pod }}]` |
+| Frontend CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `none` | `prometheus` | `[{{ instance }}]-[{{ pod }}]-cpu` |
+| Metasrv Memory per Instance | `sum(process_resident_memory_bytes{}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `decbytes` | `prometheus` | `[{{ instance }}]-[{{ pod }}]-resident` |
+| Metasrv CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `none` | `prometheus` | `[{{ instance }}]-[{{ pod }}]` |
+| Flownode Memory per Instance | `sum(process_resident_memory_bytes{}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `decbytes` | `prometheus` | `[{{ instance }}]-[{{ pod }}]` |
+| Flownode CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `none` | `prometheus` | `[{{ instance }}]-[{{ pod }}]` |
 # Frontend Requests
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| HTTP QPS per Instance | `sum by(instance, pod, path, method, code) (rate(greptime_servers_http_requests_elapsed_count{path!~"/health\|/metrics"}[$__rate_interval]))` | `timeseries` | HTTP QPS per Instance. | `prometheus` | `reqps` | `[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]` |
-| HTTP P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, path, method, code) (rate(greptime_servers_http_requests_elapsed_bucket{path!~"/health\|/metrics"}[$__rate_interval])))` | `timeseries` | HTTP P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]-p99` |
-| gRPC QPS per Instance | `sum by(instance, pod, path, code) (rate(greptime_servers_grpc_requests_elapsed_count{}[$__rate_interval]))` | `timeseries` | gRPC QPS per Instance. | `prometheus` | `reqps` | `[{{instance}}]-[{{pod}}]-[{{path}}]-[{{code}}]` |
-| gRPC P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, path, code) (rate(greptime_servers_grpc_requests_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | gRPC P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]-p99` |
-| MySQL QPS per Instance | `sum by(pod, instance)(rate(greptime_servers_mysql_query_elapsed_count{}[$__rate_interval]))` | `timeseries` | MySQL QPS per Instance. | `prometheus` | `reqps` | `[{{instance}}]-[{{pod}}]` |
-| MySQL P99 per Instance | `histogram_quantile(0.99, sum by(pod, instance, le) (rate(greptime_servers_mysql_query_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | MySQL P99 per Instance. | `prometheus` | `s` | `[{{ instance }}]-[{{ pod }}]-p99` |
-| PostgreSQL QPS per Instance | `sum by(pod, instance)(rate(greptime_servers_postgres_query_elapsed_count{}[$__rate_interval]))` | `timeseries` | PostgreSQL QPS per Instance. | `prometheus` | `reqps` | `[{{instance}}]-[{{pod}}]` |
-| PostgreSQL P99 per Instance | `histogram_quantile(0.99, sum by(pod,instance,le) (rate(greptime_servers_postgres_query_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | PostgreSQL P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-p99` |
+| HTTP QPS per Instance | `sum by(instance, pod, path, method, code) (rate(greptime_servers_http_requests_elapsed_count{path!~"/health\|/metrics"}[$__rate_interval]))` | `timeseries` | HTTP QPS per Instance. | `reqps` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]` |
+| HTTP P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, path, method, code) (rate(greptime_servers_http_requests_elapsed_bucket{path!~"/health\|/metrics"}[$__rate_interval])))` | `timeseries` | HTTP P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]-p99` |
+| gRPC QPS per Instance | `sum by(instance, pod, path, code) (rate(greptime_servers_grpc_requests_elapsed_count{}[$__rate_interval]))` | `timeseries` | gRPC QPS per Instance. | `reqps` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{path}}]-[{{code}}]` |
+| gRPC P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, path, code) (rate(greptime_servers_grpc_requests_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | gRPC P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]-p99` |
+| MySQL QPS per Instance | `sum by(pod, instance)(rate(greptime_servers_mysql_query_elapsed_count{}[$__rate_interval]))` | `timeseries` | MySQL QPS per Instance. | `reqps` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
+| MySQL P99 per Instance | `histogram_quantile(0.99, sum by(pod, instance, le) (rate(greptime_servers_mysql_query_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | MySQL P99 per Instance. | `s` | `prometheus` | `[{{ instance }}]-[{{ pod }}]-p99` |
+| PostgreSQL QPS per Instance | `sum by(pod, instance)(rate(greptime_servers_postgres_query_elapsed_count{}[$__rate_interval]))` | `timeseries` | PostgreSQL QPS per Instance. | `reqps` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
+| PostgreSQL P99 per Instance | `histogram_quantile(0.99, sum by(pod,instance,le) (rate(greptime_servers_postgres_query_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | PostgreSQL P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-p99` |
 # Frontend to Datanode
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Ingest Rows per Instance | `sum by(instance, pod)(rate(greptime_table_operator_ingest_rows{}[$__rate_interval]))` | `timeseries` | Ingestion rate by row as in each frontend | `prometheus` | `rowsps` | `[{{instance}}]-[{{pod}}]` |
-| Region Call QPS per Instance | `sum by(instance, pod, request_type) (rate(greptime_grpc_region_request_count{}[$__rate_interval]))` | `timeseries` | Region Call QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{request_type}}]` |
-| Region Call P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, request_type) (rate(greptime_grpc_region_request_bucket{}[$__rate_interval])))` | `timeseries` | Region Call P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{request_type}}]` |
+| Ingest Rows per Instance | `sum by(instance, pod)(rate(greptime_table_operator_ingest_rows{}[$__rate_interval]))` | `timeseries` | Ingestion rate by row as in each frontend | `rowsps` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
+| Region Call QPS per Instance | `sum by(instance, pod, request_type) (rate(greptime_grpc_region_request_count{}[$__rate_interval]))` | `timeseries` | Region Call QPS per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{request_type}}]` |
+| Region Call P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, request_type) (rate(greptime_grpc_region_request_bucket{}[$__rate_interval])))` | `timeseries` | Region Call P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{request_type}}]` |
 # Mito Engine
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Request OPS per Instance | `sum by(instance, pod, type) (rate(greptime_mito_handle_request_elapsed_count{}[$__rate_interval]))` | `timeseries` | Request QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{type}}]` |
-| Request P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, type) (rate(greptime_mito_handle_request_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | Request P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{type}}]` |
-| Write Buffer per Instance | `greptime_mito_write_buffer_bytes{}` | `timeseries` | Write Buffer per Instance. | `prometheus` | `decbytes` | `[{{instance}}]-[{{pod}}]` |
-| Write Rows per Instance | `sum by (instance, pod) (rate(greptime_mito_write_rows_total{}[$__rate_interval]))` | `timeseries` | Ingestion size by row counts. | `prometheus` | `rowsps` | `[{{instance}}]-[{{pod}}]` |
-| Flush OPS per Instance | `sum by(instance, pod, reason) (rate(greptime_mito_flush_requests_total{}[$__rate_interval]))` | `timeseries` | Flush QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{reason}}]` |
-| Write Stall per Instance | `sum by(instance, pod) (greptime_mito_write_stall_total{})` | `timeseries` | Write Stall per Instance. | `prometheus` | -- | `[{{instance}}]-[{{pod}}]` |
-| Read Stage OPS per Instance | `sum by(instance, pod) (rate(greptime_mito_read_stage_elapsed_count{ stage="total"}[$__rate_interval]))` | `timeseries` | Read Stage OPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]` |
-| Read Stage P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_read_stage_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | Read Stage P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]` |
-| Write Stage P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_write_stage_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | Write Stage P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]` |
-| Compaction OPS per Instance | `sum by(instance, pod) (rate(greptime_mito_compaction_total_elapsed_count{}[$__rate_interval]))` | `timeseries` | Compaction OPS per Instance. | `prometheus` | `ops` | `[{{ instance }}]-[{{pod}}]` |
-| Compaction P99 per Instance by Stage | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | Compaction latency by stage | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-p99` |
-| Compaction P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le,stage) (rate(greptime_mito_compaction_total_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | Compaction P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-compaction` |
-| WAL write size | `histogram_quantile(0.95, sum by(le,instance, pod) (rate(raft_engine_write_size_bucket[$__rate_interval])))`<br/>`histogram_quantile(0.99, sum by(le,instance,pod) (rate(raft_engine_write_size_bucket[$__rate_interval])))`<br/>`sum by (instance, pod)(rate(raft_engine_write_size_sum[$__rate_interval]))` | `timeseries` | Write-ahead logs write size as bytes. This chart includes stats of p95 and p99 size by instance, total WAL write rate. | `prometheus` | `bytes` | `[{{instance}}]-[{{pod}}]-req-size-p95` |
-| Cached Bytes per Instance | `greptime_mito_cache_bytes{}` | `timeseries` | Cached Bytes per Instance. | `prometheus` | `decbytes` | `[{{instance}}]-[{{pod}}]-[{{type}}]` |
-| Inflight Compaction | `greptime_mito_inflight_compaction_count` | `timeseries` | Ongoing compaction task count | `prometheus` | `none` | `[{{instance}}]-[{{pod}}]` |
-| WAL sync duration seconds | `histogram_quantile(0.99, sum by(le, type, node, instance, pod) (rate(raft_engine_sync_log_duration_seconds_bucket[$__rate_interval])))` | `timeseries` | Raft engine (local disk) log store sync latency, p99 | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-p99` |
-| Log Store op duration seconds | `histogram_quantile(0.99, sum by(le,logstore,optype,instance, pod) (rate(greptime_logstore_op_elapsed_bucket[$__rate_interval])))` | `timeseries` | Write-ahead log operations latency at p99 | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{logstore}}]-[{{optype}}]-p99` |
-| Inflight Flush | `greptime_mito_inflight_flush_count` | `timeseries` | Ongoing flush task count | `prometheus` | `none` | `[{{instance}}]-[{{pod}}]` |
+| Request OPS per Instance | `sum by(instance, pod, type) (rate(greptime_mito_handle_request_elapsed_count{}[$__rate_interval]))` | `timeseries` | Request QPS per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{type}}]` |
+| Request P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, type) (rate(greptime_mito_handle_request_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | Request P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{type}}]` |
+| Write Buffer per Instance | `greptime_mito_write_buffer_bytes{}` | `timeseries` | Write Buffer per Instance. | `decbytes` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
+| Write Rows per Instance | `sum by (instance, pod) (rate(greptime_mito_write_rows_total{}[$__rate_interval]))` | `timeseries` | Ingestion size by row counts. | `rowsps` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
+| Flush OPS per Instance | `sum by(instance, pod, reason) (rate(greptime_mito_flush_requests_total{}[$__rate_interval]))` | `timeseries` | Flush QPS per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{reason}}]` |
+| Write Stall per Instance | `sum by(instance, pod) (greptime_mito_write_stall_total{})` | `timeseries` | Write Stall per Instance. | `decbytes` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
+| Read Stage OPS per Instance | `sum by(instance, pod) (rate(greptime_mito_read_stage_elapsed_count{ stage="total"}[$__rate_interval]))` | `timeseries` | Read Stage OPS per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
+| Read Stage P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_read_stage_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | Read Stage P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{stage}}]` |
+| Write Stage P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_write_stage_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | Write Stage P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{stage}}]` |
+| Compaction OPS per Instance | `sum by(instance, pod) (rate(greptime_mito_compaction_total_elapsed_count{}[$__rate_interval]))` | `timeseries` | Compaction OPS per Instance. | `ops` | `prometheus` | `[{{ instance }}]-[{{pod}}]` |
+| Compaction P99 per Instance by Stage | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | Compaction latency by stage | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-p99` |
+| Compaction P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le,stage) (rate(greptime_mito_compaction_total_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | Compaction P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-compaction` |
+| WAL write size | `histogram_quantile(0.95, sum by(le,instance, pod) (rate(raft_engine_write_size_bucket[$__rate_interval])))`<br/>`histogram_quantile(0.99, sum by(le,instance,pod) (rate(raft_engine_write_size_bucket[$__rate_interval])))`<br/>`sum by (instance, pod)(rate(raft_engine_write_size_sum[$__rate_interval]))` | `timeseries` | Write-ahead logs write size as bytes. This chart includes stats of p95 and p99 size by instance, total WAL write rate. | `bytes` | `prometheus` | `[{{instance}}]-[{{pod}}]-req-size-p95` |
+| Cached Bytes per Instance | `greptime_mito_cache_bytes{}` | `timeseries` | Cached Bytes per Instance. | `decbytes` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{type}}]` |
+| Inflight Compaction | `greptime_mito_inflight_compaction_count` | `timeseries` | Ongoing compaction task count | `none` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
+| WAL sync duration seconds | `histogram_quantile(0.99, sum by(le, type, node, instance, pod) (rate(raft_engine_sync_log_duration_seconds_bucket[$__rate_interval])))` | `timeseries` | Raft engine (local disk) log store sync latency, p99 | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-p99` |
+| Log Store op duration seconds | `histogram_quantile(0.99, sum by(le,logstore,optype,instance, pod) (rate(greptime_logstore_op_elapsed_bucket[$__rate_interval])))` | `timeseries` | Write-ahead log operations latency at p99 | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{logstore}}]-[{{optype}}]-p99` |
+| Inflight Flush | `greptime_mito_inflight_flush_count` | `timeseries` | Ongoing flush task count | `none` | `prometheus` | `[{{instance}}]-[{{pod}}]` |
 # OpenDAL
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| QPS per Instance | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{}[$__rate_interval]))` | `timeseries` | QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
-| Read QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{ operation="read"}[$__rate_interval]))` | `timeseries` | Read QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
-| Read P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{operation="read"}[$__rate_interval])))` | `timeseries` | Read P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-{{scheme}}` |
-| Write QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{ operation="write"}[$__rate_interval]))` | `timeseries` | Write QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-{{scheme}}` |
-| Write P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{ operation="write"}[$__rate_interval])))` | `timeseries` | Write P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
-| List QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{ operation="list"}[$__rate_interval]))` | `timeseries` | List QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
-| List P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{ operation="list"}[$__rate_interval])))` | `timeseries` | List P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
-| Other Requests per Instance | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{operation!~"read\|write\|list\|stat"}[$__rate_interval]))` | `timeseries` | Other Requests per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
-| Other Request P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{ operation!~"read\|write\|list"}[$__rate_interval])))` | `timeseries` | Other Request P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
-| Opendal traffic | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_bytes_sum{}[$__rate_interval]))` | `timeseries` | Total traffic as in bytes by instance and operation | `prometheus` | `decbytes` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
-| OpenDAL errors per Instance | `sum by(instance, pod, scheme, operation, error) (rate(opendal_operation_errors_total{ error!="NotFound"}[$__rate_interval]))` | `timeseries` | OpenDAL error counts per Instance. | `prometheus` | -- | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]-[{{error}}]` |
+| QPS per Instance | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{}[$__rate_interval]))` | `timeseries` | QPS per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
+| Read QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{ operation="read"}[$__rate_interval]))` | `timeseries` | Read QPS per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
+| Read P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{operation="read"}[$__rate_interval])))` | `timeseries` | Read P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-{{scheme}}` |
+| Write QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{ operation="write"}[$__rate_interval]))` | `timeseries` | Write QPS per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-{{scheme}}` |
+| Write P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{ operation="write"}[$__rate_interval])))` | `timeseries` | Write P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
+| List QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{ operation="list"}[$__rate_interval]))` | `timeseries` | List QPS per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
+| List P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{ operation="list"}[$__rate_interval])))` | `timeseries` | List P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
+| Other Requests per Instance | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{operation!~"read\|write\|list\|stat"}[$__rate_interval]))` | `timeseries` | Other Requests per Instance. | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
+| Other Request P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{ operation!~"read\|write\|list"}[$__rate_interval])))` | `timeseries` | Other Request P99 per Instance. | `s` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
+| Opendal traffic | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_bytes_sum{}[$__rate_interval]))` | `timeseries` | Total traffic as in bytes by instance and operation | `ops` | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
 # Metasrv
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Region migration datanode | `greptime_meta_region_migration_stat{datanode_type="src"}`<br/>`greptime_meta_region_migration_stat{datanode_type="desc"}` | `state-timeline` | Counter of region migration by source and destination | `prometheus` | `none` | `from-datanode-{{datanode_id}}` |
-| Region migration error | `greptime_meta_region_migration_error` | `timeseries` | Counter of region migration error | `prometheus` | `none` | `__auto` |
-| Datanode load | `greptime_datanode_load` | `timeseries` | Gauge of load information of each datanode, collected via heartbeat between datanode and metasrv. This information is for metasrv to schedule workloads. | `prometheus` | `none` | `__auto` |
+| Region migration datanode | `greptime_meta_region_migration_stat{datanode_type="src"}`<br/>`greptime_meta_region_migration_stat{datanode_type="desc"}` | `state-timeline` | Counter of region migration by source and destination | `none` | `prometheus` | `from-datanode-{{datanode_id}}` |
+| Region migration error | `greptime_meta_region_migration_error` | `timeseries` | Counter of region migration error | `none` | `prometheus` | `__auto` |
+| Datanode load | `greptime_datanode_load` | `timeseries` | Gauge of load information of each datanode, collected via heartbeat between datanode and metasrv. This information is for metasrv to schedule workloads. | `none` | `prometheus` | `__auto` |
 # Flownode
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Flow Ingest / Output Rate | `sum by(instance, pod, direction) (rate(greptime_flow_processed_rows[$__rate_interval]))` | `timeseries` | Flow Ingest / Output Rate. | `prometheus` | -- | `[{{pod}}]-[{{instance}}]-[{{direction}}]` |
-| Flow Ingest Latency | `histogram_quantile(0.95, sum(rate(greptime_flow_insert_elapsed_bucket[$__rate_interval])) by (le, instance, pod))`<br/>`histogram_quantile(0.99, sum(rate(greptime_flow_insert_elapsed_bucket[$__rate_interval])) by (le, instance, pod))` | `timeseries` | Flow Ingest Latency. | `prometheus` | -- | `[{{instance}}]-[{{pod}}]-p95` |
-| Flow Operation Latency | `histogram_quantile(0.95, sum(rate(greptime_flow_processing_time_bucket[$__rate_interval])) by (le,instance,pod,type))`<br/>`histogram_quantile(0.99, sum(rate(greptime_flow_processing_time_bucket[$__rate_interval])) by (le,instance,pod,type))` | `timeseries` | Flow Operation Latency. | `prometheus` | -- | `[{{instance}}]-[{{pod}}]-[{{type}}]-p95` |
-| Flow Buffer Size per Instance | `greptime_flow_input_buf_size` | `timeseries` | Flow Buffer Size per Instance. | `prometheus` | -- | `[{{instance}}]-[{{pod}]` |
-| Flow Processing Error per Instance | `sum by(instance,pod,code) (rate(greptime_flow_errors[$__rate_interval]))` | `timeseries` | Flow Processing Error per Instance. | `prometheus` | -- | `[{{instance}}]-[{{pod}}]-[{{code}}]` |
+| Flow Ingest / Output Rate | `sum by(instance, pod, direction) (rate(greptime_flow_processed_rows[$__rate_interval]))` | `timeseries` | Flow Ingest / Output Rate. | -- | `prometheus` | `[{{pod}}]-[{{instance}}]-[{{direction}}]` |
+| Flow Ingest Latency | `histogram_quantile(0.95, sum(rate(greptime_flow_insert_elapsed_bucket[$__rate_interval])) by (le, instance, pod))`<br/>`histogram_quantile(0.99, sum(rate(greptime_flow_insert_elapsed_bucket[$__rate_interval])) by (le, instance, pod))` | `timeseries` | Flow Ingest Latency. | -- | `prometheus` | `[{{instance}}]-[{{pod}}]-p95` |
+| Flow Operation Latency | `histogram_quantile(0.95, sum(rate(greptime_flow_processing_time_bucket[$__rate_interval])) by (le,instance,pod,type))`<br/>`histogram_quantile(0.99, sum(rate(greptime_flow_processing_time_bucket[$__rate_interval])) by (le,instance,pod,type))` | `timeseries` | Flow Operation Latency. | -- | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{type}}]-p95` |
+| Flow Buffer Size per Instance | `greptime_flow_input_buf_size` | `timeseries` | Flow Buffer Size per Instance. | -- | `prometheus` | `[{{instance}}]-[{{pod}]` |
+| Flow Processing Error per Instance | `sum by(instance,pod,code) (rate(greptime_flow_errors[$__rate_interval]))` | `timeseries` | Flow Processing Error per Instance. | -- | `prometheus` | `[{{instance}}]-[{{pod}}]-[{{code}}]` |
--- a/grafana/dashboards/metrics/standalone/dashboard.yaml
+++ b/grafana/dashboards/metrics/standalone/dashboard.yaml
@@ -426,6 +426,7 @@ groups:
        - title: Write Stall per Instance
          type: timeseries
          description: Write Stall per Instance.
+          unit: decbytes
          queries:
            - expr: sum by(instance, pod) (greptime_mito_write_stall_total{})
              datasource:
@@ -657,22 +658,13 @@ groups:
        - title: Opendal traffic
          type: timeseries
          description: Total traffic as in bytes by instance and operation
-          unit: decbytes
+          unit: ops
          queries:
            - expr: sum by(instance, pod, scheme, operation) (rate(opendal_operation_bytes_sum{}[$__rate_interval]))
              datasource:
                type: prometheus
                uid: ${metrics}
              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]'
-        - title: OpenDAL errors per Instance
-          type: timeseries
-          description: OpenDAL error counts per Instance.
-          queries:
-            - expr: sum by(instance, pod, scheme, operation, error) (rate(opendal_operation_errors_total{ error!="NotFound"}[$__rate_interval]))
-              datasource:
-                type: prometheus
-                uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]-[{{error}}]'
    - title: Metasrv
      panels:
        - title: Region migration datanode
--- a/grafana/scripts/check.sh
+++ b/grafana/scripts/check.sh
@@ -1,6 +1,6 @@
 #!/usr/bin/env bash

-DASHBOARD_DIR=${1:-grafana/dashboards/metrics}
+DASHBOARD_DIR=${1:-grafana/dashboards}

 check_dashboard_description() {
  for dashboard in $(find $DASHBOARD_DIR -name "*.json"); do
@@ -25,7 +25,7 @@ check_dashboard_description() {
 check_dashboards_generation() {
  ./grafana/scripts/gen-dashboards.sh

-  if [[ -n "$(git diff --name-only grafana/dashboards/metrics)" ]]; then
+  if [[ -n "$(git diff --name-only grafana/dashboards)" ]]; then
    echo "Error: The dashboards are not generated correctly. You should execute the `make dashboards` command."
    exit 1
  fi
--- a/grafana/scripts/gen-dashboards.sh
+++ b/grafana/scripts/gen-dashboards.sh
@@ -1,8 +1,8 @@
 #! /usr/bin/env bash

-CLUSTER_DASHBOARD_DIR=${1:-grafana/dashboards/metrics/cluster}
-STANDALONE_DASHBOARD_DIR=${2:-grafana/dashboards/metrics/standalone}
-DAC_IMAGE=ghcr.io/zyy17/dac:20250423-522bd35
+CLUSTER_DASHBOARD_DIR=${1:-grafana/dashboards/cluster}
+STANDALONE_DASHBOARD_DIR=${2:-grafana/dashboards/standalone}
+DAC_IMAGE=ghcr.io/zyy17/dac:20250422-c9435ce

 remove_instance_filters() {
  # Remove the instance filters for the standalone dashboards.
@@ -10,15 +10,8 @@ remove_instance_filters() {
 }

 generate_intermediate_dashboards_and_docs() {
-  docker run -v ${PWD}:/greptimedb --rm ${DAC_IMAGE} \
-    -i /greptimedb/$CLUSTER_DASHBOARD_DIR/dashboard.json \
-    -o /greptimedb/$CLUSTER_DASHBOARD_DIR/dashboard.yaml \
-    -m /greptimedb/$CLUSTER_DASHBOARD_DIR/dashboard.md
-
-  docker run -v ${PWD}:/greptimedb --rm ${DAC_IMAGE} \
-    -i /greptimedb/$STANDALONE_DASHBOARD_DIR/dashboard.json \
-    -o /greptimedb/$STANDALONE_DASHBOARD_DIR/dashboard.yaml \
-    -m /greptimedb/$STANDALONE_DASHBOARD_DIR/dashboard.md
+  docker run -v ${PWD}:/greptimedb --rm ${DAC_IMAGE}  -i /greptimedb/$CLUSTER_DASHBOARD_DIR/dashboard.json -o /greptimedb/$CLUSTER_DASHBOARD_DIR/dashboard.yaml -m > $CLUSTER_DASHBOARD_DIR/dashboard.md
+  docker run -v ${PWD}:/greptimedb --rm ${DAC_IMAGE}  -i /greptimedb/$STANDALONE_DASHBOARD_DIR/dashboard.json -o /greptimedb/$STANDALONE_DASHBOARD_DIR/dashboard.yaml -m > $STANDALONE_DASHBOARD_DIR/dashboard.md
 }

 remove_instance_filters
--- a/rust-toolchain.toml
+++ b/rust-toolchain.toml
@@ -1,2 +1,2 @@
 [toolchain]
-channel = "nightly-2025-04-15"
+channel = "nightly-2024-12-25"
--- a/src/api/src/helper.rs
+++ b/src/api/src/helper.rs
@@ -1050,7 +1050,7 @@ pub fn value_to_grpc_value(value: Value) -> GrpcValue {
            Value::Int64(v) => Some(ValueData::I64Value(v)),
            Value::Float32(v) => Some(ValueData::F32Value(*v)),
            Value::Float64(v) => Some(ValueData::F64Value(*v)),
-            Value::String(v) => Some(ValueData::StringValue(v.into_string())),
+            Value::String(v) => Some(ValueData::StringValue(v.as_utf8().to_string())),
            Value::Binary(v) => Some(ValueData::BinaryValue(v.to_vec())),
            Value::Date(v) => Some(ValueData::DateValue(v.val())),
            Value::Timestamp(v) => Some(match v.unit() {
--- a/src/auth/src/common.rs
+++ b/src/auth/src/common.rs
@@ -36,7 +36,7 @@ pub fn userinfo_by_name(username: Option<String>) -> UserInfoRef {
 }

 pub fn user_provider_from_option(opt: &String) -> Result<UserProviderRef> {
-    let (name, content) = opt.split_once(':').with_context(|| InvalidConfigSnafu {
+    let (name, content) = opt.split_once(':').context(InvalidConfigSnafu {
        value: opt.to_string(),
        msg: "UserProviderOption must be in format `<option>:<value>`",
    })?;
@@ -57,24 +57,6 @@ pub fn user_provider_from_option(opt: &String) -> Result<UserProviderRef> {
    }
 }

-pub fn static_user_provider_from_option(opt: &String) -> Result<StaticUserProvider> {
-    let (name, content) = opt.split_once(':').with_context(|| InvalidConfigSnafu {
-        value: opt.to_string(),
-        msg: "UserProviderOption must be in format `<option>:<value>`",
-    })?;
-    match name {
-        STATIC_USER_PROVIDER => {
-            let provider = StaticUserProvider::new(content)?;
-            Ok(provider)
-        }
-        _ => InvalidConfigSnafu {
-            value: name.to_string(),
-            msg: format!("Invalid UserProviderOption, expect only {STATIC_USER_PROVIDER}"),
-        }
-        .fail(),
-    }
-}
-
 type Username<'a> = &'a str;
 type HostOrIp<'a> = &'a str;

--- a/src/auth/src/error.rs
+++ b/src/auth/src/error.rs
@@ -38,14 +38,6 @@ pub enum Error {
        location: Location,
    },

-    #[snafu(display("Failed to convert to utf8"))]
-    FromUtf8 {
-        #[snafu(source)]
-        error: std::string::FromUtf8Error,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
    #[snafu(display("Authentication source failure"))]
    AuthBackend {
        #[snafu(implicit)]
@@ -93,7 +85,7 @@ impl ErrorExt for Error {
    fn status_code(&self) -> StatusCode {
        match self {
            Error::InvalidConfig { .. } => StatusCode::InvalidArguments,
-            Error::IllegalParam { .. } | Error::FromUtf8 { .. } => StatusCode::InvalidArguments,
+            Error::IllegalParam { .. } => StatusCode::InvalidArguments,
            Error::FileWatch { .. } => StatusCode::InvalidArguments,
            Error::InternalState { .. } => StatusCode::Unexpected,
            Error::Io { .. } => StatusCode::StorageUnavailable,
--- a/src/auth/src/lib.rs
+++ b/src/auth/src/lib.rs
@@ -22,12 +22,10 @@ mod user_provider;
 pub mod tests;

 pub use common::{
-    auth_mysql, static_user_provider_from_option, user_provider_from_option, userinfo_by_name,
-    HashedPassword, Identity, Password,
+    auth_mysql, user_provider_from_option, userinfo_by_name, HashedPassword, Identity, Password,
 };
 pub use permission::{PermissionChecker, PermissionReq, PermissionResp};
 pub use user_info::UserInfo;
-pub use user_provider::static_user_provider::StaticUserProvider;
 pub use user_provider::UserProvider;

 /// pub type alias
--- a/src/auth/src/user_provider/static_user_provider.rs
+++ b/src/auth/src/user_provider/static_user_provider.rs
@@ -15,15 +15,15 @@
 use std::collections::HashMap;

 use async_trait::async_trait;
-use snafu::{OptionExt, ResultExt};
+use snafu::OptionExt;

-use crate::error::{FromUtf8Snafu, InvalidConfigSnafu, Result};
+use crate::error::{InvalidConfigSnafu, Result};
 use crate::user_provider::{authenticate_with_credential, load_credential_from_file};
 use crate::{Identity, Password, UserInfoRef, UserProvider};

 pub(crate) const STATIC_USER_PROVIDER: &str = "static_user_provider";

-pub struct StaticUserProvider {
+pub(crate) struct StaticUserProvider {
    users: HashMap<String, Vec<u8>>,
 }

@@ -60,18 +60,6 @@ impl StaticUserProvider {
                .fail(),
        }
    }
-
-    /// Return a random username/password pair
-    /// This is useful for invoking from other components in the cluster
-    pub fn get_one_user_pwd(&self) -> Result<(String, String)> {
-        let kv = self.users.iter().next().context(InvalidConfigSnafu {
-            value: "",
-            msg: "Expect at least one pair of username and password",
-        })?;
-        let username = kv.0;
-        let pwd = String::from_utf8(kv.1.clone()).context(FromUtf8Snafu)?;
-        Ok((username.clone(), pwd))
-    }
 }

 #[async_trait]
--- a/src/catalog/src/system_schema/pg_catalog/pg_namespace/oid_map.rs
+++ b/src/catalog/src/system_schema/pg_catalog/pg_namespace/oid_map.rs
@@ -84,6 +84,12 @@ mod tests {
        let key1 = "3178510";
        let key2 = "4215648";

+        // have collision
+        assert_eq!(
+            oid_map.hasher.hash_one(key1) as u32,
+            oid_map.hasher.hash_one(key2) as u32
+        );
+
        // insert them into oid_map
        let oid1 = oid_map.get_oid(key1);
        let oid2 = oid_map.get_oid(key2);
--- a/src/cli/Cargo.toml
+++ b/src/cli/Cargo.toml
@@ -51,6 +51,7 @@ opendal = { version = "0.51.1", features = [
 query.workspace = true
 rand.workspace = true
 reqwest.workspace = true
+rustyline = "10.1"
 serde.workspace = true
 serde_json.workspace = true
 servers.workspace = true
--- a/src/cli/src/cmd.rs
+++ b/src/cli/src/cmd.rs
@@ -0,0 +1,154 @@
+// Copyright 2023 Greptime Team
+//
+// 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.
+
+use crate::error::{Error, InvalidReplCommandSnafu, Result};
+
+/// Represents the parsed command from the user (which may be over many lines)
+#[derive(Debug, PartialEq)]
+pub(crate) enum ReplCommand {
+    Help,
+    UseDatabase { db_name: String },
+    Sql { sql: String },
+    Exit,
+}
+
+impl TryFrom<&str> for ReplCommand {
+    type Error = Error;
+
+    fn try_from(input: &str) -> Result<Self> {
+        let input = input.trim();
+        if input.is_empty() {
+            return InvalidReplCommandSnafu {
+                reason: "No command specified".to_string(),
+            }
+            .fail();
+        }
+
+        // If line ends with ';', it must be treated as a complete input.
+        // However, the opposite is not true.
+        let input_is_completed = input.ends_with(';');
+
+        let input = input.strip_suffix(';').map(|x| x.trim()).unwrap_or(input);
+        let lowercase = input.to_lowercase();
+        match lowercase.as_str() {
+            "help" => Ok(Self::Help),
+            "exit" | "quit" => Ok(Self::Exit),
+            _ => match input.split_once(' ') {
+                Some((maybe_use, database)) if maybe_use.to_lowercase() == "use" => {
+                    Ok(Self::UseDatabase {
+                        db_name: database.trim().to_string(),
+                    })
+                }
+                // Any valid SQL must contains at least one whitespace.
+                Some(_) if input_is_completed => Ok(Self::Sql {
+                    sql: input.to_string(),
+                }),
+                _ => InvalidReplCommandSnafu {
+                    reason: format!("unknown command '{input}', maybe input is not completed"),
+                }
+                .fail(),
+            },
+        }
+    }
+}
+
+impl ReplCommand {
+    pub fn help() -> &'static str {
+        r#"
+Available commands (case insensitive):
+- 'help': print this help
+- 'exit' or 'quit': exit the REPL
+- 'use <your database name>': switch to another database/schema context
+- Other typed in text will be treated as SQL.
+  You can enter new line while typing, just remember to end it with ';'.
+"#
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::error::Error::InvalidReplCommand;
+
+    #[test]
+    fn test_from_str() {
+        fn test_ok(s: &str, expected: ReplCommand) {
+            let actual: ReplCommand = s.try_into().unwrap();
+            assert_eq!(expected, actual, "'{}'", s);
+        }
+
+        fn test_err(s: &str) {
+            let result: Result<ReplCommand> = s.try_into();
+            assert!(matches!(result, Err(InvalidReplCommand { .. })))
+        }
+
+        test_err("");
+        test_err("  ");
+        test_err("\t");
+
+        test_ok("help", ReplCommand::Help);
+        test_ok("help", ReplCommand::Help);
+        test_ok("  help", ReplCommand::Help);
+        test_ok("  help  ", ReplCommand::Help);
+        test_ok("  HELP  ", ReplCommand::Help);
+        test_ok("  Help;  ", ReplCommand::Help);
+        test_ok("  help  ; ", ReplCommand::Help);
+
+        test_ok("exit", ReplCommand::Exit);
+        test_ok("exit;", ReplCommand::Exit);
+        test_ok("exit ;", ReplCommand::Exit);
+        test_ok("EXIT", ReplCommand::Exit);
+
+        test_ok("quit", ReplCommand::Exit);
+        test_ok("quit;", ReplCommand::Exit);
+        test_ok("quit ;", ReplCommand::Exit);
+        test_ok("QUIT", ReplCommand::Exit);
+
+        test_ok(
+            "use Foo",
+            ReplCommand::UseDatabase {
+                db_name: "Foo".to_string(),
+            },
+        );
+        test_ok(
+            "  use Foo ;  ",
+            ReplCommand::UseDatabase {
+                db_name: "Foo".to_string(),
+            },
+        );
+        // ensure that database name is case sensitive
+        test_ok(
+            "  use FOO ;  ",
+            ReplCommand::UseDatabase {
+                db_name: "FOO".to_string(),
+            },
+        );
+
+        // ensure that we aren't messing with capitalization
+        test_ok(
+            "SELECT * from foo;",
+            ReplCommand::Sql {
+                sql: "SELECT * from foo".to_string(),
+            },
+        );
+        // Input line (that don't belong to any other cases above) must ends with ';' to make it a valid SQL.
+        test_err("insert blah");
+        test_ok(
+            "insert blah;",
+            ReplCommand::Sql {
+                sql: "insert blah".to_string(),
+            },
+        );
+    }
+}
--- a/src/cli/src/error.rs
+++ b/src/cli/src/error.rs
@@ -101,6 +101,9 @@ pub enum Error {
        error: reqwest::Error,
    },

+    #[snafu(display("Invalid REPL command: {reason}"))]
+    InvalidReplCommand { reason: String },
+
    #[snafu(display("Failed to parse SQL: {}", sql))]
    ParseSql {
        sql: String,
@@ -251,6 +254,7 @@ impl ErrorExt for Error {
            Error::MissingConfig { .. }
            | Error::LoadLayeredConfig { .. }
            | Error::IllegalConfig { .. }
+            | Error::InvalidReplCommand { .. }
            | Error::InitTimezone { .. }
            | Error::ConnectEtcd { .. }
            | Error::CreateDir { .. }
--- a/src/cli/src/helper.rs
+++ b/src/cli/src/helper.rs
@@ -0,0 +1,112 @@
+// Copyright 2023 Greptime Team
+//
+// 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.
+
+use std::borrow::Cow;
+
+use rustyline::completion::Completer;
+use rustyline::highlight::{Highlighter, MatchingBracketHighlighter};
+use rustyline::hint::{Hinter, HistoryHinter};
+use rustyline::validate::{ValidationContext, ValidationResult, Validator};
+
+use crate::cmd::ReplCommand;
+
+pub(crate) struct RustylineHelper {
+    hinter: HistoryHinter,
+    highlighter: MatchingBracketHighlighter,
+}
+
+impl Default for RustylineHelper {
+    fn default() -> Self {
+        Self {
+            hinter: HistoryHinter {},
+            highlighter: MatchingBracketHighlighter::default(),
+        }
+    }
+}
+
+impl rustyline::Helper for RustylineHelper {}
+
+impl Validator for RustylineHelper {
+    fn validate(&self, ctx: &mut ValidationContext<'_>) -> rustyline::Result<ValidationResult> {
+        let input = ctx.input();
+        match ReplCommand::try_from(input) {
+            Ok(_) => Ok(ValidationResult::Valid(None)),
+            Err(e) => {
+                if input.trim_end().ends_with(';') {
+                    // If line ends with ';', it HAS to be a valid command.
+                    Ok(ValidationResult::Invalid(Some(e.to_string())))
+                } else {
+                    Ok(ValidationResult::Incomplete)
+                }
+            }
+        }
+    }
+}
+
+impl Hinter for RustylineHelper {
+    type Hint = String;
+
+    fn hint(&self, line: &str, pos: usize, ctx: &rustyline::Context<'_>) -> Option<Self::Hint> {
+        self.hinter.hint(line, pos, ctx)
+    }
+}
+
+impl Highlighter for RustylineHelper {
+    fn highlight<'l>(&self, line: &'l str, pos: usize) -> Cow<'l, str> {
+        self.highlighter.highlight(line, pos)
+    }
+
+    fn highlight_prompt<'b, 's: 'b, 'p: 'b>(
+        &'s self,
+        prompt: &'p str,
+        default: bool,
+    ) -> Cow<'b, str> {
+        self.highlighter.highlight_prompt(prompt, default)
+    }
+
+    fn highlight_hint<'h>(&self, hint: &'h str) -> Cow<'h, str> {
+        use nu_ansi_term::Style;
+        Cow::Owned(Style::new().dimmed().paint(hint).to_string())
+    }
+
+    fn highlight_candidate<'c>(
+        &self,
+        candidate: &'c str,
+        completion: rustyline::CompletionType,
+    ) -> Cow<'c, str> {
+        self.highlighter.highlight_candidate(candidate, completion)
+    }
+
+    fn highlight_char(&self, line: &str, pos: usize) -> bool {
+        self.highlighter.highlight_char(line, pos)
+    }
+}
+
+impl Completer for RustylineHelper {
+    type Candidate = String;
+
+    fn complete(
+        &self,
+        line: &str,
+        pos: usize,
+        ctx: &rustyline::Context<'_>,
+    ) -> rustyline::Result<(usize, Vec<Self::Candidate>)> {
+        // If there is a hint, use that as the auto-complete when user hits `tab`
+        if let Some(hint) = self.hinter.hint(line, pos, ctx) {
+            Ok((pos, vec![hint]))
+        } else {
+            Ok((0, vec![]))
+        }
+    }
+}
--- a/src/cli/src/lib.rs
+++ b/src/cli/src/lib.rs
@@ -13,9 +13,15 @@
 // limitations under the License.

 mod bench;
-mod database;
 pub mod error;
+// Wait for https://github.com/GreptimeTeam/greptimedb/issues/2373
+#[allow(unused)]
+mod cmd;
 mod export;
+mod helper;
+
+// Wait for https://github.com/GreptimeTeam/greptimedb/issues/2373
+mod database;
 mod import;

 use async_trait::async_trait;
--- a/src/client/src/database.rs
+++ b/src/client/src/database.rs
@@ -36,17 +36,17 @@ use common_grpc::flight::{FlightDecoder, FlightMessage};
 use common_query::Output;
 use common_recordbatch::error::ExternalSnafu;
 use common_recordbatch::RecordBatchStreamWrapper;
+use common_telemetry::error;
 use common_telemetry::tracing_context::W3cTrace;
-use common_telemetry::{error, warn};
 use futures::future;
 use futures_util::{Stream, StreamExt, TryStreamExt};
 use prost::Message;
 use snafu::{ensure, ResultExt};
-use tonic::metadata::{AsciiMetadataKey, AsciiMetadataValue, MetadataMap, MetadataValue};
+use tonic::metadata::{AsciiMetadataKey, MetadataValue};
 use tonic::transport::Channel;

 use crate::error::{
-    ConvertFlightDataSnafu, Error, FlightGetSnafu, IllegalFlightMessagesSnafu,
+    ConvertFlightDataSnafu, Error, FlightGetSnafu, IllegalFlightMessagesSnafu, InvalidAsciiSnafu,
    InvalidTonicMetadataValueSnafu, ServerSnafu,
 };
 use crate::{from_grpc_response, Client, Result};
@@ -165,27 +165,26 @@ impl Database {

        let mut request = tonic::Request::new(request);
        let metadata = request.metadata_mut();
-        Self::put_hints(metadata, hints)?;
-
+        for (key, value) in hints {
+            let key = AsciiMetadataKey::from_bytes(format!("x-greptime-hint-{}", key).as_bytes())
+                .map_err(|_| {
+                InvalidAsciiSnafu {
+                    value: key.to_string(),
+                }
+                .build()
+            })?;
+            let value = value.parse().map_err(|_| {
+                InvalidAsciiSnafu {
+                    value: value.to_string(),
+                }
+                .build()
+            })?;
+            metadata.insert(key, value);
+        }
        let response = client.handle(request).await?.into_inner();
        from_grpc_response(response)
    }

-    fn put_hints(metadata: &mut MetadataMap, hints: &[(&str, &str)]) -> Result<()> {
-        let Some(value) = hints
-            .iter()
-            .map(|(k, v)| format!("{}={}", k, v))
-            .reduce(|a, b| format!("{},{}", a, b))
-        else {
-            return Ok(());
-        };
-
-        let key = AsciiMetadataKey::from_static("x-greptime-hints");
-        let value = AsciiMetadataValue::from_str(&value).context(InvalidTonicMetadataValueSnafu)?;
-        metadata.insert(key, value);
-        Ok(())
-    }
-
    pub async fn handle(&self, request: Request) -> Result<u32> {
        let mut client = make_database_client(&self.client)?.inner;
        let request = self.to_rpc_request(request);
@@ -193,36 +192,6 @@ impl Database {
        from_grpc_response(response)
    }

-    /// Retry if connection fails, max_retries is the max number of retries, so the total wait time
-    /// is `max_retries * GRPC_CONN_TIMEOUT`
-    pub async fn handle_with_retry(&self, request: Request, max_retries: u32) -> Result<u32> {
-        let mut client = make_database_client(&self.client)?.inner;
-        let mut retries = 0;
-        let request = self.to_rpc_request(request);
-        loop {
-            let raw_response = client.handle(request.clone()).await;
-            match (raw_response, retries < max_retries) {
-                (Ok(resp), _) => return from_grpc_response(resp.into_inner()),
-                (Err(err), true) => {
-                    // determine if the error is retryable
-                    if is_grpc_retryable(&err) {
-                        // retry
-                        retries += 1;
-                        warn!("Retrying {} times with error = {:?}", retries, err);
-                        continue;
-                    }
-                }
-                (Err(err), false) => {
-                    error!(
-                        "Failed to send request to grpc handle after {} retries, error = {:?}",
-                        retries, err
-                    );
-                    return Err(err.into());
-                }
-            }
-        }
-    }
-
    #[inline]
    fn to_rpc_request(&self, request: Request) -> GreptimeRequest {
        GreptimeRequest {
@@ -243,49 +212,39 @@ impl Database {
    where
        S: AsRef<str>,
    {
-        self.sql_with_hint(sql, &[]).await
-    }
-
-    pub async fn sql_with_hint<S>(&self, sql: S, hints: &[(&str, &str)]) -> Result<Output>
-    where
-        S: AsRef<str>,
-    {
-        let request = Request::Query(QueryRequest {
+        self.do_get(Request::Query(QueryRequest {
            query: Some(Query::Sql(sql.as_ref().to_string())),
-        });
-        self.do_get(request, hints).await
+        }))
+        .await
    }

    pub async fn logical_plan(&self, logical_plan: Vec<u8>) -> Result<Output> {
-        let request = Request::Query(QueryRequest {
+        self.do_get(Request::Query(QueryRequest {
            query: Some(Query::LogicalPlan(logical_plan)),
-        });
-        self.do_get(request, &[]).await
+        }))
+        .await
    }

    pub async fn create(&self, expr: CreateTableExpr) -> Result<Output> {
-        let request = Request::Ddl(DdlRequest {
+        self.do_get(Request::Ddl(DdlRequest {
            expr: Some(DdlExpr::CreateTable(expr)),
-        });
-        self.do_get(request, &[]).await
+        }))
+        .await
    }

    pub async fn alter(&self, expr: AlterTableExpr) -> Result<Output> {
-        let request = Request::Ddl(DdlRequest {
+        self.do_get(Request::Ddl(DdlRequest {
            expr: Some(DdlExpr::AlterTable(expr)),
-        });
-        self.do_get(request, &[]).await
+        }))
+        .await
    }

-    async fn do_get(&self, request: Request, hints: &[(&str, &str)]) -> Result<Output> {
+    async fn do_get(&self, request: Request) -> Result<Output> {
        let request = self.to_rpc_request(request);
        let request = Ticket {
            ticket: request.encode_to_vec().into(),
        };

-        let mut request = tonic::Request::new(request);
-        Self::put_hints(request.metadata_mut(), hints)?;
-
        let mut client = self.client.make_flight_client()?;

        let response = client.mut_inner().do_get(request).await.or_else(|e| {
@@ -409,11 +368,6 @@ impl Database {
    }
 }

-/// by grpc standard, only `Unavailable` is retryable, see: https://github.com/grpc/grpc/blob/master/doc/statuscodes.md#status-codes-and-their-use-in-grpc
-pub fn is_grpc_retryable(err: &tonic::Status) -> bool {
-    matches!(err.code(), tonic::Code::Unavailable)
-}
-
 #[derive(Default, Debug, Clone)]
 struct FlightContext {
    auth_header: Option<AuthHeader>,
--- a/src/client/src/error.rs
+++ b/src/client/src/error.rs
@@ -110,6 +110,13 @@ pub enum Error {
        location: Location,
    },

+    #[snafu(display("Failed to parse ascii string: {}", value))]
+    InvalidAscii {
+        value: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
    #[snafu(display("Invalid Tonic metadata value"))]
    InvalidTonicMetadataValue {
        #[snafu(source)]
@@ -136,7 +143,10 @@ impl ErrorExt for Error {
            | Error::ConvertFlightData { source, .. }
            | Error::CreateTlsChannel { source, .. } => source.status_code(),
            Error::IllegalGrpcClientState { .. } => StatusCode::Unexpected,
-            Error::InvalidTonicMetadataValue { .. } => StatusCode::InvalidArguments,
+
+            Error::InvalidAscii { .. } | Error::InvalidTonicMetadataValue { .. } => {
+                StatusCode::InvalidArguments
+            }
        }
    }

--- a/src/cmd/src/bin/greptime.rs
+++ b/src/cmd/src/bin/greptime.rs
@@ -15,11 +15,9 @@
 #![doc = include_str!("../../../../README.md")]

 use clap::{Parser, Subcommand};
-use cmd::datanode::builder::InstanceBuilder;
 use cmd::error::{InitTlsProviderSnafu, Result};
 use cmd::options::GlobalOptions;
 use cmd::{cli, datanode, flownode, frontend, metasrv, standalone, App};
-use common_base::Plugins;
 use common_version::version;
 use servers::install_ring_crypto_provider;

@@ -104,10 +102,10 @@ async fn main_body() -> Result<()> {
 async fn start(cli: Command) -> Result<()> {
    match cli.subcmd {
        SubCommand::Datanode(cmd) => {
-            let opts = cmd.load_options(&cli.global_options)?;
-            let plugins = Plugins::new();
-            let builder = InstanceBuilder::try_new_with_init(opts, plugins).await?;
-            cmd.build_with(builder).await?.run().await
+            cmd.build(cmd.load_options(&cli.global_options)?)
+                .await?
+                .run()
+                .await
        }
        SubCommand::Flownode(cmd) => {
            cmd.build(cmd.load_options(&cli.global_options)?)
--- a/src/cmd/src/cli.rs
+++ b/src/cmd/src/cli.rs
@@ -58,7 +58,7 @@ impl App for Instance {
        false
    }

-    async fn stop(&mut self) -> Result<()> {
+    async fn stop(&self) -> Result<()> {
        Ok(())
    }
 }
--- a/src/cmd/src/datanode.rs
+++ b/src/cmd/src/datanode.rs
@@ -12,27 +12,33 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-pub mod builder;
-
+use std::sync::Arc;
 use std::time::Duration;

 use async_trait::async_trait;
+use cache::build_datanode_cache_registry;
+use catalog::kvbackend::MetaKvBackend;
 use clap::Parser;
+use common_base::Plugins;
 use common_config::Configurable;
+use common_meta::cache::LayeredCacheRegistryBuilder;
 use common_telemetry::logging::TracingOptions;
 use common_telemetry::{info, warn};
+use common_version::{short_version, version};
 use common_wal::config::DatanodeWalConfig;
-use datanode::datanode::Datanode;
-use meta_client::MetaClientOptions;
-use snafu::{ensure, ResultExt};
+use datanode::datanode::{Datanode, DatanodeBuilder};
+use datanode::service::DatanodeServiceBuilder;
+use meta_client::{MetaClientOptions, MetaClientType};
+use servers::Mode;
+use snafu::{ensure, OptionExt, ResultExt};
 use tracing_appender::non_blocking::WorkerGuard;

-use crate::datanode::builder::InstanceBuilder;
 use crate::error::{
-    LoadLayeredConfigSnafu, MissingConfigSnafu, Result, ShutdownDatanodeSnafu, StartDatanodeSnafu,
+    LoadLayeredConfigSnafu, MetaClientInitSnafu, MissingConfigSnafu, Result, ShutdownDatanodeSnafu,
+    StartDatanodeSnafu,
 };
 use crate::options::{GlobalOptions, GreptimeOptions};
-use crate::App;
+use crate::{log_versions, App};

 pub const APP_NAME: &str = "greptime-datanode";

@@ -77,7 +83,7 @@ impl App for Instance {
        self.datanode.start().await.context(StartDatanodeSnafu)
    }

-    async fn stop(&mut self) -> Result<()> {
+    async fn stop(&self) -> Result<()> {
        self.datanode
            .shutdown()
            .await
@@ -92,8 +98,8 @@ pub struct Command {
 }

 impl Command {
-    pub async fn build_with(&self, builder: InstanceBuilder) -> Result<Instance> {
-        self.subcmd.build_with(builder).await
+    pub async fn build(&self, opts: DatanodeOptions) -> Result<Instance> {
+        self.subcmd.build(opts).await
    }

    pub fn load_options(&self, global_options: &GlobalOptions) -> Result<DatanodeOptions> {
@@ -109,12 +115,9 @@ enum SubCommand {
 }

 impl SubCommand {
-    async fn build_with(&self, builder: InstanceBuilder) -> Result<Instance> {
+    async fn build(&self, opts: DatanodeOptions) -> Result<Instance> {
        match self {
-            SubCommand::Start(cmd) => {
-                info!("Building datanode with {:#?}", cmd);
-                builder.build().await
-            }
+            SubCommand::Start(cmd) => cmd.build(opts).await,
        }
    }
 }
@@ -260,6 +263,74 @@ impl StartCommand {

        Ok(())
    }
+
+    async fn build(&self, opts: DatanodeOptions) -> Result<Instance> {
+        common_runtime::init_global_runtimes(&opts.runtime);
+
+        let guard = common_telemetry::init_global_logging(
+            APP_NAME,
+            &opts.component.logging,
+            &opts.component.tracing,
+            opts.component.node_id.map(|x| x.to_string()),
+        );
+        log_versions(version(), short_version(), APP_NAME);
+
+        info!("Datanode start command: {:#?}", self);
+        info!("Datanode options: {:#?}", opts);
+
+        let plugin_opts = opts.plugins;
+        let mut opts = opts.component;
+        opts.grpc.detect_server_addr();
+        let mut plugins = Plugins::new();
+        plugins::setup_datanode_plugins(&mut plugins, &plugin_opts, &opts)
+            .await
+            .context(StartDatanodeSnafu)?;
+
+        let member_id = opts
+            .node_id
+            .context(MissingConfigSnafu { msg: "'node_id'" })?;
+
+        let meta_config = opts.meta_client.as_ref().context(MissingConfigSnafu {
+            msg: "'meta_client_options'",
+        })?;
+
+        let meta_client = meta_client::create_meta_client(
+            MetaClientType::Datanode { member_id },
+            meta_config,
+            None,
+        )
+        .await
+        .context(MetaClientInitSnafu)?;
+
+        let meta_backend = Arc::new(MetaKvBackend {
+            client: meta_client.clone(),
+        });
+
+        // Builds cache registry for datanode.
+        let layered_cache_registry = Arc::new(
+            LayeredCacheRegistryBuilder::default()
+                .add_cache_registry(build_datanode_cache_registry(meta_backend.clone()))
+                .build(),
+        );
+
+        let mut datanode = DatanodeBuilder::new(opts.clone(), plugins, Mode::Distributed)
+            .with_meta_client(meta_client)
+            .with_kv_backend(meta_backend)
+            .with_cache_registry(layered_cache_registry)
+            .build()
+            .await
+            .context(StartDatanodeSnafu)?;
+
+        let services = DatanodeServiceBuilder::new(&opts)
+            .with_default_grpc_server(&datanode.region_server())
+            .enable_http_service()
+            .build()
+            .await
+            .context(StartDatanodeSnafu)?;
+        datanode.setup_services(services);
+
+        Ok(Instance::new(datanode, guard))
+    }
 }

 #[cfg(test)]
@@ -281,6 +352,7 @@ mod tests {
        common_telemetry::init_default_ut_logging();
        let mut file = create_named_temp_file();
        let toml_str = r#"
+            mode = "distributed"
            enable_memory_catalog = false
            node_id = 42

@@ -307,6 +379,7 @@ mod tests {
    fn test_read_from_config_file() {
        let mut file = create_named_temp_file();
        let toml_str = r#"
+            mode = "distributed"
            enable_memory_catalog = false
            node_id = 42

@@ -472,6 +545,7 @@ mod tests {
    fn test_config_precedence_order() {
        let mut file = create_named_temp_file();
        let toml_str = r#"
+            mode = "distributed"
            enable_memory_catalog = false
            node_id = 42
            rpc_addr = "127.0.0.1:3001"
--- a/src/cmd/src/datanode/builder.rs
+++ b/src/cmd/src/datanode/builder.rs
@@ -1,137 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// 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.
-
-use std::sync::Arc;
-
-use cache::build_datanode_cache_registry;
-use catalog::kvbackend::MetaKvBackend;
-use common_base::Plugins;
-use common_meta::cache::LayeredCacheRegistryBuilder;
-use common_telemetry::info;
-use common_version::{short_version, version};
-use datanode::datanode::DatanodeBuilder;
-use datanode::service::DatanodeServiceBuilder;
-use meta_client::MetaClientType;
-use snafu::{OptionExt, ResultExt};
-use tracing_appender::non_blocking::WorkerGuard;
-
-use crate::datanode::{DatanodeOptions, Instance, APP_NAME};
-use crate::error::{MetaClientInitSnafu, MissingConfigSnafu, Result, StartDatanodeSnafu};
-use crate::log_versions;
-
-/// Builder for Datanode instance.
-pub struct InstanceBuilder {
-    guard: Vec<WorkerGuard>,
-    opts: DatanodeOptions,
-    datanode_builder: DatanodeBuilder,
-}
-
-impl InstanceBuilder {
-    /// Try to create a new [InstanceBuilder], and do some initialization work like allocating
-    /// runtime resources, setting up global logging and plugins, etc.
-    pub async fn try_new_with_init(
-        mut opts: DatanodeOptions,
-        mut plugins: Plugins,
-    ) -> Result<Self> {
-        let guard = Self::init(&mut opts, &mut plugins).await?;
-
-        let datanode_builder = Self::datanode_builder(&opts, plugins).await?;
-
-        Ok(Self {
-            guard,
-            opts,
-            datanode_builder,
-        })
-    }
-
-    async fn init(opts: &mut DatanodeOptions, plugins: &mut Plugins) -> Result<Vec<WorkerGuard>> {
-        common_runtime::init_global_runtimes(&opts.runtime);
-
-        let dn_opts = &mut opts.component;
-        let guard = common_telemetry::init_global_logging(
-            APP_NAME,
-            &dn_opts.logging,
-            &dn_opts.tracing,
-            dn_opts.node_id.map(|x| x.to_string()),
-        );
-
-        log_versions(version(), short_version(), APP_NAME);
-
-        plugins::setup_datanode_plugins(plugins, &opts.plugins, dn_opts)
-            .await
-            .context(StartDatanodeSnafu)?;
-
-        dn_opts.grpc.detect_server_addr();
-
-        info!("Initialized Datanode instance with {:#?}", opts);
-        Ok(guard)
-    }
-
-    async fn datanode_builder(opts: &DatanodeOptions, plugins: Plugins) -> Result<DatanodeBuilder> {
-        let dn_opts = &opts.component;
-
-        let member_id = dn_opts
-            .node_id
-            .context(MissingConfigSnafu { msg: "'node_id'" })?;
-        let meta_client_options = dn_opts.meta_client.as_ref().context(MissingConfigSnafu {
-            msg: "meta client options",
-        })?;
-        let client = meta_client::create_meta_client(
-            MetaClientType::Datanode { member_id },
-            meta_client_options,
-            Some(&plugins),
-        )
-        .await
-        .context(MetaClientInitSnafu)?;
-
-        let backend = Arc::new(MetaKvBackend {
-            client: client.clone(),
-        });
-        let mut builder = DatanodeBuilder::new(dn_opts.clone(), plugins.clone(), backend.clone());
-
-        let registry = Arc::new(
-            LayeredCacheRegistryBuilder::default()
-                .add_cache_registry(build_datanode_cache_registry(backend))
-                .build(),
-        );
-        builder
-            .with_cache_registry(registry)
-            .with_meta_client(client.clone());
-        Ok(builder)
-    }
-
-    /// Get the mutable builder for Datanode, in case you want to change some fields before the
-    /// final construction.
-    pub fn mut_datanode_builder(&mut self) -> &mut DatanodeBuilder {
-        &mut self.datanode_builder
-    }
-
-    /// Try to build the Datanode instance.
-    pub async fn build(self) -> Result<Instance> {
-        let mut datanode = self
-            .datanode_builder
-            .build()
-            .await
-            .context(StartDatanodeSnafu)?;
-
-        let services = DatanodeServiceBuilder::new(&self.opts.component)
-            .with_default_grpc_server(&datanode.region_server())
-            .enable_http_service()
-            .build()
-            .context(StartDatanodeSnafu)?;
-        datanode.setup_services(services);
-
-        Ok(Instance::new(datanode, self.guard))
-    }
-}
--- a/src/cmd/src/error.rs
+++ b/src/cmd/src/error.rs
@@ -177,6 +177,9 @@ pub enum Error {
        source: meta_srv::error::Error,
    },

+    #[snafu(display("Invalid REPL command: {reason}"))]
+    InvalidReplCommand { reason: String },
+
    #[snafu(display("Failed to parse SQL: {}", sql))]
    ParseSql {
        sql: String,
@@ -328,6 +331,7 @@ impl ErrorExt for Error {
            Error::MissingConfig { .. }
            | Error::LoadLayeredConfig { .. }
            | Error::IllegalConfig { .. }
+            | Error::InvalidReplCommand { .. }
            | Error::InitTimezone { .. }
            | Error::ConnectEtcd { .. }
            | Error::CreateDir { .. }
--- a/src/cmd/src/flownode.rs
+++ b/src/cmd/src/flownode.rs
@@ -33,8 +33,7 @@ use common_telemetry::info;
 use common_telemetry::logging::TracingOptions;
 use common_version::{short_version, version};
 use flow::{
-    get_flow_auth_options, FlownodeBuilder, FlownodeInstance, FlownodeServiceBuilder,
-    FrontendClient, FrontendInvoker,
+    FlownodeBuilder, FlownodeInstance, FlownodeServiceBuilder, FrontendClient, FrontendInvoker,
 };
 use meta_client::{MetaClientOptions, MetaClientType};
 use snafu::{ensure, OptionExt, ResultExt};
@@ -83,14 +82,10 @@ impl App for Instance {
    }

    async fn start(&mut self) -> Result<()> {
-        plugins::start_flownode_plugins(self.flownode.flow_engine().plugins().clone())
-            .await
-            .context(StartFlownodeSnafu)?;
-
        self.flownode.start().await.context(StartFlownodeSnafu)
    }

-    async fn stop(&mut self) -> Result<()> {
+    async fn stop(&self) -> Result<()> {
        self.flownode
            .shutdown()
            .await
@@ -156,9 +151,6 @@ struct StartCommand {
    /// HTTP request timeout in seconds.
    #[clap(long)]
    http_timeout: Option<u64>,
-    /// User Provider cfg, for auth, currently only support static user provider
-    #[clap(long)]
-    user_provider: Option<String>,
 }

 impl StartCommand {
@@ -222,10 +214,6 @@ impl StartCommand {
            opts.http.timeout = Duration::from_secs(http_timeout);
        }

-        if let Some(user_provider) = &self.user_provider {
-            opts.user_provider = Some(user_provider.clone());
-        }
-
        ensure!(
            opts.node_id.is_some(),
            MissingConfigSnafu {
@@ -250,15 +238,9 @@ impl StartCommand {
        info!("Flownode start command: {:#?}", self);
        info!("Flownode options: {:#?}", opts);

-        let plugin_opts = opts.plugins;
        let mut opts = opts.component;
        opts.grpc.detect_server_addr();

-        let mut plugins = Plugins::new();
-        plugins::setup_flownode_plugins(&mut plugins, &plugin_opts, &opts)
-            .await
-            .context(StartFlownodeSnafu)?;
-
        let member_id = opts
            .node_id
            .context(MissingConfigSnafu { msg: "'node_id'" })?;
@@ -333,12 +315,10 @@ impl StartCommand {
        );

        let flow_metadata_manager = Arc::new(FlowMetadataManager::new(cached_meta_backend.clone()));
-        let flow_auth_header = get_flow_auth_options(&opts).context(StartFlownodeSnafu)?;
-        let frontend_client =
-            FrontendClient::from_meta_client(meta_client.clone(), flow_auth_header);
+        let frontend_client = FrontendClient::from_meta_client(meta_client.clone());
        let flownode_builder = FlownodeBuilder::new(
            opts.clone(),
-            plugins,
+            Plugins::new(),
            table_metadata_manager,
            catalog_manager.clone(),
            flow_metadata_manager,
@@ -351,6 +331,7 @@ impl StartCommand {
            .with_grpc_server(flownode.flownode_server().clone())
            .enable_http_service()
            .build()
+            .await
            .context(StartFlownodeSnafu)?;
        flownode.setup_services(services);
        let flownode = flownode;
--- a/src/cmd/src/frontend.rs
+++ b/src/cmd/src/frontend.rs
@@ -89,7 +89,7 @@ impl App for Instance {
            .context(error::StartFrontendSnafu)
    }

-    async fn stop(&mut self) -> Result<()> {
+    async fn stop(&self) -> Result<()> {
        self.frontend
            .shutdown()
            .await
@@ -382,6 +382,7 @@ impl StartCommand {

        let servers = Services::new(opts, instance.clone(), plugins)
            .build()
+            .await
            .context(error::StartFrontendSnafu)?;

        let frontend = Frontend {
@@ -447,6 +448,8 @@ mod tests {
    fn test_read_from_config_file() {
        let mut file = create_named_temp_file();
        let toml_str = r#"
+            mode = "distributed"
+
            [http]
            addr = "127.0.0.1:4000"
            timeout = "0s"
@@ -535,6 +538,8 @@ mod tests {
    fn test_config_precedence_order() {
        let mut file = create_named_temp_file();
        let toml_str = r#"
+            mode = "distributed"
+
            [http]
            addr = "127.0.0.1:4000"

--- a/src/cmd/src/lib.rs
+++ b/src/cmd/src/lib.rs
@@ -74,7 +74,7 @@ pub trait App: Send {
        true
    }

-    async fn stop(&mut self) -> Result<()>;
+    async fn stop(&self) -> Result<()>;

    async fn run(&mut self) -> Result<()> {
        info!("Starting app: {}", self.name());
--- a/src/cmd/src/metasrv.rs
+++ b/src/cmd/src/metasrv.rs
@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use std::fmt;
 use std::time::Duration;

 use async_trait::async_trait;
@@ -69,7 +68,7 @@ impl App for Instance {
        self.instance.start().await.context(StartMetaServerSnafu)
    }

-    async fn stop(&mut self) -> Result<()> {
+    async fn stop(&self) -> Result<()> {
        self.instance
            .shutdown()
            .await
@@ -132,8 +131,8 @@ impl SubCommand {
    }
 }

-#[derive(Default, Parser)]
-pub struct StartCommand {
+#[derive(Debug, Default, Parser)]
+struct StartCommand {
    /// The address to bind the gRPC server.
    #[clap(long, alias = "bind-addr")]
    rpc_bind_addr: Option<String>,
@@ -172,29 +171,8 @@ pub struct StartCommand {
    backend: Option<BackendImpl>,
 }

-impl fmt::Debug for StartCommand {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("StartCommand")
-            .field("rpc_bind_addr", &self.rpc_bind_addr)
-            .field("rpc_server_addr", &self.rpc_server_addr)
-            .field("store_addrs", &self.sanitize_store_addrs())
-            .field("config_file", &self.config_file)
-            .field("selector", &self.selector)
-            .field("use_memory_store", &self.use_memory_store)
-            .field("enable_region_failover", &self.enable_region_failover)
-            .field("http_addr", &self.http_addr)
-            .field("http_timeout", &self.http_timeout)
-            .field("env_prefix", &self.env_prefix)
-            .field("data_home", &self.data_home)
-            .field("store_key_prefix", &self.store_key_prefix)
-            .field("max_txn_ops", &self.max_txn_ops)
-            .field("backend", &self.backend)
-            .finish()
-    }
-}
-
 impl StartCommand {
-    pub fn load_options(&self, global_options: &GlobalOptions) -> Result<MetasrvOptions> {
+    fn load_options(&self, global_options: &GlobalOptions) -> Result<MetasrvOptions> {
        let mut opts = MetasrvOptions::load_layered_options(
            self.config_file.as_deref(),
            self.env_prefix.as_ref(),
@@ -206,15 +184,6 @@ impl StartCommand {
        Ok(opts)
    }

-    fn sanitize_store_addrs(&self) -> Option<Vec<String>> {
-        self.store_addrs.as_ref().map(|addrs| {
-            addrs
-                .iter()
-                .map(|addr| common_meta::kv_backend::util::sanitize_connection_string(addr))
-                .collect()
-        })
-    }
-
    // The precedence order is: cli > config file > environment variables > default values.
    fn merge_with_cli_options(
        &self,
@@ -292,7 +261,7 @@ impl StartCommand {
        Ok(())
    }

-    pub async fn build(&self, opts: MetasrvOptions) -> Result<Instance> {
+    async fn build(&self, opts: MetasrvOptions) -> Result<Instance> {
        common_runtime::init_global_runtimes(&opts.runtime);

        let guard = common_telemetry::init_global_logging(
--- a/src/cmd/src/standalone.rs
+++ b/src/cmd/src/standalone.rs
@@ -56,8 +56,8 @@ use datanode::datanode::{Datanode, DatanodeBuilder};
 use datanode::region_server::RegionServer;
 use file_engine::config::EngineConfig as FileEngineConfig;
 use flow::{
-    FlowConfig, FlownodeBuilder, FlownodeInstance, FlownodeOptions, FrontendClient,
-    FrontendInvoker, GrpcQueryHandlerWithBoxedError, StreamingEngine,
+    FlowConfig, FlowStreamingEngine, FlownodeBuilder, FlownodeInstance, FlownodeOptions,
+    FrontendClient, FrontendInvoker, GrpcQueryHandlerWithBoxedError,
 };
 use frontend::frontend::{Frontend, FrontendOptions};
 use frontend::instance::builder::FrontendBuilder;
@@ -75,6 +75,7 @@ use servers::export_metrics::{ExportMetricsOption, ExportMetricsTask};
 use servers::grpc::GrpcOptions;
 use servers::http::HttpOptions;
 use servers::tls::{TlsMode, TlsOption};
+use servers::Mode;
 use snafu::ResultExt;
 use tokio::sync::RwLock;
 use tracing_appender::non_blocking::WorkerGuard;
@@ -255,8 +256,8 @@ pub struct Instance {

 impl Instance {
    /// Find the socket addr of a server by its `name`.
-    pub fn server_addr(&self, name: &str) -> Option<SocketAddr> {
-        self.frontend.server_handlers().addr(name)
+    pub async fn server_addr(&self, name: &str) -> Option<SocketAddr> {
+        self.frontend.server_handlers().addr(name).await
    }
 }

@@ -293,7 +294,7 @@ impl App for Instance {
        Ok(())
    }

-    async fn stop(&mut self) -> Result<()> {
+    async fn stop(&self) -> Result<()> {
        self.frontend
            .shutdown()
            .await
@@ -496,9 +497,12 @@ impl StartCommand {
            .build(),
        );

-        let mut builder = DatanodeBuilder::new(dn_opts, plugins.clone(), kv_backend.clone());
-        builder.with_cache_registry(layered_cache_registry.clone());
-        let datanode = builder.build().await.context(error::StartDatanodeSnafu)?;
+        let datanode = DatanodeBuilder::new(dn_opts, plugins.clone(), Mode::Standalone)
+            .with_kv_backend(kv_backend.clone())
+            .with_cache_registry(layered_cache_registry.clone())
+            .build()
+            .await
+            .context(error::StartDatanodeSnafu)?;

        let information_extension = Arc::new(StandaloneInformationExtension::new(
            datanode.region_server(),
@@ -540,9 +544,9 @@ impl StartCommand {

        // set the ref to query for the local flow state
        {
-            let flow_streaming_engine = flownode.flow_engine().streaming_engine();
+            let flow_worker_manager = flownode.flow_engine().streaming_engine();
            information_extension
-                .set_flow_streaming_engine(flow_streaming_engine)
+                .set_flow_worker_manager(flow_worker_manager)
                .await;
        }

@@ -611,10 +615,10 @@ impl StartCommand {
            .replace(weak_grpc_handler);

        // set the frontend invoker for flownode
-        let flow_streaming_engine = flownode.flow_engine().streaming_engine();
+        let flow_worker_manager = flownode.flow_engine().streaming_engine();
        // flow server need to be able to use frontend to write insert requests back
        let invoker = FrontendInvoker::build_from(
-            flow_streaming_engine.clone(),
+            flow_worker_manager.clone(),
            catalog_manager.clone(),
            kv_backend.clone(),
            layered_cache_registry.clone(),
@@ -623,13 +627,14 @@ impl StartCommand {
        )
        .await
        .context(error::StartFlownodeSnafu)?;
-        flow_streaming_engine.set_frontend_invoker(invoker).await;
+        flow_worker_manager.set_frontend_invoker(invoker).await;

        let export_metrics_task = ExportMetricsTask::try_new(&opts.export_metrics, Some(&plugins))
            .context(error::ServersSnafu)?;

        let servers = Services::new(opts, fe_instance.clone(), plugins)
            .build()
+            .await
            .context(error::StartFrontendSnafu)?;

        let frontend = Frontend {
@@ -698,7 +703,7 @@ pub struct StandaloneInformationExtension {
    region_server: RegionServer,
    procedure_manager: ProcedureManagerRef,
    start_time_ms: u64,
-    flow_streaming_engine: RwLock<Option<Arc<StreamingEngine>>>,
+    flow_worker_manager: RwLock<Option<Arc<FlowStreamingEngine>>>,
 }

 impl StandaloneInformationExtension {
@@ -707,14 +712,14 @@ impl StandaloneInformationExtension {
            region_server,
            procedure_manager,
            start_time_ms: common_time::util::current_time_millis() as u64,
-            flow_streaming_engine: RwLock::new(None),
+            flow_worker_manager: RwLock::new(None),
        }
    }

-    /// Set the flow streaming engine for the standalone instance.
-    pub async fn set_flow_streaming_engine(&self, flow_streaming_engine: Arc<StreamingEngine>) {
-        let mut guard = self.flow_streaming_engine.write().await;
-        *guard = Some(flow_streaming_engine);
+    /// Set the flow worker manager for the standalone instance.
+    pub async fn set_flow_worker_manager(&self, flow_worker_manager: Arc<FlowStreamingEngine>) {
+        let mut guard = self.flow_worker_manager.write().await;
+        *guard = Some(flow_worker_manager);
    }
 }

@@ -793,7 +798,7 @@ impl InformationExtension for StandaloneInformationExtension {

    async fn flow_stats(&self) -> std::result::Result<Option<FlowStat>, Self::Error> {
        Ok(Some(
-            self.flow_streaming_engine
+            self.flow_worker_manager
                .read()
                .await
                .as_ref()
@@ -853,6 +858,8 @@ mod tests {
    fn test_read_from_config_file() {
        let mut file = create_named_temp_file();
        let toml_str = r#"
+            mode = "distributed"
+
            enable_memory_catalog = true

            [wal]
@@ -983,6 +990,8 @@ mod tests {
    fn test_config_precedence_order() {
        let mut file = create_named_temp_file();
        let toml_str = r#"
+            mode = "standalone"
+
            [http]
            addr = "127.0.0.1:4000"

--- a/src/cmd/tests/load_config_test.rs
+++ b/src/cmd/tests/load_config_test.rs
@@ -74,7 +74,6 @@ fn test_load_datanode_example_config() {
                RegionEngineConfig::File(FileEngineConfig {}),
                RegionEngineConfig::Metric(MetricEngineConfig {
                    experimental_sparse_primary_key_encoding: false,
-                    flush_metadata_region_interval: Duration::from_secs(30),
                }),
            ],
            logging: LoggingOptions {
@@ -217,7 +216,6 @@ fn test_load_standalone_example_config() {
                RegionEngineConfig::File(FileEngineConfig {}),
                RegionEngineConfig::Metric(MetricEngineConfig {
                    experimental_sparse_primary_key_encoding: false,
-                    flush_metadata_region_interval: Duration::from_secs(30),
                }),
            ],
            storage: StorageConfig {
--- a/src/common/config/src/config.rs
+++ b/src/common/config/src/config.rs
@@ -111,9 +111,11 @@ mod tests {
    use serde::{Deserialize, Serialize};

    use super::*;
+    use crate::Mode;

-    #[derive(Debug, Serialize, Deserialize, Default)]
+    #[derive(Debug, Serialize, Deserialize)]
    struct TestDatanodeConfig {
+        mode: Mode,
        node_id: Option<u64>,
        logging: LoggingOptions,
        meta_client: Option<MetaClientOptions>,
@@ -121,6 +123,19 @@ mod tests {
        storage: StorageConfig,
    }

+    impl Default for TestDatanodeConfig {
+        fn default() -> Self {
+            Self {
+                mode: Mode::Distributed,
+                node_id: None,
+                logging: LoggingOptions::default(),
+                meta_client: None,
+                wal: DatanodeWalConfig::default(),
+                storage: StorageConfig::default(),
+            }
+        }
+    }
+
    impl Configurable for TestDatanodeConfig {
        fn env_list_keys() -> Option<&'static [&'static str]> {
            Some(&["meta_client.metasrv_addrs"])
@@ -131,6 +146,7 @@ mod tests {
    fn test_load_layered_options() {
        let mut file = create_named_temp_file();
        let toml_str = r#"
+            mode = "distributed"
            enable_memory_catalog = false
            rpc_addr = "127.0.0.1:3001"
            rpc_hostname = "127.0.0.1"
--- a/src/common/config/src/lib.rs
+++ b/src/common/config/src/lib.rs
@@ -26,6 +26,16 @@ pub fn metadata_store_dir(store_dir: &str) -> String {
    format!("{store_dir}/metadata")
 }

+/// The Server running mode
+#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq, Copy)]
+#[serde(rename_all = "lowercase")]
+pub enum Mode {
+    // The single process mode.
+    Standalone,
+    // The distributed cluster mode.
+    Distributed,
+}
+
 #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(default)]
 pub struct KvBackendConfig {
--- a/src/common/function/Cargo.toml
+++ b/src/common/function/Cargo.toml
@@ -13,7 +13,7 @@ default = ["geo"]
 geo = ["geohash", "h3o", "s2", "wkt", "geo-types", "dep:geo"]

 [dependencies]
-ahash.workspace = true
+ahash = "0.8"
 api.workspace = true
 arc-swap = "1.0"
 async-trait.workspace = true
--- a/src/common/function/src/aggr.rs
+++ b/src/common/function/src/aggr.rs
@@ -19,4 +19,4 @@ mod uddsketch_state;
 pub use geo_path::{GeoPathAccumulator, GEO_PATH_NAME};
 pub(crate) use hll::HllStateType;
 pub use hll::{HllState, HLL_MERGE_NAME, HLL_NAME};
-pub use uddsketch_state::{UddSketchState, UDDSKETCH_MERGE_NAME, UDDSKETCH_STATE_NAME};
+pub use uddsketch_state::{UddSketchState, UDDSKETCH_STATE_NAME};
--- a/src/common/function/src/aggr/uddsketch_state.rs
+++ b/src/common/function/src/aggr/uddsketch_state.rs
@@ -31,28 +31,23 @@ use datafusion::physical_plan::expressions::Literal;
 use datafusion::prelude::create_udaf;
 use datatypes::arrow::array::ArrayRef;
 use datatypes::arrow::datatypes::{DataType, Float64Type};
-use serde::{Deserialize, Serialize};
 use uddsketch::{SketchHashKey, UDDSketch};

 pub const UDDSKETCH_STATE_NAME: &str = "uddsketch_state";

-pub const UDDSKETCH_MERGE_NAME: &str = "uddsketch_merge";
-
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug)]
 pub struct UddSketchState {
    uddsketch: UDDSketch,
-    error_rate: f64,
 }

 impl UddSketchState {
    pub fn new(bucket_size: u64, error_rate: f64) -> Self {
        Self {
            uddsketch: UDDSketch::new(bucket_size, error_rate),
-            error_rate,
        }
    }

-    pub fn state_udf_impl() -> AggregateUDF {
+    pub fn udf_impl() -> AggregateUDF {
        create_udaf(
            UDDSKETCH_STATE_NAME,
            vec![DataType::Int64, DataType::Float64, DataType::Float64],
@@ -66,55 +61,18 @@ impl UddSketchState {
        )
    }

-    /// Create a UDF for the `uddsketch_merge` function.
-    ///
-    /// `uddsketch_merge` accepts bucket size, error rate, and a binary column of states generated by `uddsketch_state`
-    /// and merges them into a single state.
-    ///
-    /// The bucket size and error rate must be the same as the original state.
-    pub fn merge_udf_impl() -> AggregateUDF {
-        create_udaf(
-            UDDSKETCH_MERGE_NAME,
-            vec![DataType::Int64, DataType::Float64, DataType::Binary],
-            Arc::new(DataType::Binary),
-            Volatility::Immutable,
-            Arc::new(|args| {
-                let (bucket_size, error_rate) = downcast_accumulator_args(args)?;
-                Ok(Box::new(UddSketchState::new(bucket_size, error_rate)))
-            }),
-            Arc::new(vec![DataType::Binary]),
-        )
-    }
-
    fn update(&mut self, value: f64) {
        self.uddsketch.add_value(value);
    }

-    fn merge(&mut self, raw: &[u8]) -> DfResult<()> {
-        if let Ok(uddsketch) = bincode::deserialize::<Self>(raw) {
-            if uddsketch.uddsketch.count() != 0 {
-                if self.uddsketch.max_allowed_buckets() != uddsketch.uddsketch.max_allowed_buckets()
-                    || (self.error_rate - uddsketch.error_rate).abs() >= 1e-9
-                {
-                    return Err(DataFusionError::Plan(format!(
-                        "Merging UDDSketch with different parameters: arguments={:?} vs actual input={:?}",
-                        (
-                            self.uddsketch.max_allowed_buckets(),
-                            self.error_rate
-                        ),
-                        (uddsketch.uddsketch.max_allowed_buckets(), uddsketch.error_rate)
-                    )));
-                }
-                self.uddsketch.merge_sketch(&uddsketch.uddsketch);
+    fn merge(&mut self, raw: &[u8]) {
+        if let Ok(uddsketch) = bincode::deserialize::<UDDSketch>(raw) {
+            if uddsketch.count() != 0 {
+                self.uddsketch.merge_sketch(&uddsketch);
            }
        } else {
            trace!("Warning: Failed to deserialize UDDSketch from {:?}", raw);
-            return Err(DataFusionError::Plan(
-                "Failed to deserialize UDDSketch from binary".to_string(),
-            ));
        }
-
-        Ok(())
    }
 }

@@ -155,21 +113,9 @@ fn downcast_accumulator_args(args: AccumulatorArgs) -> DfResult<(u64, f64)> {
 impl DfAccumulator for UddSketchState {
    fn update_batch(&mut self, values: &[ArrayRef]) -> DfResult<()> {
        let array = &values[2]; // the third column is data value
-        match array.data_type() {
-            DataType::Float64 => {
-                let f64_array = as_primitive_array::<Float64Type>(array)?;
-                for v in f64_array.iter().flatten() {
-                    self.update(v);
-                }
-            }
-            // meaning instantiate as `uddsketch_merge`
-            DataType::Binary => self.merge_batch(&[array.clone()])?,
-            _ => {
-                return not_impl_err!(
-                    "UDDSketch functions do not support data type: {}",
-                    array.data_type()
-                )
-            }
+        let f64_array = as_primitive_array::<Float64Type>(array)?;
+        for v in f64_array.iter().flatten() {
+            self.update(v);
        }

        Ok(())
@@ -177,7 +123,7 @@ impl DfAccumulator for UddSketchState {

    fn evaluate(&mut self) -> DfResult<ScalarValue> {
        Ok(ScalarValue::Binary(Some(
-            bincode::serialize(&self).map_err(|e| {
+            bincode::serialize(&self.uddsketch).map_err(|e| {
                DataFusionError::Internal(format!("Failed to serialize UDDSketch: {}", e))
            })?,
        )))
@@ -204,7 +150,7 @@ impl DfAccumulator for UddSketchState {

    fn state(&mut self) -> DfResult<Vec<ScalarValue>> {
        Ok(vec![ScalarValue::Binary(Some(
-            bincode::serialize(&self).map_err(|e| {
+            bincode::serialize(&self.uddsketch).map_err(|e| {
                DataFusionError::Internal(format!("Failed to serialize UDDSketch: {}", e))
            })?,
        ))])
@@ -214,7 +160,7 @@ impl DfAccumulator for UddSketchState {
        let array = &states[0];
        let binary_array = as_binary_array(array)?;
        for v in binary_array.iter().flatten() {
-            self.merge(v)?;
+            self.merge(v);
        }

        Ok(())
@@ -236,8 +182,8 @@ mod tests {

        let result = state.evaluate().unwrap();
        if let ScalarValue::Binary(Some(bytes)) = result {
-            let deserialized: UddSketchState = bincode::deserialize(&bytes).unwrap();
-            assert_eq!(deserialized.uddsketch.count(), 3);
+            let deserialized: UDDSketch = bincode::deserialize(&bytes).unwrap();
+            assert_eq!(deserialized.count(), 3);
        } else {
            panic!("Expected binary scalar value");
        }
@@ -255,15 +201,13 @@ mod tests {
        // Create new state and merge the serialized data
        let mut new_state = UddSketchState::new(10, 0.01);
        if let ScalarValue::Binary(Some(bytes)) = &serialized {
-            new_state.merge(bytes).unwrap();
+            new_state.merge(bytes);

            // Verify the merged state matches original by comparing deserialized values
-            let original_sketch: UddSketchState = bincode::deserialize(bytes).unwrap();
-            let original_sketch = original_sketch.uddsketch;
+            let original_sketch: UDDSketch = bincode::deserialize(bytes).unwrap();
            let new_result = new_state.evaluate().unwrap();
            if let ScalarValue::Binary(Some(new_bytes)) = new_result {
-                let new_sketch: UddSketchState = bincode::deserialize(&new_bytes).unwrap();
-                let new_sketch = new_sketch.uddsketch;
+                let new_sketch: UDDSketch = bincode::deserialize(&new_bytes).unwrap();
                assert_eq!(original_sketch.count(), new_sketch.count());
                assert_eq!(original_sketch.sum(), new_sketch.sum());
                assert_eq!(original_sketch.mean(), new_sketch.mean());
@@ -300,8 +244,7 @@ mod tests {

        let result = state.evaluate().unwrap();
        if let ScalarValue::Binary(Some(bytes)) = result {
-            let deserialized: UddSketchState = bincode::deserialize(&bytes).unwrap();
-            let deserialized = deserialized.uddsketch;
+            let deserialized: UDDSketch = bincode::deserialize(&bytes).unwrap();
            assert_eq!(deserialized.count(), 3);
        } else {
            panic!("Expected binary scalar value");
@@ -330,8 +273,7 @@ mod tests {

            let result = merged_state.evaluate().unwrap();
            if let ScalarValue::Binary(Some(bytes)) = result {
-                let deserialized: UddSketchState = bincode::deserialize(&bytes).unwrap();
-                let deserialized = deserialized.uddsketch;
+                let deserialized: UDDSketch = bincode::deserialize(&bytes).unwrap();
                assert_eq!(deserialized.count(), 2);
            } else {
                panic!("Expected binary scalar value");
--- a/src/common/function/src/scalars/matches_term.rs
+++ b/src/common/function/src/scalars/matches_term.rs
@@ -12,9 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use std::fmt;
-use std::iter::repeat_n;
 use std::sync::Arc;
+use std::{fmt, iter};

 use common_query::error::{InvalidFuncArgsSnafu, Result};
 use common_query::prelude::Volatility;
@@ -127,10 +126,9 @@ impl Function for MatchesTermFunction {
            let term = term_column.get_ref(0).as_string().unwrap();
            match term {
                None => {
-                    return Ok(Arc::new(BooleanVector::from_iter(repeat_n(
-                        None,
-                        text_column.len(),
-                    ))));
+                    return Ok(Arc::new(BooleanVector::from_iter(
+                        iter::repeat(None).take(text_column.len()),
+                    )));
                }
                Some(term) => Some(MatchesTermFinder::new(term)),
            }
@@ -219,7 +217,7 @@ impl MatchesTermFinder {
        }

        let mut pos = 0;
-        while let Some(found_pos) = self.finder.find(&text.as_bytes()[pos..]) {
+        while let Some(found_pos) = self.finder.find(text[pos..].as_bytes()) {
            let actual_pos = pos + found_pos;

            let prev_ok = self.starts_with_non_alnum
--- a/src/common/function/src/scalars/timestamp.rs
+++ b/src/common/function/src/scalars/timestamp.rs
@@ -13,8 +13,10 @@
 // limitations under the License.

 use std::sync::Arc;
+mod greatest;
 mod to_unixtime;

+use greatest::GreatestFunction;
 use to_unixtime::ToUnixtimeFunction;

 use crate::function_registry::FunctionRegistry;
@@ -24,5 +26,6 @@ pub(crate) struct TimestampFunction;
 impl TimestampFunction {
    pub fn register(registry: &FunctionRegistry) {
        registry.register(Arc::new(ToUnixtimeFunction));
+        registry.register(Arc::new(GreatestFunction));
    }
 }
--- a/src/common/function/src/scalars/timestamp/greatest.rs
+++ b/src/common/function/src/scalars/timestamp/greatest.rs
@@ -0,0 +1,328 @@
+// Copyright 2023 Greptime Team
+//
+// 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.
+
+use std::fmt::{self};
+
+use common_query::error::{
+    self, ArrowComputeSnafu, InvalidFuncArgsSnafu, Result, UnsupportedInputDataTypeSnafu,
+};
+use common_query::prelude::{Signature, Volatility};
+use datafusion::arrow::compute::kernels::cmp::gt;
+use datatypes::arrow::array::AsArray;
+use datatypes::arrow::compute::cast;
+use datatypes::arrow::compute::kernels::zip;
+use datatypes::arrow::datatypes::{
+    DataType as ArrowDataType, Date32Type, TimeUnit, TimestampMicrosecondType,
+    TimestampMillisecondType, TimestampNanosecondType, TimestampSecondType,
+};
+use datatypes::prelude::ConcreteDataType;
+use datatypes::types::TimestampType;
+use datatypes::vectors::{Helper, VectorRef};
+use snafu::{ensure, ResultExt};
+
+use crate::function::{Function, FunctionContext};
+
+#[derive(Clone, Debug, Default)]
+pub struct GreatestFunction;
+
+const NAME: &str = "greatest";
+
+macro_rules! gt_time_types {
+    ($ty: ident, $columns:expr) => {{
+        let column1 = $columns[0].to_arrow_array();
+        let column2 = $columns[1].to_arrow_array();
+
+        let column1 = column1.as_primitive::<$ty>();
+        let column2 = column2.as_primitive::<$ty>();
+        let boolean_array = gt(&column1, &column2).context(ArrowComputeSnafu)?;
+
+        let result = zip::zip(&boolean_array, &column1, &column2).context(ArrowComputeSnafu)?;
+        Helper::try_into_vector(&result).context(error::FromArrowArraySnafu)
+    }};
+}
+
+impl Function for GreatestFunction {
+    fn name(&self) -> &str {
+        NAME
+    }
+
+    fn return_type(&self, input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        ensure!(
+            input_types.len() == 2,
+            InvalidFuncArgsSnafu {
+                err_msg: format!(
+                    "The length of the args is not correct, expect exactly two, have: {}",
+                    input_types.len()
+                )
+            }
+        );
+
+        match &input_types[0] {
+            ConcreteDataType::String(_) => Ok(ConcreteDataType::timestamp_millisecond_datatype()),
+            ConcreteDataType::Date(_) => Ok(ConcreteDataType::date_datatype()),
+            ConcreteDataType::Timestamp(ts_type) => Ok(ConcreteDataType::Timestamp(*ts_type)),
+            _ => UnsupportedInputDataTypeSnafu {
+                function: NAME,
+                datatypes: input_types,
+            }
+            .fail(),
+        }
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::uniform(
+            2,
+            vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::date_datatype(),
+                ConcreteDataType::timestamp_nanosecond_datatype(),
+                ConcreteDataType::timestamp_microsecond_datatype(),
+                ConcreteDataType::timestamp_millisecond_datatype(),
+                ConcreteDataType::timestamp_second_datatype(),
+            ],
+            Volatility::Immutable,
+        )
+    }
+
+    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure!(
+            columns.len() == 2,
+            InvalidFuncArgsSnafu {
+                err_msg: format!(
+                    "The length of the args is not correct, expect exactly two, have: {}",
+                    columns.len()
+                ),
+            }
+        );
+        match columns[0].data_type() {
+            ConcreteDataType::String(_) => {
+                let column1 = cast(
+                    &columns[0].to_arrow_array(),
+                    &ArrowDataType::Timestamp(TimeUnit::Millisecond, None),
+                )
+                .context(ArrowComputeSnafu)?;
+                let column1 = column1.as_primitive::<TimestampMillisecondType>();
+                let column2 = cast(
+                    &columns[1].to_arrow_array(),
+                    &ArrowDataType::Timestamp(TimeUnit::Millisecond, None),
+                )
+                .context(ArrowComputeSnafu)?;
+                let column2 = column2.as_primitive::<TimestampMillisecondType>();
+                let boolean_array = gt(&column1, &column2).context(ArrowComputeSnafu)?;
+                let result =
+                    zip::zip(&boolean_array, &column1, &column2).context(ArrowComputeSnafu)?;
+                Ok(Helper::try_into_vector(&result).context(error::FromArrowArraySnafu)?)
+            }
+            ConcreteDataType::Date(_) => gt_time_types!(Date32Type, columns),
+            ConcreteDataType::Timestamp(ts_type) => match ts_type {
+                TimestampType::Second(_) => gt_time_types!(TimestampSecondType, columns),
+                TimestampType::Millisecond(_) => {
+                    gt_time_types!(TimestampMillisecondType, columns)
+                }
+                TimestampType::Microsecond(_) => {
+                    gt_time_types!(TimestampMicrosecondType, columns)
+                }
+                TimestampType::Nanosecond(_) => {
+                    gt_time_types!(TimestampNanosecondType, columns)
+                }
+            },
+            _ => UnsupportedInputDataTypeSnafu {
+                function: NAME,
+                datatypes: columns.iter().map(|c| c.data_type()).collect::<Vec<_>>(),
+            }
+            .fail(),
+        }
+    }
+}
+
+impl fmt::Display for GreatestFunction {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "GREATEST")
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use common_time::timestamp::TimeUnit;
+    use common_time::{Date, Timestamp};
+    use datatypes::types::{
+        DateType, TimestampMicrosecondType, TimestampMillisecondType, TimestampNanosecondType,
+        TimestampSecondType,
+    };
+    use datatypes::value::Value;
+    use datatypes::vectors::{
+        DateVector, StringVector, TimestampMicrosecondVector, TimestampMillisecondVector,
+        TimestampNanosecondVector, TimestampSecondVector, Vector,
+    };
+    use paste::paste;
+
+    use super::*;
+    #[test]
+    fn test_greatest_takes_string_vector() {
+        let function = GreatestFunction;
+        assert_eq!(
+            function
+                .return_type(&[
+                    ConcreteDataType::string_datatype(),
+                    ConcreteDataType::string_datatype()
+                ])
+                .unwrap(),
+            ConcreteDataType::timestamp_millisecond_datatype()
+        );
+        let columns = vec![
+            Arc::new(StringVector::from(vec![
+                "1970-01-01".to_string(),
+                "2012-12-23".to_string(),
+            ])) as _,
+            Arc::new(StringVector::from(vec![
+                "2001-02-01".to_string(),
+                "1999-01-01".to_string(),
+            ])) as _,
+        ];
+
+        let result = function
+            .eval(&FunctionContext::default(), &columns)
+            .unwrap();
+        let result = result
+            .as_any()
+            .downcast_ref::<TimestampMillisecondVector>()
+            .unwrap();
+        assert_eq!(result.len(), 2);
+        assert_eq!(
+            result.get(0),
+            Value::Timestamp(Timestamp::from_str("2001-02-01 00:00:00", None).unwrap())
+        );
+        assert_eq!(
+            result.get(1),
+            Value::Timestamp(Timestamp::from_str("2012-12-23 00:00:00", None).unwrap())
+        );
+    }
+
+    #[test]
+    fn test_greatest_takes_date_vector() {
+        let function = GreatestFunction;
+        assert_eq!(
+            function
+                .return_type(&[
+                    ConcreteDataType::date_datatype(),
+                    ConcreteDataType::date_datatype()
+                ])
+                .unwrap(),
+            ConcreteDataType::Date(DateType)
+        );
+
+        let columns = vec![
+            Arc::new(DateVector::from_slice(vec![-1, 2])) as _,
+            Arc::new(DateVector::from_slice(vec![0, 1])) as _,
+        ];
+
+        let result = function
+            .eval(&FunctionContext::default(), &columns)
+            .unwrap();
+        let result = result.as_any().downcast_ref::<DateVector>().unwrap();
+        assert_eq!(result.len(), 2);
+        assert_eq!(
+            result.get(0),
+            Value::Date(Date::from_str_utc("1970-01-01").unwrap())
+        );
+        assert_eq!(
+            result.get(1),
+            Value::Date(Date::from_str_utc("1970-01-03").unwrap())
+        );
+    }
+
+    #[test]
+    fn test_greatest_takes_datetime_vector() {
+        let function = GreatestFunction;
+        assert_eq!(
+            function
+                .return_type(&[
+                    ConcreteDataType::timestamp_millisecond_datatype(),
+                    ConcreteDataType::timestamp_millisecond_datatype()
+                ])
+                .unwrap(),
+            ConcreteDataType::timestamp_millisecond_datatype()
+        );
+
+        let columns = vec![
+            Arc::new(TimestampMillisecondVector::from_slice(vec![-1, 2])) as _,
+            Arc::new(TimestampMillisecondVector::from_slice(vec![0, 1])) as _,
+        ];
+
+        let result = function
+            .eval(&FunctionContext::default(), &columns)
+            .unwrap();
+        let result = result
+            .as_any()
+            .downcast_ref::<TimestampMillisecondVector>()
+            .unwrap();
+        assert_eq!(result.len(), 2);
+        assert_eq!(
+            result.get(0),
+            Value::Timestamp(Timestamp::from_str("1970-01-01 00:00:00", None).unwrap())
+        );
+        assert_eq!(
+            result.get(1),
+            Value::Timestamp(Timestamp::from_str("1970-01-01 00:00:00.002", None).unwrap())
+        );
+    }
+
+    macro_rules! test_timestamp {
+        ($type: expr,$unit: ident) => {
+            paste! {
+                #[test]
+                fn [<test_greatest_takes_ $unit:lower _vector>]() {
+                    let function = GreatestFunction;
+                    assert_eq!(
+                        function.return_type(&[$type, $type]).unwrap(),
+                        ConcreteDataType::Timestamp(TimestampType::$unit([<Timestamp $unit Type>]))
+                    );
+
+                    let columns = vec![
+                        Arc::new([<Timestamp $unit Vector>]::from_slice(vec![-1, 2])) as _,
+                        Arc::new([<Timestamp $unit Vector>]::from_slice(vec![0, 1])) as _,
+                    ];
+
+                    let result = function.eval(&FunctionContext::default(), &columns).unwrap();
+                    let result = result.as_any().downcast_ref::<[<Timestamp $unit Vector>]>().unwrap();
+                    assert_eq!(result.len(), 2);
+                    assert_eq!(
+                        result.get(0),
+                        Value::Timestamp(Timestamp::new(0, TimeUnit::$unit))
+                    );
+                    assert_eq!(
+                        result.get(1),
+                        Value::Timestamp(Timestamp::new(2, TimeUnit::$unit))
+                    );
+                }
+            }
+        }
+    }
+
+    test_timestamp!(
+        ConcreteDataType::timestamp_nanosecond_datatype(),
+        Nanosecond
+    );
+    test_timestamp!(
+        ConcreteDataType::timestamp_microsecond_datatype(),
+        Microsecond
+    );
+    test_timestamp!(
+        ConcreteDataType::timestamp_millisecond_datatype(),
+        Millisecond
+    );
+    test_timestamp!(ConcreteDataType::timestamp_second_datatype(), Second);
+}
--- a/src/common/function/src/scalars/uddsketch_calc.rs
+++ b/src/common/function/src/scalars/uddsketch_calc.rs
@@ -115,13 +115,6 @@ impl Function for UddSketchCalcFunction {
                }
            };

-            // Check if the sketch is empty, if so, return null
-            // This is important to avoid panics when calling estimate_quantile on an empty sketch
-            // In practice, this will happen if input is all null
-            if sketch.bucket_iter().count() == 0 {
-                builder.push_null();
-                continue;
-            }
            // Compute the estimated quantile from the sketch
            let result = sketch.estimate_quantile(perc);
            builder.push(Some(result));
--- a/src/common/grpc/src/flight.rs
+++ b/src/common/grpc/src/flight.rs
@@ -192,10 +192,6 @@ impl FlightDecoder {
            }
        }
    }
-
-    pub fn schema(&self) -> Option<&SchemaRef> {
-        self.schema.as_ref()
-    }
 }

 pub fn flight_messages_to_recordbatches(messages: Vec<FlightMessage>) -> Result<RecordBatches> {
--- a/src/common/grpc/src/lib.rs
+++ b/src/common/grpc/src/lib.rs
@@ -18,5 +18,4 @@ pub mod flight;
 pub mod precision;
 pub mod select;

-pub use arrow_flight::FlightData;
 pub use error::Error;
--- a/src/common/meta/src/cache/flow/table_flownode.rs
+++ b/src/common/meta/src/cache/flow/table_flownode.rs
@@ -187,7 +187,6 @@ mod tests {
                    },
                    flownode_ids: BTreeMap::from([(0, 1), (1, 2), (2, 3)]),
                    catalog_name: DEFAULT_CATALOG_NAME.to_string(),
-                    query_context: None,
                    flow_name: "my_flow".to_string(),
                    raw_sql: "sql".to_string(),
                    expire_after: Some(300),
--- a/src/common/meta/src/ddl/create_flow.rs
+++ b/src/common/meta/src/ddl/create_flow.rs
@@ -449,7 +449,6 @@ impl From<&CreateFlowData> for (FlowInfoValue, Vec<(FlowPartitionId, FlowRouteVa
            sink_table_name,
            flownode_ids,
            catalog_name,
-            query_context: Some(value.query_context.clone()),
            flow_name,
            raw_sql: sql,
            expire_after,
--- a/src/common/meta/src/error.rs
+++ b/src/common/meta/src/error.rs
@@ -790,14 +790,6 @@ pub enum Error {
        #[snafu(source)]
        source: common_procedure::error::Error,
    },
-
-    #[snafu(display("Failed to parse timezone"))]
-    InvalidTimeZone {
-        #[snafu(implicit)]
-        location: Location,
-        #[snafu(source)]
-        error: common_time::error::Error,
-    },
 }

 pub type Result<T> = std::result::Result<T, Error>;
@@ -869,8 +861,7 @@ impl ErrorExt for Error {
            | InvalidSetDatabaseOption { .. }
            | InvalidUnsetDatabaseOption { .. }
            | InvalidTopicNamePrefix { .. }
-            | InvalidTimeZone { .. } => StatusCode::InvalidArguments,
-            InvalidFlowRequestBody { .. } => StatusCode::InvalidArguments,
+            | InvalidFlowRequestBody { .. } => StatusCode::InvalidArguments,

            FlowNotFound { .. } => StatusCode::FlowNotFound,
            FlowRouteNotFound { .. } => StatusCode::Unexpected,
--- a/src/common/meta/src/instruction.rs
+++ b/src/common/meta/src/instruction.rs
@@ -217,9 +217,7 @@ pub enum Instruction {
    /// Invalidates batch cache.
    InvalidateCaches(Vec<CacheIdent>),
    /// Flushes regions.
-    FlushRegions(FlushRegions),
-    /// Flushes a single region.
-    FlushRegion(RegionId),
+    FlushRegion(FlushRegions),
 }

 /// The reply of [UpgradeRegion].
@@ -250,7 +248,6 @@ pub enum InstructionReply {
    CloseRegion(SimpleReply),
    UpgradeRegion(UpgradeRegionReply),
    DowngradeRegion(DowngradeRegionReply),
-    FlushRegion(SimpleReply),
 }

 impl Display for InstructionReply {
@@ -262,7 +259,6 @@ impl Display for InstructionReply {
            Self::DowngradeRegion(reply) => {
                write!(f, "InstructionReply::DowngradeRegion({})", reply)
            }
-            Self::FlushRegion(reply) => write!(f, "InstructionReply::FlushRegion({})", reply),
        }
    }
 }
--- a/src/common/meta/src/key.rs
+++ b/src/common/meta/src/key.rs
@@ -112,7 +112,7 @@ pub mod test_utils;
 mod tombstone;
 pub mod topic_name;
 pub mod topic_region;
-pub mod txn_helper;
+pub(crate) mod txn_helper;
 pub mod view_info;

 use std::collections::{BTreeMap, HashMap, HashSet};
--- a/src/common/meta/src/key/flow.rs
+++ b/src/common/meta/src/key/flow.rs
@@ -452,7 +452,6 @@ mod tests {
        };
        FlowInfoValue {
            catalog_name: catalog_name.to_string(),
-            query_context: None,
            flow_name: flow_name.to_string(),
            source_table_ids,
            sink_table_name,
@@ -626,7 +625,6 @@ mod tests {
        };
        let flow_value = FlowInfoValue {
            catalog_name: "greptime".to_string(),
-            query_context: None,
            flow_name: "flow".to_string(),
            source_table_ids: vec![1024, 1025, 1026],
            sink_table_name: another_sink_table_name,
@@ -866,7 +864,6 @@ mod tests {
        };
        let flow_value = FlowInfoValue {
            catalog_name: "greptime".to_string(),
-            query_context: None,
            flow_name: "flow".to_string(),
            source_table_ids: vec![1024, 1025, 1026],
            sink_table_name: another_sink_table_name,
--- a/src/common/meta/src/key/flow/flow_info.rs
+++ b/src/common/meta/src/key/flow/flow_info.rs
@@ -121,13 +121,6 @@ pub struct FlowInfoValue {
    pub(crate) flownode_ids: BTreeMap<FlowPartitionId, FlownodeId>,
    /// The catalog name.
    pub(crate) catalog_name: String,
-    /// The query context used when create flow.
-    /// Although flow doesn't belong to any schema, this query_context is needed to remember
-    /// the query context when `create_flow` is executed
-    /// for recovering flow using the same sql&query_context after db restart.
-    /// if none, should use default query context
-    #[serde(default)]
-    pub(crate) query_context: Option<crate::rpc::ddl::QueryContext>,
    /// The flow name.
    pub(crate) flow_name: String,
    /// The raw sql.
@@ -162,10 +155,6 @@ impl FlowInfoValue {
        &self.catalog_name
    }

-    pub fn query_context(&self) -> &Option<crate::rpc::ddl::QueryContext> {
-        &self.query_context
-    }
-
    pub fn flow_name(&self) -> &String {
        &self.flow_name
    }
--- a/src/common/meta/src/key/table_route.rs
+++ b/src/common/meta/src/key/table_route.rs
@@ -478,11 +478,10 @@ impl TableRouteStorage {
        ))
    }

-    // TODO(LFC): restore its original visibility after some test utility codes are refined
    /// Builds a update table route transaction,
    /// it expected the remote value equals the `current_table_route_value`.
    /// It retrieves the latest value if the comparing failed.
-    pub fn build_update_txn(
+    pub(crate) fn build_update_txn(
        &self,
        table_id: TableId,
        current_table_route_value: &DeserializedValueWithBytes<TableRouteValue>,
--- a/src/common/meta/src/key/txn_helper.rs
+++ b/src/common/meta/src/key/txn_helper.rs
@@ -25,7 +25,7 @@ pub struct TxnOpGetResponseSet(Vec<KeyValue>);

 impl TxnOpGetResponseSet {
    /// Returns a filter to consume a [KeyValue] where the key equals `key`.
-    pub fn filter(key: Vec<u8>) -> impl FnMut(&mut TxnOpGetResponseSet) -> Option<Vec<u8>> {
+    pub(crate) fn filter(key: Vec<u8>) -> impl FnMut(&mut TxnOpGetResponseSet) -> Option<Vec<u8>> {
        move |set| {
            let pos = set.0.iter().position(|kv| kv.key == key);
            match pos {
@@ -36,7 +36,7 @@ impl TxnOpGetResponseSet {
    }

    /// Returns a decoder to decode bytes to `DeserializedValueWithBytes<T>`.
-    pub fn decode_with<F, T>(
+    pub(crate) fn decode_with<F, T>(
        mut f: F,
    ) -> impl FnMut(&mut TxnOpGetResponseSet) -> Result<Option<DeserializedValueWithBytes<T>>>
    where
--- a/src/common/meta/src/kv_backend.rs
+++ b/src/common/meta/src/kv_backend.rs
@@ -35,7 +35,7 @@ pub mod memory;
 pub mod rds;
 pub mod test;
 pub mod txn;
-pub mod util;
+
 pub type KvBackendRef<E = Error> = Arc<dyn KvBackend<Error = E> + Send + Sync>;

 #[async_trait]
--- a/src/common/meta/src/kv_backend/util.rs
+++ b/src/common/meta/src/kv_backend/util.rs
@@ -1,85 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// 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.
-
-/// Removes sensitive information like passwords from connection strings.
-///
-/// This function sanitizes connection strings by removing credentials:
-/// - For URL format (mysql://user:password@host:port/db): Removes everything before '@'
-/// - For parameter format (host=localhost password=secret): Removes the password parameter
-/// - For URL format without credentials (mysql://host:port/db): Removes the protocol prefix
-///
-/// # Arguments
-///
-/// * `conn_str` - The connection string to sanitize
-///
-/// # Returns
-///
-/// A sanitized version of the connection string with sensitive information removed
-pub fn sanitize_connection_string(conn_str: &str) -> String {
-    // Case 1: URL format with credentials (mysql://user:password@host:port/db)
-    // Extract everything after the '@' symbol
-    if let Some(at_pos) = conn_str.find('@') {
-        return conn_str[at_pos + 1..].to_string();
-    }
-
-    // Case 2: Parameter format with password (host=localhost password=secret dbname=mydb)
-    // Filter out any parameter that starts with "password="
-    if conn_str.contains("password=") {
-        return conn_str
-            .split_whitespace()
-            .filter(|param| !param.starts_with("password="))
-            .collect::<Vec<_>>()
-            .join(" ");
-    }
-
-    // Case 3: URL format without credentials (mysql://host:port/db)
-    // Extract everything after the protocol prefix
-    if let Some(host_part) = conn_str.split("://").nth(1) {
-        return host_part.to_string();
-    }
-
-    // Case 4: Already sanitized or unknown format
-    // Return as is
-    conn_str.to_string()
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_sanitize_connection_string() {
-        // Test URL format with username/password
-        let conn_str = "mysql://user:password123@localhost:3306/db";
-        assert_eq!(sanitize_connection_string(conn_str), "localhost:3306/db");
-
-        // Test URL format without credentials
-        let conn_str = "mysql://localhost:3306/db";
-        assert_eq!(sanitize_connection_string(conn_str), "localhost:3306/db");
-
-        // Test parameter format with password
-        let conn_str = "host=localhost port=5432 user=postgres password=secret dbname=mydb";
-        assert_eq!(
-            sanitize_connection_string(conn_str),
-            "host=localhost port=5432 user=postgres dbname=mydb"
-        );
-
-        // Test parameter format without password
-        let conn_str = "host=localhost port=5432 user=postgres dbname=mydb";
-        assert_eq!(
-            sanitize_connection_string(conn_str),
-            "host=localhost port=5432 user=postgres dbname=mydb"
-        );
-    }
-}
--- a/src/common/meta/src/lib.rs
+++ b/src/common/meta/src/lib.rs
@@ -15,6 +15,8 @@
 #![feature(assert_matches)]
 #![feature(btree_extract_if)]
 #![feature(let_chains)]
+#![feature(extract_if)]
+#![feature(hash_extract_if)]

 pub mod cache;
 pub mod cache_invalidator;
--- a/src/common/meta/src/lock_key.rs
+++ b/src/common/meta/src/lock_key.rs
@@ -121,8 +121,8 @@ pub enum FlowNameLock {
 }

 impl FlowNameLock {
-    pub fn new(catalog: &str, flow_name: &str) -> Self {
-        Self::Write(format!("{catalog}.{flow_name}"))
+    pub fn new(catalog: &str, table: &str) -> Self {
+        Self::Write(format!("{catalog}.{table}"))
    }
 }

--- a/src/common/meta/src/region_registry.rs
+++ b/src/common/meta/src/region_registry.rs
@@ -113,10 +113,8 @@ impl LeaderRegionManifestInfo {
    pub fn prunable_entry_id(&self) -> u64 {
        match self {
            LeaderRegionManifestInfo::Mito {
-                flushed_entry_id,
-                topic_latest_entry_id,
-                ..
-            } => (*flushed_entry_id).max(*topic_latest_entry_id),
+                flushed_entry_id, ..
+            } => *flushed_entry_id,
            LeaderRegionManifestInfo::Metric {
                data_flushed_entry_id,
                data_topic_latest_entry_id,
--- a/src/common/meta/src/rpc/ddl.rs
+++ b/src/common/meta/src/rpc/ddl.rs
@@ -35,20 +35,17 @@ use api::v1::{
 };
 use base64::engine::general_purpose;
 use base64::Engine as _;
-use common_time::{DatabaseTimeToLive, Timezone};
+use common_time::DatabaseTimeToLive;
 use prost::Message;
 use serde::{Deserialize, Serialize};
 use serde_with::{serde_as, DefaultOnNull};
-use session::context::{QueryContextBuilder, QueryContextRef};
+use session::context::QueryContextRef;
 use snafu::{OptionExt, ResultExt};
 use table::metadata::{RawTableInfo, TableId};
 use table::table_name::TableName;
 use table::table_reference::TableReference;

-use crate::error::{
-    self, InvalidSetDatabaseOptionSnafu, InvalidTimeZoneSnafu, InvalidUnsetDatabaseOptionSnafu,
-    Result,
-};
+use crate::error::{self, InvalidSetDatabaseOptionSnafu, InvalidUnsetDatabaseOptionSnafu, Result};
 use crate::key::FlowId;

 /// DDL tasks
@@ -1205,7 +1202,7 @@ impl From<DropFlowTask> for PbDropFlowTask {
    }
 }

-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct QueryContext {
    current_catalog: String,
    current_schema: String,
@@ -1226,19 +1223,6 @@ impl From<QueryContextRef> for QueryContext {
    }
 }

-impl TryFrom<QueryContext> for session::context::QueryContext {
-    type Error = error::Error;
-    fn try_from(value: QueryContext) -> std::result::Result<Self, Self::Error> {
-        Ok(QueryContextBuilder::default()
-            .current_catalog(value.current_catalog)
-            .current_schema(value.current_schema)
-            .timezone(Timezone::from_tz_string(&value.timezone).context(InvalidTimeZoneSnafu)?)
-            .extensions(value.extensions)
-            .channel((value.channel as u32).into())
-            .build())
-    }
-}
-
 impl From<QueryContext> for PbQueryContext {
    fn from(
        QueryContext {
--- a/src/common/meta/src/rpc/router.rs
+++ b/src/common/meta/src/rpc/router.rs
@@ -176,12 +176,15 @@ impl TableRoute {
                })?
                .into();

-            let leader_peer = peers.get(region_route.leader_peer_index as usize).cloned();
+            let leader_peer = peers
+                .get(region_route.leader_peer_index as usize)
+                .cloned()
+                .map(Into::into);

            let follower_peers = region_route
                .follower_peer_indexes
                .into_iter()
-                .filter_map(|x| peers.get(x as usize).cloned())
+                .filter_map(|x| peers.get(x as usize).cloned().map(Into::into))
                .collect::<Vec<_>>();

            region_routes.push(RegionRoute {
--- a/src/common/procedure-test/src/lib.rs
+++ b/src/common/procedure-test/src/lib.rs
@@ -18,13 +18,11 @@ use std::collections::HashMap;
 use std::sync::Arc;

 use async_trait::async_trait;
-use common_procedure::local::{acquire_dynamic_key_lock, DynamicKeyLockGuard};
-use common_procedure::rwlock::KeyRwLock;
 use common_procedure::store::poison_store::PoisonStore;
 use common_procedure::test_util::InMemoryPoisonStore;
 use common_procedure::{
    Context, ContextProvider, Output, PoisonKey, Procedure, ProcedureId, ProcedureState,
-    ProcedureWithId, Result, Status, StringKey,
+    ProcedureWithId, Result, Status,
 };

 /// A Mock [ContextProvider].
@@ -32,7 +30,6 @@ use common_procedure::{
 pub struct MockContextProvider {
    states: HashMap<ProcedureId, ProcedureState>,
    poison_manager: InMemoryPoisonStore,
-    dynamic_key_lock: Arc<KeyRwLock<String>>,
 }

 impl MockContextProvider {
@@ -41,7 +38,6 @@ impl MockContextProvider {
        MockContextProvider {
            states,
            poison_manager: InMemoryPoisonStore::default(),
-            dynamic_key_lock: Arc::new(KeyRwLock::new()),
        }
    }

@@ -62,10 +58,6 @@ impl ContextProvider for MockContextProvider {
            .try_put_poison(key.to_string(), procedure_id.to_string())
            .await
    }
-
-    async fn acquire_lock(&self, key: &StringKey) -> DynamicKeyLockGuard {
-        acquire_dynamic_key_lock(&self.dynamic_key_lock, key).await
-    }
 }

 /// Executes a procedure until it returns [Status::Done].
--- a/src/common/procedure/src/lib.rs
+++ b/src/common/procedure/src/lib.rs
@@ -20,7 +20,6 @@ pub mod error;
 pub mod local;
 pub mod options;
 mod procedure;
-pub mod rwlock;
 pub mod store;
 pub mod watcher;

@@ -29,8 +28,8 @@ pub mod test_util;

 pub use crate::error::{Error, Result};
 pub use crate::procedure::{
-    BoxedProcedure, BoxedProcedureLoader, Context, ContextProvider, ContextProviderRef, LockKey,
-    Output, ParseIdError, PoisonKey, PoisonKeys, Procedure, ProcedureId, ProcedureInfo,
-    ProcedureManager, ProcedureManagerRef, ProcedureState, ProcedureWithId, Status, StringKey,
+    BoxedProcedure, BoxedProcedureLoader, Context, ContextProvider, LockKey, Output, ParseIdError,
+    PoisonKey, PoisonKeys, Procedure, ProcedureId, ProcedureInfo, ProcedureManager,
+    ProcedureManagerRef, ProcedureState, ProcedureWithId, Status, StringKey,
 };
 pub use crate::watcher::Watcher;
--- a/src/common/procedure/src/local.rs
+++ b/src/common/procedure/src/local.rs
@@ -13,6 +13,7 @@
 // limitations under the License.

 mod runner;
+mod rwlock;

 use std::collections::hash_map::Entry;
 use std::collections::{HashMap, HashSet, VecDeque};
@@ -29,6 +30,7 @@ use snafu::{ensure, OptionExt, ResultExt};
 use tokio::sync::watch::{self, Receiver, Sender};
 use tokio::sync::{Mutex as TokioMutex, Notify};

+use self::rwlock::KeyRwLock;
 use crate::error::{
    self, DuplicateProcedureSnafu, Error, LoaderConflictSnafu, ManagerNotStartSnafu,
    PoisonKeyNotDefinedSnafu, ProcedureNotFoundSnafu, Result, StartRemoveOutdatedMetaTaskSnafu,
@@ -36,12 +38,11 @@ use crate::error::{
 };
 use crate::local::runner::Runner;
 use crate::procedure::{BoxedProcedureLoader, InitProcedureState, PoisonKeys, ProcedureInfo};
-use crate::rwlock::{KeyRwLock, OwnedKeyRwLockGuard};
 use crate::store::poison_store::PoisonStoreRef;
 use crate::store::{ProcedureMessage, ProcedureMessages, ProcedureStore, StateStoreRef};
 use crate::{
    BoxedProcedure, ContextProvider, LockKey, PoisonKey, ProcedureId, ProcedureManager,
-    ProcedureState, ProcedureWithId, StringKey, Watcher,
+    ProcedureState, ProcedureWithId, Watcher,
 };

 /// The expired time of a procedure's metadata.
@@ -156,80 +157,12 @@ struct LoadedProcedure {
    step: u32,
 }

-/// The dynamic lock for procedure execution.
-///
-/// Unlike the procedure-level locks, these locks are acquired dynamically by the procedure
-/// during execution. They are only held when the procedure specifically needs these keys
-/// and are released as soon as the procedure no longer needs them.
-/// This allows for more fine-grained concurrency control during procedure execution.
-pub(crate) type DynamicKeyLock = Arc<KeyRwLock<String>>;
-
-/// Acquires a dynamic key lock for the given key.
-///
-/// This function takes a reference to the dynamic key lock and a pointer to the key.
-/// It then matches the key type and acquires the appropriate lock.
-pub async fn acquire_dynamic_key_lock(
-    lock: &DynamicKeyLock,
-    key: &StringKey,
-) -> DynamicKeyLockGuard {
-    match key {
-        StringKey::Share(key) => {
-            let guard = lock.read(key.to_string()).await;
-            DynamicKeyLockGuard {
-                guard: Some(OwnedKeyRwLockGuard::from(guard)),
-                key: key.to_string(),
-                lock: lock.clone(),
-            }
-        }
-        StringKey::Exclusive(key) => {
-            let guard = lock.write(key.to_string()).await;
-            DynamicKeyLockGuard {
-                guard: Some(OwnedKeyRwLockGuard::from(guard)),
-                key: key.to_string(),
-                lock: lock.clone(),
-            }
-        }
-    }
-}
-/// A guard for the dynamic key lock.
-///
-/// This guard is used to release the lock when the procedure no longer needs it.
-/// It also ensures that the lock is cleaned up when the guard is dropped.
-pub struct DynamicKeyLockGuard {
-    guard: Option<OwnedKeyRwLockGuard>,
-    key: String,
-    lock: DynamicKeyLock,
-}
-
-impl Drop for DynamicKeyLockGuard {
-    fn drop(&mut self) {
-        if let Some(guard) = self.guard.take() {
-            drop(guard);
-        }
-        self.lock.clean_keys(&[self.key.to_string()]);
-    }
-}
-
 /// Shared context of the manager.
 pub(crate) struct ManagerContext {
    /// Procedure loaders. The key is the type name of the procedure which the loader returns.
    loaders: Mutex<HashMap<String, BoxedProcedureLoader>>,
-    /// The key lock for the procedure.
-    ///
-    /// The lock keys are defined in `Procedure::lock_key()`.
-    /// These locks are acquired before the procedure starts and released after the procedure finishes.
-    /// They ensure exclusive access to resources throughout the entire procedure lifecycle.
    key_lock: KeyRwLock<String>,
-    /// The dynamic lock for procedure execution.
-    ///
-    /// Unlike the procedure-level locks, these locks are acquired dynamically by the procedure
-    /// during execution. They are only held when the procedure specifically needs these keys
-    /// and are released as soon as the procedure no longer needs them.
-    /// This allows for more fine-grained concurrency control during procedure execution.
-    dynamic_key_lock: DynamicKeyLock,
-    /// Procedures in the manager.
    procedures: RwLock<HashMap<ProcedureId, ProcedureMetaRef>>,
-    /// Running procedures.
    running_procedures: Mutex<HashSet<ProcedureId>>,
    /// Ids and finished time of finished procedures.
    finished_procedures: Mutex<VecDeque<(ProcedureId, Instant)>>,
@@ -266,10 +199,6 @@ impl ContextProvider for ManagerContext {
        let procedure_id = procedure_id.to_string();
        self.poison_manager.try_put_poison(key, procedure_id).await
    }
-
-    async fn acquire_lock(&self, key: &StringKey) -> DynamicKeyLockGuard {
-        acquire_dynamic_key_lock(&self.dynamic_key_lock, key).await
-    }
 }

 impl ManagerContext {
@@ -277,7 +206,6 @@ impl ManagerContext {
    fn new(poison_manager: PoisonStoreRef) -> ManagerContext {
        ManagerContext {
            key_lock: KeyRwLock::new(),
-            dynamic_key_lock: Arc::new(KeyRwLock::new()),
            loaders: Mutex::new(HashMap::new()),
            procedures: RwLock::new(HashMap::new()),
            running_procedures: Mutex::new(HashSet::new()),
--- a/src/common/procedure/src/local/runner.rs
+++ b/src/common/procedure/src/local/runner.rs
@@ -23,9 +23,9 @@ use snafu::ResultExt;
 use tokio::time;

 use crate::error::{self, ProcedurePanicSnafu, Result, RollbackTimesExceededSnafu};
+use crate::local::rwlock::OwnedKeyRwLockGuard;
 use crate::local::{ManagerContext, ProcedureMeta, ProcedureMetaRef};
 use crate::procedure::{Output, StringKey};
-use crate::rwlock::OwnedKeyRwLockGuard;
 use crate::store::{ProcedureMessage, ProcedureStore};
 use crate::{
    BoxedProcedure, Context, Error, Procedure, ProcedureId, ProcedureState, ProcedureWithId, Status,
@@ -581,7 +581,6 @@ impl Runner {

 #[cfg(test)]
 mod tests {
-    use std::sync::atomic::{AtomicU64, Ordering};
    use std::sync::Arc;

    use async_trait::async_trait;
@@ -589,14 +588,13 @@ mod tests {
    use common_error::mock::MockError;
    use common_error::status_code::StatusCode;
    use common_test_util::temp_dir::create_temp_dir;
-    use futures::future::join_all;
    use futures_util::future::BoxFuture;
    use futures_util::FutureExt;
    use object_store::{EntryMode, ObjectStore};
    use tokio::sync::mpsc;

    use super::*;
-    use crate::local::{test_util, DynamicKeyLockGuard};
+    use crate::local::test_util;
    use crate::procedure::PoisonKeys;
    use crate::store::proc_path;
    use crate::test_util::InMemoryPoisonStore;
@@ -668,10 +666,6 @@ mod tests {
            ) -> Result<()> {
                unimplemented!()
            }
-
-            async fn acquire_lock(&self, _key: &StringKey) -> DynamicKeyLockGuard {
-                unimplemented!()
-            }
        }

        Context {
@@ -1680,66 +1674,4 @@ mod tests {
        // If the procedure is poisoned, the poison key shouldn't be deleted.
        assert_eq!(procedure_id, ROOT_ID);
    }
-
-    fn test_procedure_with_dynamic_lock(
-        shared_atomic_value: Arc<AtomicU64>,
-        id: u64,
-    ) -> (BoxedProcedure, Arc<ProcedureMeta>) {
-        let exec_fn = move |ctx: Context| {
-            let moved_shared_atomic_value = shared_atomic_value.clone();
-            let moved_ctx = ctx.clone();
-            async move {
-                debug!("Acquiring write lock, id: {}", id);
-                let key = StringKey::Exclusive("test_lock".to_string());
-                let guard = moved_ctx.provider.acquire_lock(&key).await;
-                debug!("Acquired write lock, id: {}", id);
-                let millis = rand::rng().random_range(10..=50);
-                tokio::time::sleep(Duration::from_millis(millis)).await;
-                let value = moved_shared_atomic_value.load(Ordering::Relaxed);
-                moved_shared_atomic_value.store(value + 1, Ordering::Relaxed);
-                debug!("Dropping write lock, id: {}", id);
-                drop(guard);
-
-                Ok(Status::done())
-            }
-            .boxed()
-        };
-
-        let adapter = ProcedureAdapter {
-            data: "dynamic_lock".to_string(),
-            lock_key: LockKey::new_exclusive([]),
-            poison_keys: PoisonKeys::new([]),
-            exec_fn,
-            rollback_fn: None,
-        };
-        let meta = adapter.new_meta(ROOT_ID);
-
-        (Box::new(adapter), meta)
-    }
-
-    #[tokio::test(flavor = "multi_thread")]
-    async fn test_execute_with_dynamic_lock() {
-        common_telemetry::init_default_ut_logging();
-        let shared_atomic_value = Arc::new(AtomicU64::new(0));
-        let (procedure1, meta1) = test_procedure_with_dynamic_lock(shared_atomic_value.clone(), 1);
-        let (procedure2, meta2) = test_procedure_with_dynamic_lock(shared_atomic_value.clone(), 2);
-
-        let dir = create_temp_dir("dynamic_lock");
-        let object_store = test_util::new_object_store(&dir);
-        let procedure_store = Arc::new(ProcedureStore::from_object_store(object_store.clone()));
-        let mut runner1 = new_runner(meta1.clone(), procedure1, procedure_store.clone());
-        let mut runner2 = new_runner(meta2.clone(), procedure2, procedure_store.clone());
-        let ctx1 = context_with_provider(
-            meta1.id,
-            runner1.manager_ctx.clone() as Arc<dyn ContextProvider>,
-        );
-        let ctx2 = context_with_provider(
-            meta2.id,
-            // use same manager ctx as runner1
-            runner1.manager_ctx.clone() as Arc<dyn ContextProvider>,
-        );
-        let tasks = [runner1.execute_once(&ctx1), runner2.execute_once(&ctx2)];
-        join_all(tasks).await;
-        assert_eq!(shared_atomic_value.load(Ordering::Relaxed), 2);
-    }
 }
--- a/src/common/procedure/src/local/rwlock.rs
+++ b/src/common/procedure/src/local/rwlock.rs
@@ -18,18 +18,8 @@ use std::sync::{Arc, Mutex};

 use tokio::sync::{OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock};

-/// A guard that owns a read or write lock on a key.
-///
-/// This enum wraps either a read or write lock guard obtained from a `KeyRwLock`.
-/// The guard is automatically released when it is dropped.
 pub enum OwnedKeyRwLockGuard {
-    /// Represents a shared read lock on a key.
-    /// Multiple read locks can be held simultaneously for the same key.
    Read { _guard: OwnedRwLockReadGuard<()> },
-
-    /// Represents an exclusive write lock on a key.
-    /// Only one write lock can be held at a time for a given key,
-    /// and no read locks can be held simultaneously with a write lock.
    Write { _guard: OwnedRwLockWriteGuard<()> },
 }

@@ -46,7 +36,7 @@ impl From<OwnedRwLockWriteGuard<()>> for OwnedKeyRwLockGuard {
 }

 /// Locks based on a key, allowing other keys to lock independently.
-#[derive(Debug, Default)]
+#[derive(Debug)]
 pub struct KeyRwLock<K> {
    /// The inner map of locks for specific keys.
    inner: Mutex<HashMap<K, Arc<RwLock<()>>>>,
--- a/src/common/procedure/src/procedure.rs
+++ b/src/common/procedure/src/procedure.rs
@@ -25,7 +25,6 @@ use snafu::{ResultExt, Snafu};
 use uuid::Uuid;

 use crate::error::{self, Error, Result};
-use crate::local::DynamicKeyLockGuard;
 use crate::watcher::Watcher;

 pub type Output = Arc<dyn Any + Send + Sync>;
@@ -145,9 +144,6 @@ pub trait ContextProvider: Send + Sync {
    /// This method is used to mark a resource as being operated on by a procedure.
    /// If the poison key already exists with a different value, the operation will fail.
    async fn try_put_poison(&self, key: &PoisonKey, procedure_id: ProcedureId) -> Result<()>;
-
-    /// Acquires a key lock for the procedure.
-    async fn acquire_lock(&self, key: &StringKey) -> DynamicKeyLockGuard;
 }

 /// Reference-counted pointer to [ContextProvider].
--- a/src/common/query/src/logical_plan/accumulator.rs
+++ b/src/common/query/src/logical_plan/accumulator.rs
@@ -24,7 +24,7 @@ use datatypes::prelude::*;
 use datatypes::vectors::{Helper as VectorHelper, VectorRef};
 use snafu::ResultExt;

-use crate::error::{self, FromScalarValueSnafu, IntoVectorSnafu, Result};
+use crate::error::{self, Error, FromScalarValueSnafu, IntoVectorSnafu, Result};
 use crate::prelude::*;

 pub type AggregateFunctionCreatorRef = Arc<dyn AggregateFunctionCreator>;
@@ -166,7 +166,8 @@ impl DfAccumulator for DfAccumulatorAdaptor {
        let output_type = self.creator.output_type()?;
        let scalar_value = value
            .try_to_scalar_value(&output_type)
-            .context(error::ToScalarValueSnafu)?;
+            .context(error::ToScalarValueSnafu)
+            .map_err(Error::from)?;
        Ok(scalar_value)
    }

--- a/src/common/query/src/request.rs
+++ b/src/common/query/src/request.rs
@@ -17,7 +17,6 @@ use datafusion_expr::LogicalPlan;
 use store_api::storage::RegionId;

 /// The query request to be handled by the RegionServer (Datanode).
-#[derive(Clone, Debug)]
 pub struct QueryRequest {
    /// The header of this request. Often to store some context of the query. None means all to defaults.
    pub header: Option<RegionRequestHeader>,
--- a/src/datanode/src/datanode.rs
+++ b/src/datanode/src/datanode.rs
@@ -43,10 +43,10 @@ use mito2::config::MitoConfig;
 use mito2::engine::MitoEngine;
 use object_store::manager::{ObjectStoreManager, ObjectStoreManagerRef};
 use object_store::util::normalize_dir;
-use query::dummy_catalog::TableProviderFactoryRef;
 use query::QueryEngineFactory;
 use servers::export_metrics::ExportMetricsTask;
 use servers::server::ServerHandlers;
+use servers::Mode;
 use snafu::{ensure, OptionExt, ResultExt};
 use store_api::path_utils::{region_dir, WAL_DIR};
 use store_api::region_engine::{RegionEngineRef, RegionRole};
@@ -57,8 +57,8 @@ use tokio::sync::Notify;

 use crate::config::{DatanodeOptions, RegionEngineConfig, StorageConfig};
 use crate::error::{
-    self, BuildMetricEngineSnafu, BuildMitoEngineSnafu, CreateDirSnafu, GetMetadataSnafu,
-    MissingCacheSnafu, MissingNodeIdSnafu, OpenLogStoreSnafu, Result, ShutdownInstanceSnafu,
+    self, BuildMitoEngineSnafu, CreateDirSnafu, GetMetadataSnafu, MissingCacheSnafu,
+    MissingKvBackendSnafu, MissingNodeIdSnafu, OpenLogStoreSnafu, Result, ShutdownInstanceSnafu,
    ShutdownServerSnafu, StartServerSnafu,
 };
 use crate::event_listener::{
@@ -129,7 +129,7 @@ impl Datanode {
        self.services = services;
    }

-    pub async fn shutdown(&mut self) -> Result<()> {
+    pub async fn shutdown(&self) -> Result<()> {
        self.services
            .shutdown_all()
            .await
@@ -157,49 +157,50 @@ impl Datanode {

 pub struct DatanodeBuilder {
    opts: DatanodeOptions,
-    table_provider_factory: Option<TableProviderFactoryRef>,
+    mode: Mode,
    plugins: Plugins,
    meta_client: Option<MetaClientRef>,
-    kv_backend: KvBackendRef,
+    kv_backend: Option<KvBackendRef>,
    cache_registry: Option<Arc<LayeredCacheRegistry>>,
 }

 impl DatanodeBuilder {
-    pub fn new(opts: DatanodeOptions, plugins: Plugins, kv_backend: KvBackendRef) -> Self {
+    /// `kv_backend` is optional. If absent, the builder will try to build one
+    /// by using the given `opts`
+    pub fn new(opts: DatanodeOptions, plugins: Plugins, mode: Mode) -> Self {
        Self {
            opts,
-            table_provider_factory: None,
+            mode,
            plugins,
            meta_client: None,
-            kv_backend,
+            kv_backend: None,
            cache_registry: None,
        }
    }

-    pub fn options(&self) -> &DatanodeOptions {
-        &self.opts
+    pub fn with_meta_client(self, meta_client: MetaClientRef) -> Self {
+        Self {
+            meta_client: Some(meta_client),
+            ..self
+        }
    }

-    pub fn with_meta_client(&mut self, client: MetaClientRef) -> &mut Self {
-        self.meta_client = Some(client);
-        self
+    pub fn with_cache_registry(self, cache_registry: Arc<LayeredCacheRegistry>) -> Self {
+        Self {
+            cache_registry: Some(cache_registry),
+            ..self
+        }
    }

-    pub fn with_cache_registry(&mut self, registry: Arc<LayeredCacheRegistry>) -> &mut Self {
-        self.cache_registry = Some(registry);
-        self
-    }
-
-    pub fn kv_backend(&self) -> &KvBackendRef {
-        &self.kv_backend
-    }
-
-    pub fn with_table_provider_factory(&mut self, factory: TableProviderFactoryRef) -> &mut Self {
-        self.table_provider_factory = Some(factory);
-        self
+    pub fn with_kv_backend(self, kv_backend: KvBackendRef) -> Self {
+        Self {
+            kv_backend: Some(kv_backend),
+            ..self
+        }
    }

    pub async fn build(mut self) -> Result<Datanode> {
+        let mode = &self.mode;
        let node_id = self.opts.node_id.context(MissingNodeIdSnafu)?;

        let meta_client = self.meta_client.take();
@@ -209,6 +210,8 @@ impl DatanodeBuilder {
        // writable upon open.
        let controlled_by_metasrv = meta_client.is_some();

+        let kv_backend = self.kv_backend.take().context(MissingKvBackendSnafu)?;
+
        // build and initialize region server
        let (region_event_listener, region_event_receiver) = if controlled_by_metasrv {
            let (tx, rx) = new_region_server_event_channel();
@@ -230,7 +233,7 @@ impl DatanodeBuilder {
            .new_region_server(schema_metadata_manager, region_event_listener)
            .await?;

-        let datanode_table_manager = DatanodeTableManager::new(self.kv_backend.clone());
+        let datanode_table_manager = DatanodeTableManager::new(kv_backend.clone());
        let table_values = datanode_table_manager
            .tables(node_id)
            .try_collect::<Vec<_>>()
@@ -270,18 +273,19 @@ impl DatanodeBuilder {
            None
        };

-        let is_standalone = heartbeat_task.is_none();
        let greptimedb_telemetry_task = get_greptimedb_telemetry_task(
            Some(self.opts.storage.data_home.clone()),
-            is_standalone && self.opts.enable_telemetry,
+            mode,
+            self.opts.enable_telemetry,
        )
        .await;

-        let leases_notifier = if self.opts.require_lease_before_startup && !is_standalone {
-            Some(Arc::new(Notify::new()))
-        } else {
-            None
-        };
+        let leases_notifier =
+            if self.opts.require_lease_before_startup && matches!(mode, Mode::Distributed) {
+                Some(Arc::new(Notify::new()))
+            } else {
+                None
+            };

        let export_metrics_task =
            ExportMetricsTask::try_new(&self.opts.export_metrics, Some(&self.plugins))
@@ -359,11 +363,7 @@ impl DatanodeBuilder {
        );
        let query_engine = query_engine_factory.query_engine();

-        let table_provider_factory = self
-            .table_provider_factory
-            .clone()
-            .unwrap_or_else(|| Arc::new(DummyTableProviderFactory));
-
+        let table_provider_factory = Arc::new(DummyTableProviderFactory);
        let mut region_server = RegionServer::with_table_provider(
            query_engine,
            common_runtime::global_runtime(),
@@ -398,46 +398,44 @@ impl DatanodeBuilder {
        schema_metadata_manager: SchemaMetadataManagerRef,
        plugins: Plugins,
    ) -> Result<Vec<RegionEngineRef>> {
-        let mut metric_engine_config = metric_engine::config::EngineConfig::default();
-        let mut mito_engine_config = MitoConfig::default();
-        let mut file_engine_config = file_engine::config::EngineConfig::default();
+        let mut engines = vec![];
+        let mut metric_engine_config = opts.region_engine.iter().find_map(|c| match c {
+            RegionEngineConfig::Metric(config) => Some(config.clone()),
+            _ => None,
+        });

        for engine in &opts.region_engine {
            match engine {
                RegionEngineConfig::Mito(config) => {
-                    mito_engine_config = config.clone();
+                    let mito_engine = Self::build_mito_engine(
+                        opts,
+                        object_store_manager.clone(),
+                        config.clone(),
+                        schema_metadata_manager.clone(),
+                        plugins.clone(),
+                    )
+                    .await?;
+
+                    let metric_engine = MetricEngine::new(
+                        mito_engine.clone(),
+                        metric_engine_config.take().unwrap_or_default(),
+                    );
+                    engines.push(Arc::new(mito_engine) as _);
+                    engines.push(Arc::new(metric_engine) as _);
                }
                RegionEngineConfig::File(config) => {
-                    file_engine_config = config.clone();
+                    let engine = FileRegionEngine::new(
+                        config.clone(),
+                        object_store_manager.default_object_store().clone(), // TODO: implement custom storage for file engine
+                    );
+                    engines.push(Arc::new(engine) as _);
                }
-                RegionEngineConfig::Metric(metric_config) => {
-                    metric_engine_config = metric_config.clone();
+                RegionEngineConfig::Metric(_) => {
+                    // Already handled in `build_mito_engine`.
                }
            }
        }
-
-        let mito_engine = Self::build_mito_engine(
-            opts,
-            object_store_manager.clone(),
-            mito_engine_config,
-            schema_metadata_manager.clone(),
-            plugins.clone(),
-        )
-        .await?;
-
-        let metric_engine = MetricEngine::try_new(mito_engine.clone(), metric_engine_config)
-            .context(BuildMetricEngineSnafu)?;
-
-        let file_engine = FileRegionEngine::new(
-            file_engine_config,
-            object_store_manager.default_object_store().clone(), // TODO: implement custom storage for file engine
-        );
-
-        Ok(vec![
-            Arc::new(mito_engine) as _,
-            Arc::new(metric_engine) as _,
-            Arc::new(file_engine) as _,
-        ])
+        Ok(engines)
    }

    /// Builds [MitoEngine] according to options.
@@ -635,6 +633,7 @@ mod tests {
    use common_meta::kv_backend::memory::MemoryKvBackend;
    use common_meta::kv_backend::KvBackendRef;
    use mito2::engine::MITO_ENGINE_NAME;
+    use servers::Mode;
    use store_api::region_request::RegionRequest;
    use store_api::storage::RegionId;

@@ -670,19 +669,19 @@ mod tests {
        let kv_backend = Arc::new(MemoryKvBackend::new());
        let layered_cache_registry = Arc::new(
            LayeredCacheRegistryBuilder::default()
-                .add_cache_registry(build_datanode_cache_registry(kv_backend.clone()))
+                .add_cache_registry(build_datanode_cache_registry(kv_backend))
                .build(),
        );

-        let mut builder = DatanodeBuilder::new(
+        let builder = DatanodeBuilder::new(
            DatanodeOptions {
                node_id: Some(0),
                ..Default::default()
            },
            Plugins::default(),
-            kv_backend,
-        );
-        builder.with_cache_registry(layered_cache_registry);
+            Mode::Standalone,
+        )
+        .with_cache_registry(layered_cache_registry);

        let kv = Arc::new(MemoryKvBackend::default()) as _;
        setup_table_datanode(&kv).await;
--- a/src/datanode/src/error.rs
+++ b/src/datanode/src/error.rs
@@ -150,6 +150,12 @@ pub enum Error {
        location: Location,
    },

+    #[snafu(display("Expect KvBackend but not found"))]
+    MissingKvBackend {
+        #[snafu(implicit)]
+        location: Location,
+    },
+
    #[snafu(display("Invalid SQL, error: {}", msg))]
    InvalidSql { msg: String },

@@ -330,13 +336,6 @@ pub enum Error {
        location: Location,
    },

-    #[snafu(display("Failed to build metric engine"))]
-    BuildMetricEngine {
-        source: metric_engine::error::Error,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
    #[snafu(display("Failed to serialize options to TOML"))]
    TomlFormat {
        #[snafu(implicit)]
@@ -420,6 +419,7 @@ impl ErrorExt for Error {
            | MissingRequiredField { .. }
            | RegionEngineNotFound { .. }
            | ParseAddr { .. }
+            | MissingKvBackend { .. }
            | TomlFormat { .. } => StatusCode::InvalidArguments,

            PayloadNotExist { .. }
@@ -452,7 +452,6 @@ impl ErrorExt for Error {

            FindLogicalRegions { source, .. } => source.status_code(),
            BuildMitoEngine { source, .. } => source.status_code(),
-            BuildMetricEngine { source, .. } => source.status_code(),
            ConcurrentQueryLimiterClosed { .. } | ConcurrentQueryLimiterTimeout { .. } => {
                StatusCode::RegionBusy
            }
--- a/src/datanode/src/greptimedb_telemetry.rs
+++ b/src/datanode/src/greptimedb_telemetry.rs
@@ -20,6 +20,7 @@ use common_greptimedb_telemetry::{
    default_get_uuid, Collector, GreptimeDBTelemetry, GreptimeDBTelemetryTask,
    Mode as VersionReporterMode, TELEMETRY_INTERVAL,
 };
+use servers::Mode;

 struct StandaloneGreptimeDBTelemetryCollector {
    uuid: Option<String>,
@@ -54,6 +55,7 @@ impl Collector for StandaloneGreptimeDBTelemetryCollector {

 pub async fn get_greptimedb_telemetry_task(
    working_home: Option<String>,
+    mode: &Mode,
    enable: bool,
 ) -> Arc<GreptimeDBTelemetryTask> {
    if !enable || cfg!(test) || cfg!(debug_assertions) {
@@ -62,14 +64,19 @@ pub async fn get_greptimedb_telemetry_task(
    // Always enable.
    let should_report = Arc::new(AtomicBool::new(true));

-    let uuid = default_get_uuid(&working_home);
-    Arc::new(GreptimeDBTelemetryTask::enable(
-        TELEMETRY_INTERVAL,
-        Box::new(GreptimeDBTelemetry::new(
-            working_home,
-            Box::new(StandaloneGreptimeDBTelemetryCollector { uuid, retry: 0 }),
-            should_report.clone(),
+    match mode {
+        Mode::Standalone => Arc::new(GreptimeDBTelemetryTask::enable(
+            TELEMETRY_INTERVAL,
+            Box::new(GreptimeDBTelemetry::new(
+                working_home.clone(),
+                Box::new(StandaloneGreptimeDBTelemetryCollector {
+                    uuid: default_get_uuid(&working_home),
+                    retry: 0,
+                }),
+                should_report.clone(),
+            )),
+            should_report,
        )),
-        should_report,
-    ))
+        Mode::Distributed => Arc::new(GreptimeDBTelemetryTask::disable()),
+    }
 }
--- a/src/datanode/src/heartbeat/handler.rs
+++ b/src/datanode/src/heartbeat/handler.rs
@@ -39,7 +39,6 @@ pub struct RegionHeartbeatResponseHandler {
    region_server: RegionServer,
    catchup_tasks: TaskTracker<()>,
    downgrade_tasks: TaskTracker<()>,
-    flush_tasks: TaskTracker<()>,
 }

 /// Handler of the instruction.
@@ -51,7 +50,6 @@ pub struct HandlerContext {
    region_server: RegionServer,
    catchup_tasks: TaskTracker<()>,
    downgrade_tasks: TaskTracker<()>,
-    flush_tasks: TaskTracker<()>,
 }

 impl HandlerContext {
@@ -65,7 +63,6 @@ impl HandlerContext {
            region_server,
            catchup_tasks: TaskTracker::new(),
            downgrade_tasks: TaskTracker::new(),
-            flush_tasks: TaskTracker::new(),
        }
    }
 }
@@ -77,7 +74,6 @@ impl RegionHeartbeatResponseHandler {
            region_server,
            catchup_tasks: TaskTracker::new(),
            downgrade_tasks: TaskTracker::new(),
-            flush_tasks: TaskTracker::new(),
        }
    }

@@ -99,11 +95,8 @@ impl RegionHeartbeatResponseHandler {
                handler_context.handle_upgrade_region_instruction(upgrade_region)
            })),
            Instruction::InvalidateCaches(_) => InvalidHeartbeatResponseSnafu.fail(),
-            Instruction::FlushRegions(flush_regions) => Ok(Box::new(move |handler_context| {
-                handler_context.handle_flush_regions_instruction(flush_regions)
-            })),
-            Instruction::FlushRegion(flush_region) => Ok(Box::new(move |handler_context| {
-                handler_context.handle_flush_region_instruction(flush_region)
+            Instruction::FlushRegion(flush_regions) => Ok(Box::new(move |handler_context| {
+                handler_context.handle_flush_region_instruction(flush_regions)
            })),
        }
    }
@@ -118,7 +111,6 @@ impl HeartbeatResponseHandler for RegionHeartbeatResponseHandler {
                | Some((_, Instruction::CloseRegion { .. }))
                | Some((_, Instruction::DowngradeRegion { .. }))
                | Some((_, Instruction::UpgradeRegion { .. }))
-                | Some((_, Instruction::FlushRegion { .. }))
        )
    }

@@ -132,14 +124,12 @@ impl HeartbeatResponseHandler for RegionHeartbeatResponseHandler {
        let region_server = self.region_server.clone();
        let catchup_tasks = self.catchup_tasks.clone();
        let downgrade_tasks = self.downgrade_tasks.clone();
-        let flush_tasks = self.flush_tasks.clone();
        let handler = Self::build_handler(instruction)?;
        let _handle = common_runtime::spawn_global(async move {
            let reply = handler(HandlerContext {
                region_server,
                catchup_tasks,
                downgrade_tasks,
-                flush_tasks,
            })
            .await;

--- a/src/datanode/src/heartbeat/handler/flush_region.rs
+++ b/src/datanode/src/heartbeat/handler/flush_region.rs
@@ -12,17 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use common_meta::instruction::{FlushRegions, InstructionReply, SimpleReply};
+use common_meta::instruction::{FlushRegions, InstructionReply};
 use common_telemetry::warn;
 use futures_util::future::BoxFuture;
 use store_api::region_request::{RegionFlushRequest, RegionRequest};
-use store_api::storage::RegionId;

 use crate::error;
 use crate::heartbeat::handler::HandlerContext;

 impl HandlerContext {
-    pub(crate) fn handle_flush_regions_instruction(
+    pub(crate) fn handle_flush_region_instruction(
        self,
        flush_regions: FlushRegions,
    ) -> BoxFuture<'static, Option<InstructionReply>> {
@@ -50,59 +49,6 @@ impl HandlerContext {
            None
        })
    }
-
-    pub(crate) fn handle_flush_region_instruction(
-        self,
-        region_id: RegionId,
-    ) -> BoxFuture<'static, Option<InstructionReply>> {
-        Box::pin(async move {
-            let Some(writable) = self.region_server.is_region_leader(region_id) else {
-                return Some(InstructionReply::FlushRegion(SimpleReply {
-                    result: false,
-                    error: Some("Region is not leader".to_string()),
-                }));
-            };
-
-            if !writable {
-                return Some(InstructionReply::FlushRegion(SimpleReply {
-                    result: false,
-                    error: Some("Region is not writable".to_string()),
-                }));
-            }
-
-            let region_server_moved = self.region_server.clone();
-            let register_result = self
-                .flush_tasks
-                .try_register(
-                    region_id,
-                    Box::pin(async move {
-                        let request = RegionRequest::Flush(RegionFlushRequest {
-                            row_group_size: None,
-                        });
-                        region_server_moved
-                            .handle_request(region_id, request)
-                            .await?;
-                        Ok(())
-                    }),
-                )
-                .await;
-            if register_result.is_busy() {
-                warn!("Another flush task is running for the region: {region_id}");
-            }
-            let mut watcher = register_result.into_watcher();
-            let result = self.flush_tasks.wait_until_finish(&mut watcher).await;
-            match result {
-                Ok(()) => Some(InstructionReply::FlushRegion(SimpleReply {
-                    result: true,
-                    error: None,
-                })),
-                Err(err) => Some(InstructionReply::FlushRegion(SimpleReply {
-                    result: false,
-                    error: Some(format!("{err:?}")),
-                })),
-            }
-        })
-    }
 }

 #[cfg(test)]
@@ -138,7 +84,7 @@ mod tests {

        let reply = handler_context
            .clone()
-            .handle_flush_regions_instruction(FlushRegions {
+            .handle_flush_region_instruction(FlushRegions {
                region_ids: region_ids.clone(),
            })
            .await;
@@ -148,7 +94,7 @@ mod tests {
        flushed_region_ids.write().unwrap().clear();
        let not_found_region_ids = (0..2).map(|i| RegionId::new(2048, i)).collect::<Vec<_>>();
        let reply = handler_context
-            .handle_flush_regions_instruction(FlushRegions {
+            .handle_flush_region_instruction(FlushRegions {
                region_ids: not_found_region_ids.clone(),
            })
            .await;
--- a/src/datanode/src/heartbeat/task_tracker.rs
+++ b/src/datanode/src/heartbeat/task_tracker.rs
@@ -144,11 +144,6 @@ impl<T: Send + Sync + Clone + 'static> TaskTracker<T> {
        }
    }

-    /// Waits for a [RegisterResult] and returns a [WaitResult].
-    pub(crate) async fn wait_until_finish(&self, watcher: &mut TaskWatcher<T>) -> Result<T> {
-        wait(watcher).await
-    }
-
    /// Tries to register a new async task, returns [RegisterResult::Busy] if previous task is running.
    pub(crate) async fn try_register(
        &self,
--- a/src/datanode/src/service.rs
+++ b/src/datanode/src/service.rs
@@ -62,7 +62,7 @@ impl<'a> DatanodeServiceBuilder<'a> {
        }
    }

-    pub fn build(mut self) -> Result<ServerHandlers> {
+    pub async fn build(mut self) -> Result<ServerHandlers> {
        let handlers = ServerHandlers::default();

        if let Some(grpc_server) = self.grpc_server.take() {
@@ -70,7 +70,7 @@ impl<'a> DatanodeServiceBuilder<'a> {
                addr: &self.opts.grpc.bind_addr,
            })?;
            let handler: ServerHandler = (Box::new(grpc_server), addr);
-            handlers.insert(handler);
+            handlers.insert(handler).await;
        }

        if self.enable_http_service {
@@ -82,7 +82,7 @@ impl<'a> DatanodeServiceBuilder<'a> {
                addr: &self.opts.http.addr,
            })?;
            let handler: ServerHandler = (Box::new(http_server), addr);
-            handlers.insert(handler);
+            handlers.insert(handler).await;
        }

        Ok(handlers)
--- a/src/datatypes/src/schema/constraint.rs
+++ b/src/datatypes/src/schema/constraint.rs
@@ -20,7 +20,6 @@ use snafu::{ensure, ResultExt};

 use crate::data_type::{ConcreteDataType, DataType};
 use crate::error::{self, Result};
-use crate::types::cast;
 use crate::value::Value;
 use crate::vectors::operations::VectorOp;
 use crate::vectors::{TimestampMillisecondVector, VectorRef};
@@ -179,18 +178,6 @@ impl ColumnDefaultConstraint {
        }
    }

-    /// Cast default value to given type
-    pub fn cast_to_datatype(&self, data_type: &ConcreteDataType) -> Result<Self> {
-        match self {
-            ColumnDefaultConstraint::Value(v) => Ok(Self::Value(cast(v.clone(), data_type)?)),
-            ColumnDefaultConstraint::Function(expr) => match &expr[..] {
-                // no need to cast, since function always require a data_type when need to create default value
-                CURRENT_TIMESTAMP | CURRENT_TIMESTAMP_FN | NOW_FN => Ok(self.clone()),
-                _ => error::UnsupportedDefaultExprSnafu { expr }.fail(),
-            },
-        }
-    }
-
    /// Only create default vector if it's impure, i.e., it's a function.
    ///
    /// This helps to delay creating constant default values to mito engine while also keeps impure default have consistent values
@@ -266,10 +253,9 @@ fn create_current_timestamp_vector(
    data_type: &ConcreteDataType,
    num_rows: usize,
 ) -> Result<VectorRef> {
-    let current_timestamp_vector = TimestampMillisecondVector::from_values(std::iter::repeat_n(
-        util::current_time_millis(),
-        num_rows,
-    ));
+    let current_timestamp_vector = TimestampMillisecondVector::from_values(
+        std::iter::repeat(util::current_time_millis()).take(num_rows),
+    );
    if data_type.is_timestamp() {
        current_timestamp_vector.cast(data_type)
    } else {
--- a/src/datatypes/src/vectors/constant.rs
+++ b/src/datatypes/src/vectors/constant.rs
@@ -198,7 +198,8 @@ impl fmt::Debug for ConstantVector {

 impl Serializable for ConstantVector {
    fn serialize_to_json(&self) -> Result<Vec<serde_json::Value>> {
-        std::iter::repeat_n(self.get(0), self.len())
+        std::iter::repeat(self.get(0))
+            .take(self.len())
            .map(serde_json::Value::try_from)
            .collect::<serde_json::Result<_>>()
            .context(SerializeSnafu)
--- a/src/datatypes/src/vectors/decimal.rs
+++ b/src/datatypes/src/vectors/decimal.rs
@@ -412,7 +412,7 @@ pub(crate) fn replicate_decimal128(
                    // Safety: std::iter::Repeat and std::iter::Take implement TrustedLen.
                    builder
                        .mutable_array
-                        .append_trusted_len_iter(std::iter::repeat_n(data, repeat_times));
+                        .append_trusted_len_iter(std::iter::repeat(data).take(repeat_times));
                }
            }
            None => {
--- a/src/datatypes/src/vectors/null.rs
+++ b/src/datatypes/src/vectors/null.rs
@@ -120,7 +120,9 @@ impl fmt::Debug for NullVector {

 impl Serializable for NullVector {
    fn serialize_to_json(&self) -> Result<Vec<serde_json::Value>> {
-        Ok(std::iter::repeat_n(serde_json::Value::Null, self.len()).collect())
+        Ok(std::iter::repeat(serde_json::Value::Null)
+            .take(self.len())
+            .collect())
    }
 }

--- a/src/datatypes/src/vectors/primitive.rs
+++ b/src/datatypes/src/vectors/primitive.rs
@@ -388,7 +388,7 @@ pub(crate) fn replicate_primitive<T: LogicalPrimitiveType>(
                    // Safety: std::iter::Repeat and std::iter::Take implement TrustedLen.
                    builder
                        .mutable_array
-                        .append_trusted_len_iter(std::iter::repeat_n(data, repeat_times));
+                        .append_trusted_len_iter(std::iter::repeat(data).take(repeat_times));
                }
            }
            None => {
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
discord9	ed676d97c7	refactor: rename FlowWorkerManager to FlowStreamingEngine	2025-04-23 11:36:30 +08:00
discord9	14b2badded	chore: better error variant	2025-04-22 19:52:19 +08:00
discord9	3626a50395	chore: use better error variant	2025-04-22 17:30:24 +08:00
discord9	0d0dad4ba2	chore: update docs	2025-04-22 17:09:42 +08:00
discord9	ae00e28b2a	refactor: per review partially	2025-04-22 17:09:42 +08:00
discord9	92d2fafb33	chore: per review rename args	2025-04-22 17:09:42 +08:00
discord9	30b3600597	chore: per review	2025-04-22 17:09:42 +08:00
discord9	87f1a8c622	refactor: per review	2025-04-22 17:09:42 +08:00
discord9	8e815fc385	chore: add comments per review	2025-04-22 17:09:42 +08:00
discord9	ca46bd04ee	chore: better logging	2025-04-22 17:09:42 +08:00
discord9	d32ade7399	fix: query without time window also clean dirty time window	2025-04-22 17:09:42 +08:00
discord9	b4aa0c8b8b	refactor: per review	2025-04-22 17:09:42 +08:00
discord9	e647559d27	refactor: AddAutoColumnRewriter check for Projection	2025-04-22 17:09:42 +08:00
discord9	d2c4767d41	docs: explain `nodeid` use in check task	2025-04-22 17:09:42 +08:00
discord9	82cee11eea	test: add align time window test	2025-04-22 17:09:42 +08:00
discord9	6d0470c3fb	feat: flush_flow flush all ranges now	2025-04-22 17:09:42 +08:00
discord9	47a267e29c	fix: add locks for create/drop flow&docs: update docs	2025-04-22 17:09:42 +08:00
discord9	fa13d06fc6	chore: update proto to main branch	2025-04-22 17:09:42 +08:00
discord9	26d9517c3e	chore: update proto	2025-04-22 17:09:42 +08:00
discord9	a7da9af5de	feat: use flow batching engine broken: try using logical plan fix: use dummy catalog for logical plan fix: insert plan exec&sqlness grpc addr feat: use frontend instance in flownode in standalone feat: flow type in metasrv&fix: flush flow out of sync& column name alias tests: sqlness update tests: sqlness flow rebuild udpate chore: per review refactor: keep chnl mgr refactor: use catalog mgr for get table tests: use valid sql fix: add more check refactor: put flow type determine to frontend	2025-04-22 17:09:42 +08:00