fix: choose frontend randomly

* feat: set read-preference for grpc client

* todo

* address PR comments

* fix ci

.coderabbit.yaml

@@ -1,15 +0,0 @@
-# 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

.github/actions/setup-greptimedb-cluster/with-remote-wal.yaml

@@ -7,7 +7,8 @@ meta:
     provider = "kafka"
     broker_endpoints = ["kafka.kafka-cluster.svc.cluster.local:9092"]
     num_topics = 3
-    auto_prune_topic_records = true
+    auto_prune_interval = "30s"
+    trigger_flush_threshold = 100
 
     [datanode]
     [datanode.client]
@@ -22,6 +23,7 @@ datanode:
     provider = "kafka"
     broker_endpoints = ["kafka.kafka-cluster.svc.cluster.local:9092"]
     linger = "2ms"
+    overwrite_entry_start_id = true
 frontend:
   configData: |-
     [runtime]

.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"
+      echo "GITHUB_EVENT_NAME is empty" >&2
       exit 1
   fi
 
   if [ -z "$NEXT_RELEASE_VERSION" ]; then
-      echo "NEXT_RELEASE_VERSION is empty"
-      exit 1
+      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)
   fi
 
   if [ -z "$NIGHTLY_RELEASE_PREFIX" ]; then
-      echo "NIGHTLY_RELEASE_PREFIX is empty"
+      echo "NIGHTLY_RELEASE_PREFIX is empty" >&2
       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"
+      echo "COMMIT_SHA is empty in dev build" >&2
       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"
+      echo "GITHUB_REF_NAME is empty in push event" >&2
       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"
+    echo "Unsupported GITHUB_EVENT_NAME: $GITHUB_EVENT_NAME" >&2
     exit 1
   fi
 }

.github/scripts/update-dev-builder-version.sh

@@ -0,0 +1,37 @@
+#!/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

.github/workflows/develop.yml

@@ -22,6 +22,7 @@ concurrency:
 
 jobs:
   check-typos-and-docs:
+    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
     name: Check typos and docs
     runs-on: ubuntu-latest
     steps:
@@ -36,6 +37,7 @@ 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:
@@ -45,6 +47,7 @@ jobs:
       - uses: korandoru/hawkeye@v5
 
   check:
+    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
     name: Check
     runs-on: ${{ matrix.os }}
     strategy:
@@ -71,6 +74,7 @@ jobs:
         run: cargo check --locked --workspace --all-targets
 
   toml:
+    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
     name: Toml Check
     runs-on: ubuntu-latest
     timeout-minutes: 60
@@ -85,6 +89,7 @@ jobs:
         run: taplo format --check
 
   build:
+    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
     name: Build GreptimeDB binaries
     runs-on: ${{ matrix.os }}
     strategy:
@@ -127,6 +132,7 @@ jobs:
           version: current
 
   fuzztest:
+    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
     name: Fuzz Test
     needs: build
     runs-on: ubuntu-latest
@@ -183,6 +189,7 @@ jobs:
           max-total-time: 120
 
   unstable-fuzztest:
+    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
     name: Unstable Fuzz Test
     needs: build-greptime-ci
     runs-on: ubuntu-latest
@@ -244,6 +251,7 @@ jobs:
           retention-days: 3
 
   build-greptime-ci:
+    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
     name: Build GreptimeDB binary (profile-CI)
     runs-on: ${{ matrix.os }}
     strategy:
@@ -285,6 +293,7 @@ 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
@@ -416,6 +425,7 @@ 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
@@ -563,6 +573,7 @@ jobs:
           docker system prune -f
 
   sqlness:
+    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
     name: Sqlness Test (${{ matrix.mode.name }})
     needs: build
     runs-on: ${{ matrix.os }}
@@ -609,6 +620,7 @@ jobs:
           retention-days: 3
 
   fmt:
+    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
     name: Rustfmt
     runs-on: ubuntu-latest
     timeout-minutes: 60
@@ -626,6 +638,7 @@ jobs:
         run: make fmt-check
 
   clippy:
+    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
     name: Clippy
     runs-on: ubuntu-latest
     timeout-minutes: 60
@@ -651,6 +664,7 @@ jobs:
         run: make clippy
 
   conflict-check:
+    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
     name: Check for conflict
     runs-on: ubuntu-latest
     steps:
@@ -661,7 +675,7 @@ jobs:
         uses: olivernybroe/action-conflict-finder@v4.0
 
   test:
-    if: github.event_name != 'merge_group'
+    if: ${{ github.repository == 'GreptimeTeam/greptimedb' && github.event_name != 'merge_group' }}
     runs-on: ubuntu-22.04-arm
     timeout-minutes: 60
     needs:  [conflict-check, clippy, fmt]
@@ -713,7 +727,7 @@ jobs:
           UNITTEST_LOG_DIR: "__unittest_logs"
 
   coverage:
-    if: github.event_name == 'merge_group'
+    if: ${{ github.repository == 'GreptimeTeam/greptimedb' && github.event_name == 'merge_group' }}
     runs-on: ubuntu-22.04-8-cores
     timeout-minutes: 60
     steps:
@@ -773,6 +787,7 @@ jobs:
           verbose: true
 
   # compat:
+  #   if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
   #   name: Compatibility Test
   #   needs: build
   #   runs-on: ubuntu-22.04

.github/workflows/grafana.yml

@@ -21,32 +21,6 @@ jobs:
         run: sudo apt-get install -y jq
 
       # Make the check.sh script executable
-      - name: Make check.sh executable
-        run: chmod +x grafana/check.sh
-
-      # Run the check.sh script
-      - name: Run check.sh
-        run: ./grafana/check.sh
-
-      # Only run summary.sh for pull_request events (not for merge queues or final pushes)
-      - name: Check if this is a pull request
-        id: check-pr
+      - name: Check grafana dashboards
         run: |
-          if [[ "${{ github.event_name }}" == "pull_request" ]]; then
-            echo "is_pull_request=true" >> $GITHUB_OUTPUT
-          else
-            echo "is_pull_request=false" >> $GITHUB_OUTPUT
-          fi
-
-      # Make the summary.sh script executable
-      - name: Make summary.sh executable
-        if: steps.check-pr.outputs.is_pull_request == 'true'
-        run: chmod +x grafana/summary.sh
-
-      # Run the summary.sh script and add its output to the GitHub Job Summary
-      - name: Run summary.sh and add to Job Summary
-        if: steps.check-pr.outputs.is_pull_request == 'true'
-        run: |
-          SUMMARY=$(./grafana/summary.sh)
-          echo "### Summary of Grafana Panels" >> $GITHUB_STEP_SUMMARY
-          echo "$SUMMARY" >> $GITHUB_STEP_SUMMARY
+          make check-dashboards

.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: 60
+    timeout-minutes: 45
     steps:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
           persist-credentials: false
-      - uses: cachix/install-nix-action@v27
+      - uses: cachix/install-nix-action@v31
         with:
           nix_path: nixpkgs=channel:nixos-24.11
-      - run: nix develop --command cargo build
+      - run: nix develop --command cargo build --bin greptime
 
   check-status:
     name: Check status

.github/workflows/release-dev-builder-images.yaml

@@ -24,11 +24,19 @@ 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
-    if: ${{ inputs.release_dev_builder_ubuntu_image || inputs.release_dev_builder_centos_image || inputs.release_dev_builder_android_image }} # Only manually trigger this job.
+    # 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 }}
     runs-on: ubuntu-latest
     outputs:
       version: ${{ steps.set-version.outputs.version }}
@@ -57,9 +65,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 }}
-          build-dev-builder-centos: ${{ inputs.release_dev_builder_centos_image }}
-          build-dev-builder-android: ${{ inputs.release_dev_builder_android_image }}
+          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' }}
 
   release-dev-builder-images-ecr:
     name: Release dev builder images to AWS ECR
@@ -85,7 +93,7 @@ jobs:
 
       - name: Push dev-builder-ubuntu image
         shell: bash
-        if: ${{ inputs.release_dev_builder_ubuntu_image }}
+        if: ${{ inputs.release_dev_builder_ubuntu_image || github.event_name == 'push' }}
         env:
           IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
           IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -106,7 +114,7 @@ jobs:
 
       - name: Push dev-builder-centos image
         shell: bash
-        if: ${{ inputs.release_dev_builder_centos_image }}
+        if: ${{ inputs.release_dev_builder_centos_image || github.event_name == 'push' }}
         env:
           IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
           IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -127,7 +135,7 @@ jobs:
 
       - name: Push dev-builder-android image
         shell: bash
-        if: ${{ inputs.release_dev_builder_android_image }}
+        if: ${{ inputs.release_dev_builder_android_image || github.event_name == 'push' }}
         env:
           IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
           IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -162,7 +170,7 @@ jobs:
 
       - name: Push dev-builder-ubuntu image
         shell: bash
-        if: ${{ inputs.release_dev_builder_ubuntu_image }}
+        if: ${{ inputs.release_dev_builder_ubuntu_image || github.event_name == 'push' }}
         env:
           IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
           IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -176,7 +184,7 @@ jobs:
 
       - name: Push dev-builder-centos image
         shell: bash
-        if: ${{ inputs.release_dev_builder_centos_image }}
+        if: ${{ inputs.release_dev_builder_centos_image || github.event_name == 'push' }}
         env:
           IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
           IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -190,7 +198,7 @@ jobs:
 
       - name: Push dev-builder-android image
         shell: bash
-        if: ${{ inputs.release_dev_builder_android_image }}
+        if: ${{ inputs.release_dev_builder_android_image || github.event_name == 'push' }}
         env:
           IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
           IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -201,3 +209,24 @@ 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 }}

.github/workflows/release.yml

@@ -90,8 +90,6 @@ 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:
@@ -135,7 +133,6 @@ 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
@@ -467,6 +464,29 @@ 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

.gitignore

@@ -28,6 +28,7 @@ debug/
 # Logs
 **/__unittest_logs
 logs/
+!grafana/dashboards/logs/
 
 # cpython's generated python byte code
 **/__pycache__/

Cargo.toml

@@ -68,15 +68,16 @@ members = [
 resolver = "2"
 
 [workspace.package]
-version = "0.14.0"
+version = "0.15.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)'] }
 
@@ -112,15 +113,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 = "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" }
+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" }
 deadpool = "0.12"
 deadpool-postgres = "0.14"
 derive_builder = "0.20"
@@ -129,7 +130,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 = "b6d9cffd43c4e6358805a798f17e03e232994b82" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "17a3550751c8b1e02ec16be40101d5f24dc255c3" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"
@@ -162,7 +163,7 @@ paste = "1.0"
 pin-project = "1.0"
 prometheus = { version = "0.13.3", features = ["process"] }
 promql-parser = { version = "0.5.1", features = ["ser"] }
-prost = "0.13"
+prost = { version = "0.13", features = ["no-recursion-limit"] }
 raft-engine = { version = "0.4.1", default-features = false }
 rand = "0.9"
 ratelimit = "0.10"
@@ -191,7 +192,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 = "e98e6b322426a9d397a71efef17075966223c089", features = [
+sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "0cf6c04490d59435ee965edd2078e8855bd8471e", features = [
     "visitor",
     "serde",
 ] } # branch = "v0.54.x"
@@ -269,6 +270,9 @@ metric-engine = { path = "src/metric-engine" }
 mito2 = { path = "src/mito2" }
 object-store = { path = "src/object-store" }
 operator = { path = "src/operator" }
+otel-arrow-rust = { git = "https://github.com/open-telemetry/otel-arrow", rev = "5d551412d2a12e689cde4d84c14ef29e36784e51", features = [
+    "server",
+] }
 partition = { path = "src/partition" }
 pipeline = { path = "src/pipeline" }
 plugins = { path = "src/plugins" }

Makefile

@@ -8,7 +8,7 @@ CARGO_BUILD_OPTS := --locked
 IMAGE_REGISTRY ?= docker.io
 IMAGE_NAMESPACE ?= greptime
 IMAGE_TAG ?= latest
-DEV_BUILDER_IMAGE_TAG ?= 2024-12-25-a71b93dd-20250305072908
+DEV_BUILDER_IMAGE_TAG ?= 2025-04-15-1a517ec8-20250428023155
 BUILDX_MULTI_PLATFORM_BUILD ?= false
 BUILDX_BUILDER_NAME ?= gtbuilder
 BASE_IMAGE ?= ubuntu
@@ -222,6 +222,16 @@ start-cluster: ## Start the greptimedb cluster with etcd by using docker compose
 stop-cluster: ## Stop the greptimedb cluster that created by docker compose.
 	docker compose -f ./docker/docker-compose/cluster-with-etcd.yaml stop
 
+##@ Grafana
+
+.PHONY: check-dashboards
+check-dashboards: ## Check the Grafana dashboards.
+	@./grafana/scripts/check.sh
+
+.PHONY: dashboards
+dashboards: ## Generate the Grafana dashboards for standalone mode and intermediate dashboards.
+	@./grafana/scripts/gen-dashboards.sh
+
 ##@ Docs
 config-docs: ## Generate configuration documentation from toml files.
 	docker run --rm \

README.md

@@ -6,7 +6,9 @@
   </picture>
 </p>
 
-<h2 align="center">Unified & Cost-Effective Observability  Database for Metrics, Logs, and Events</h2>
+<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">
@@ -49,74 +51,77 @@
 </div>
 
 - [Introduction](#introduction)
-- [**Features: Why GreptimeDB**](#why-greptimedb)
-- [Architecture](https://docs.greptime.com/contributor-guide/overview/#architecture)
-- [Try it for free](#try-greptimedb)
+- [⭐ Key Features](#features)
+- [Quick Comparison](#quick-comparison)
+- [Architecture](#architecture)
+- [Try GreptimeDB](#try-greptimedb)
 - [Getting Started](#getting-started)
-- [Project Status](#project-status)
-- [Join the community](#community)
-  - [Contributing](#contributing)
+- [Build From Source](#build-from-source)
 - [Tools & Extensions](#tools--extensions)
+- [Project Status](#project-status)
+- [Community](#community)
 - [License](#license)
+- [Commercial Support](#commercial-support)
+- [Contributing](#contributing)
 - [Acknowledgement](#acknowledgement)
 
 ## Introduction
 
-**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.
+**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.
 
-## News
+## Features
 
-**[GreptimeDB tops JSONBench's billion-record cold run test!](https://greptime.com/blogs/2025-03-18-jsonbench-greptimedb-performance)**
+|   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. |
 
-## Why GreptimeDB
+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).
 
-Our core developers have been building observability data platforms for years. Based on our best practices, GreptimeDB was born to give you:
+## Quick Comparison
 
-* **Unified Processing of Observability Data**
+| 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          |
 
-  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.
+**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)
 
-* **High Performance and Cost-effective**
+Read [more benchmark reports](https://docs.greptime.com/user-guide/concepts/features-that-you-concern#how-is-greptimedbs-performance-compared-to-other-solutions).
 
-   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.
+## Architecture
 
-* **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).
+* 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">
 
 ## Try GreptimeDB
 
 ### 1. [Live Demo](https://greptime.com/playground)
 
-Try out the features of GreptimeDB right from your browser.
+Experience GreptimeDB directly in your browser.
 
 ### 2. [GreptimeCloud](https://console.greptime.cloud/)
 
 Start instantly with a free cluster.
 
-### 3. Docker Image
-
-To install GreptimeDB locally, the recommended way is via Docker:
+### 3. Docker (Local Quickstart)
 
 ```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 \
@@ -124,114 +129,90 @@ 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)
 
-Access the dashboard via `http://localhost:4000/dashboard`.
-
-Read more about [Installation](https://docs.greptime.com/getting-started/installation/overview) on docs.
+**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.
 
 ## Getting Started
 
-* [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)
+- [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)
 
-## Build
-
-Check the prerequisite:
+## Build From Source
 
+**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 GreptimeDB binary:
-
-```shell
+**Build and Run:**
+```bash
 make
-```
-
-Run a standalone server:
-
-```shell
 cargo run -- standalone start
 ```
 
 ## Tools & Extensions
 
-### 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.
+- **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)
 
 ## Project Status
 
-GreptimeDB is currently in Beta. We are targeting GA (General Availability) with v1.0 release by Early 2025.
+> **Status:** Beta.  
+> **GA (v1.0):** Targeted for mid 2025.
 
-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
+- 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
 
 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
 
-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:
+We invite you to engage and contribute!
 
-- GreptimeDB Community on [Slack](https://greptime.com/slack)
-- GreptimeDB [GitHub Discussions forum](https://github.com/GreptimeTeam/greptimedb/discussions)
-- Greptime official [website](https://greptime.com)
-
-In addition, you may:
-
-- 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
-
-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.
+- [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)
 
 ## 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.
+GreptimeDB is licensed under the [Apache License 2.0](https://apache.org/licenses/LICENSE-2.0.txt).
+
+## Commercial Support
+
+Running GreptimeDB in your organization?  
+We offer enterprise add-ons, services, training, and consulting.  
+[Contact us](https://greptime.com/contactus) for details.
 
 ## Contributing
 
-Please refer to [contribution guidelines](CONTRIBUTING.md) and [internal concepts docs](https://docs.greptime.com/contributor-guide/overview.html) for more information.
+- 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.
 
 ## Acknowledgement
 
-Special thanks to all the contributors who have propelled GreptimeDB forward. For a complete list of contributors, please refer to [AUTHOR.md](AUTHOR.md).
+Special thanks to all contributors! See [AUTHORS.md](https://github.com/GreptimeTeam/greptimedb/blob/main/AUTHOR.md).
 
-- 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"/>
+- 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)

config/config.md

@@ -319,6 +319,7 @@
 | `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. |

config/metasrv.example.toml

@@ -50,6 +50,10 @@ 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"
 

cyborg/bin/bump-website-version.ts

@@ -0,0 +1,57 @@
+/*
+ * 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);
+}

flake.lock

@@ -8,11 +8,11 @@
         "rust-analyzer-src": "rust-analyzer-src"
       },
       "locked": {
-        "lastModified": 1737613896,
-        "narHash": "sha256-ldqXIglq74C7yKMFUzrS9xMT/EVs26vZpOD68Sh7OcU=",
+        "lastModified": 1745735608,
+        "narHash": "sha256-L0jzm815XBFfF2wCFmR+M1CF+beIEFj6SxlqVKF59Ec=",
         "owner": "nix-community",
         "repo": "fenix",
-        "rev": "303a062fdd8e89f233db05868468975d17855d80",
+        "rev": "c39a78eba6ed2a022cc3218db90d485077101496",
         "type": "github"
       },
       "original": {
@@ -41,11 +41,11 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1737569578,
-        "narHash": "sha256-6qY0pk2QmUtBT9Mywdvif0i/CLVgpCjMUn6g9vB+f3M=",
+        "lastModified": 1745487689,
+        "narHash": "sha256-FQoi3R0NjQeBAsEOo49b5tbDPcJSMWc3QhhaIi9eddw=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "47addd76727f42d351590c905d9d1905ca895b82",
+        "rev": "5630cf13cceac06cefe9fc607e8dfa8fb342dde3",
         "type": "github"
       },
       "original": {
@@ -65,11 +65,11 @@
     "rust-analyzer-src": {
       "flake": false,
       "locked": {
-        "lastModified": 1737581772,
-        "narHash": "sha256-t1P2Pe3FAX9TlJsCZbmJ3wn+C4qr6aSMypAOu8WNsN0=",
+        "lastModified": 1745694049,
+        "narHash": "sha256-fxvRYH/tS7hGQeg9zCVh5RBcSWT+JGJet7RA8Ss+rC0=",
         "owner": "rust-lang",
         "repo": "rust-analyzer",
-        "rev": "582af7ee9c8d84f5d534272fc7de9f292bd849be",
+        "rev": "d8887c0758bbd2d5f752d5bd405d4491e90e7ed6",
         "type": "github"
       },
       "original": {

flake.nix

@@ -21,7 +21,7 @@
         lib = nixpkgs.lib;
         rustToolchain = fenix.packages.${system}.fromToolchainName {
           name = (lib.importTOML ./rust-toolchain.toml).toolchain.channel;
-          sha256 = "sha256-f/CVA1EC61EWbh0SjaRNhLL0Ypx2ObupbzigZp8NmL4=";
+          sha256 = "sha256-arzEYlWLGGYeOhECHpBxQd2joZ4rPKV3qLNnZ+eql6A=";
         };
       in
       {

grafana/README.md

@@ -1,61 +1,122 @@
-Grafana dashboard for GreptimeDB
---------------------------------
+# Grafana dashboards for GreptimeDB
 
-GreptimeDB's official Grafana dashboard.
+## Overview
 
-Status notify: we are still working on this config. It's expected to change frequently in the recent days. Please feel free to submit your feedback and/or contribution to this dashboard 🤗
+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**.
 
-If you use Helm [chart](https://github.com/GreptimeTeam/helm-charts) to deploy GreptimeDB cluster, you can enable self-monitoring by setting the following values in your Helm chart:
+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.
+
+- **Metrics Dashboards**
+
+  - `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.
+
+- **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.
+
+## Data Sources
+
+The following data sources are used to fetch metrics and logs:
+
+- **`${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.
+
+## Instance Filters
+
+To deploy the dashboards for multiple scenarios (K8s, bare metal, etc.), we prefer to use the `instance` label when filtering instances.
+
+Additionally, we recommend including the `pod` label in the legend to make it easier to identify each instance, even though this field will be empty in bare metal scenarios.
+
+For example, the following query is recommended:
+
+```promql
+sum(process_resident_memory_bytes{instance=~"$datanode"}) by (instance, pod)
+```
+
+And the legend will be like: `[{{instance}}]-[{{ pod }}]`.
+
+## Deployment
+
+### (Recommended) 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;
 
-The standalone GreptimeDB instance will collect metrics from your cluster and the dashboard will be available in the Grafana UI. For detailed deployment instructions, please refer to our [Kubernetes deployment guide](https://docs.greptime.com/nightly/user-guide/deployments/deploy-on-kubernetes/getting-started).
+The standalone GreptimeDB instance will collect metrics from your cluster, and the dashboard will be available in the Grafana UI. For detailed deployment instructions, please refer to our [Kubernetes deployment guide](https://docs.greptime.com/nightly/user-guide/deployments/deploy-on-kubernetes/getting-started).
 
-# How to use
+### Self-host Prometheus and import dashboards manually
 
-## `greptimedb.json`
+1. **Configure Prometheus to scrape the cluster**
 
-Open Grafana Dashboard page, choose `New` -> `Import`. And upload `greptimedb.json` file.
+   The following is an example configuration(**Please modify it according to your actual situation**):
 
-## `greptimedb-cluster.json`
+    ```yml
+    # example config
+    # only to indicate how to assign labels to each target
+    # modify yours accordingly
+    scrape_configs:
+      - job_name: metasrv
+        static_configs:
+        - targets: ['<metasrv-ip>:<port>']
 
-This cluster dashboard provides a comprehensive view of incoming requests, response statuses, and internal activities such as flush and compaction, with a layered structure from frontend to datanode. Designed with a focus on alert functionality, its primary aim is to highlight any anomalies in metrics, allowing users to quickly pinpoint the cause of errors.
+      - job_name: datanode
+        static_configs:
+        - targets: ['<datanode0-ip>:<port>', '<datanode1-ip>:<port>', '<datanode2-ip>:<port>']
 
-We use Prometheus to scrape off metrics from nodes in GreptimeDB cluster, Grafana to visualize the diagram. Any compatible stack should work too.
+      - job_name: frontend
+        static_configs:
+        - targets: ['<frontend-ip>:<port>']
+    ```
 
-__Note__: This dashboard is still in an early stage of development. Any issue or advice on improvement is welcomed.
+2. **Configure the data sources in Grafana**
 
-### Configuration
+   You need to add two data sources in Grafana:
 
-Please ensure the following configuration before importing the dashboard into Grafana.
+   - Prometheus: It is the Prometheus instance that scrapes the GreptimeDB metrics.
+   - Information Schema: It is the MySQL port of the current monitored instance. The dashboard will use this datasource to show the information schema of the current instance.
 
-__1. Prometheus scrape config__
+3. **Import the dashboards based on your deployment scenario**
 
-Configure Prometheus to scrape the cluster.
-
-```yml
-# example config
-# only to indicate how to assign labels to each target
-# modify yours accordingly
-scrape_configs:
-  - job_name: metasrv
-    static_configs:
-    - targets: ['<metasrv-ip>:<port>']
-
-  - job_name: datanode
-    static_configs:
-    - targets: ['<datanode0-ip>:<port>', '<datanode1-ip>:<port>', '<datanode2-ip>:<port>']
-
-  - job_name: frontend
-    static_configs:
-    - targets: ['<frontend-ip>:<port>']
-```
-
-__2. Grafana config__
-
-Create a Prometheus data source in Grafana before using this dashboard. We use `datasource` as a variable in Grafana dashboard so that multiple environments are supported.
-
-### Usage
-
-Use `datasource` or `instance` on the upper-left corner to filter data from certain node.
+   - **Cluster**: Import the `dashboards/metrics/cluster/dashboard.json` dashboard.
+   - **Standalone**: Import the `dashboards/metrics/standalone/dashboard.json` dashboard.

grafana/check.sh

@@ -1,19 +0,0 @@
-#!/usr/bin/env bash
-
-BASEDIR=$(dirname "$0")
-
-# Use jq to check for panels with empty or missing descriptions
-invalid_panels=$(cat $BASEDIR/greptimedb-cluster.json | jq -r '
-  .panels[]
-  | select((.type == "stats" or .type == "timeseries") and (.description == "" or .description == null))
-')
-
-# Check if any invalid panels were found
-if [[ -n "$invalid_panels" ]]; then
-  echo "Error: The following panels have empty or missing descriptions:"
-  echo "$invalid_panels"
-  exit 1
-else
-  echo "All panels with type 'stats' or 'timeseries' have valid descriptions."
-  exit 0
-fi

grafana/dashboards/logs/dashboard.json

@@ -0,0 +1,292 @@
+{
+  "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
+}

grafana/dashboards/metrics/cluster/dashboard.md

@@ -0,0 +1,97 @@
+# 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` | -- |
+# 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` |
+# 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` |
+# 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 }}]` |
+# 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` |
+# 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}}]` |
+# 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}}]` |
+# 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}}]` |
+# 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` |
+# 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}}]` |

grafana/dashboards/metrics/cluster/dashboard.yaml

@@ -0,0 +1,769 @@
+groups:
+    - title: Overview
+      panels:
+        - title: Uptime
+          type: stat
+          description: The start time of GreptimeDB.
+          unit: s
+          queries:
+            - expr: time() - process_start_time_seconds
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: __auto
+        - title: Version
+          type: stat
+          description: GreptimeDB version.
+          queries:
+            - expr: SELECT pkg_version FROM information_schema.build_info
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+        - title: Total Ingestion Rate
+          type: stat
+          description: Total ingestion rate.
+          unit: rowsps
+          queries:
+            - expr: sum(rate(greptime_table_operator_ingest_rows[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: __auto
+        - title: Total Storage Size
+          type: stat
+          description: Total number of data file size.
+          unit: decbytes
+          queries:
+            - expr: select SUM(disk_size) from information_schema.region_statistics;
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+        - title: Total Rows
+          type: stat
+          description: Total number of data rows in the cluster. Calculated by sum of rows from each region.
+          unit: sishort
+          queries:
+            - expr: select SUM(region_rows) from information_schema.region_statistics;
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+        - title: Deployment
+          type: stat
+          description: The deployment topology of GreptimeDB.
+          queries:
+            - expr: SELECT count(*) as datanode FROM information_schema.cluster_info WHERE peer_type = 'DATANODE';
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+            - expr: SELECT count(*) as frontend FROM information_schema.cluster_info WHERE peer_type = 'FRONTEND';
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+            - expr: SELECT count(*) as metasrv FROM information_schema.cluster_info WHERE peer_type = 'METASRV';
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+            - expr: SELECT count(*) as flownode FROM information_schema.cluster_info WHERE peer_type = 'FLOWNODE';
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+        - title: Database Resources
+          type: stat
+          description: The number of the key resources in GreptimeDB.
+          queries:
+            - expr: SELECT COUNT(*) as databases FROM information_schema.schemata WHERE schema_name NOT IN ('greptime_private', 'information_schema')
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+            - expr: SELECT COUNT(*) as tables FROM information_schema.tables WHERE table_schema != 'information_schema'
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+            - expr: SELECT COUNT(region_id) as regions FROM information_schema.region_peers
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+            - expr: SELECT COUNT(*) as flows FROM information_schema.flows
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+        - title: Data Size
+          type: stat
+          description: The data size of wal/index/manifest in the GreptimeDB.
+          unit: decbytes
+          queries:
+            - expr: SELECT SUM(memtable_size) * 0.42825 as WAL FROM information_schema.region_statistics;
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+            - expr: SELECT SUM(index_size) as index FROM information_schema.region_statistics;
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+            - expr: SELECT SUM(manifest_size) as manifest FROM information_schema.region_statistics;
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+    - title: Ingestion
+      panels:
+        - title: Total Ingestion Rate
+          type: timeseries
+          description: |
+            Total ingestion rate.
+
+            Here we listed 3 primary protocols:
+
+            - Prometheus remote write
+            - Greptime's gRPC API (when using our ingest SDK)
+            - Log ingestion http API
+          unit: rowsps
+          queries:
+            - expr: sum(rate(greptime_table_operator_ingest_rows{instance=~"$frontend"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: ingestion
+        - title: Ingestion Rate by Type
+          type: timeseries
+          description: |
+            Total ingestion rate.
+
+            Here we listed 3 primary protocols:
+
+            - Prometheus remote write
+            - Greptime's gRPC API (when using our ingest SDK)
+            - Log ingestion http API
+          unit: rowsps
+          queries:
+            - expr: sum(rate(greptime_servers_http_logs_ingestion_counter[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: http-logs
+            - expr: sum(rate(greptime_servers_prometheus_remote_write_samples[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: prometheus-remote-write
+    - title: Queries
+      panels:
+        - title: Total Query Rate
+          type: timeseries
+          description: |-
+            Total rate of query API calls by protocol. This metric is collected from frontends.
+
+            Here we listed 3 main protocols:
+            - MySQL
+            - Postgres
+            - Prometheus API
+
+            Note that there are some other minor query APIs like /sql are not included
+          unit: reqps
+          queries:
+            - expr: sum (rate(greptime_servers_mysql_query_elapsed_count{instance=~"$frontend"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: mysql
+            - expr: sum (rate(greptime_servers_postgres_query_elapsed_count{instance=~"$frontend"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: pg
+            - expr: sum (rate(greptime_servers_http_promql_elapsed_counte{instance=~"$frontend"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: promql
+    - title: Resources
+      panels:
+        - title: Datanode Memory per Instance
+          type: timeseries
+          description: Current memory usage by instance
+          unit: decbytes
+          queries:
+            - expr: sum(process_resident_memory_bytes{instance=~"$datanode"}) by (instance, pod)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{ pod }}]'
+        - title: Datanode CPU Usage per Instance
+          type: timeseries
+          description: Current cpu usage by instance
+          unit: none
+          queries:
+            - expr: sum(rate(process_cpu_seconds_total{instance=~"$datanode"}[$__rate_interval]) * 1000) by (instance, pod)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{ pod }}]'
+        - title: Frontend Memory per Instance
+          type: timeseries
+          description: Current memory usage by instance
+          unit: decbytes
+          queries:
+            - expr: sum(process_resident_memory_bytes{instance=~"$frontend"}) by (instance, pod)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{ pod }}]'
+        - title: Frontend CPU Usage per Instance
+          type: timeseries
+          description: Current cpu usage by instance
+          unit: none
+          queries:
+            - expr: sum(rate(process_cpu_seconds_total{instance=~"$frontend"}[$__rate_interval]) * 1000) by (instance, pod)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{ pod }}]-cpu'
+        - title: Metasrv Memory per Instance
+          type: timeseries
+          description: Current memory usage by instance
+          unit: decbytes
+          queries:
+            - expr: sum(process_resident_memory_bytes{instance=~"$metasrv"}) by (instance, pod)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{ pod }}]-resident'
+        - title: Metasrv CPU Usage per Instance
+          type: timeseries
+          description: Current cpu usage by instance
+          unit: none
+          queries:
+            - expr: sum(rate(process_cpu_seconds_total{instance=~"$metasrv"}[$__rate_interval]) * 1000) by (instance, pod)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{ pod }}]'
+        - title: Flownode Memory per Instance
+          type: timeseries
+          description: Current memory usage by instance
+          unit: decbytes
+          queries:
+            - expr: sum(process_resident_memory_bytes{instance=~"$flownode"}) by (instance, pod)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{ pod }}]'
+        - title: Flownode CPU Usage per Instance
+          type: timeseries
+          description: Current cpu usage by instance
+          unit: none
+          queries:
+            - expr: sum(rate(process_cpu_seconds_total{instance=~"$flownode"}[$__rate_interval]) * 1000) by (instance, pod)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{ pod }}]'
+    - title: Frontend Requests
+      panels:
+        - title: HTTP QPS per Instance
+          type: timeseries
+          description: HTTP QPS per Instance.
+          unit: reqps
+          queries:
+            - expr: sum by(instance, pod, path, method, code) (rate(greptime_servers_http_requests_elapsed_count{instance=~"$frontend",path!~"/health|/metrics"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]'
+        - title: HTTP P99 per Instance
+          type: timeseries
+          description: HTTP P99 per Instance.
+          unit: s
+          queries:
+            - expr: 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])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]-p99'
+        - title: gRPC QPS per Instance
+          type: timeseries
+          description: gRPC QPS per Instance.
+          unit: reqps
+          queries:
+            - expr: sum by(instance, pod, path, code) (rate(greptime_servers_grpc_requests_elapsed_count{instance=~"$frontend"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{path}}]-[{{code}}]'
+        - title: gRPC P99 per Instance
+          type: timeseries
+          description: gRPC P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, path, code) (rate(greptime_servers_grpc_requests_elapsed_bucket{instance=~"$frontend"}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]-p99'
+        - title: MySQL QPS per Instance
+          type: timeseries
+          description: MySQL QPS per Instance.
+          unit: reqps
+          queries:
+            - expr: sum by(pod, instance)(rate(greptime_servers_mysql_query_elapsed_count{instance=~"$frontend"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+        - title: MySQL P99 per Instance
+          type: timeseries
+          description: MySQL P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(pod, instance, le) (rate(greptime_servers_mysql_query_elapsed_bucket{instance=~"$frontend"}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{ pod }}]-p99'
+        - title: PostgreSQL QPS per Instance
+          type: timeseries
+          description: PostgreSQL QPS per Instance.
+          unit: reqps
+          queries:
+            - expr: sum by(pod, instance)(rate(greptime_servers_postgres_query_elapsed_count{instance=~"$frontend"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+        - title: PostgreSQL P99 per Instance
+          type: timeseries
+          description: PostgreSQL P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(pod,instance,le) (rate(greptime_servers_postgres_query_elapsed_bucket{instance=~"$frontend"}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-p99'
+    - title: Frontend to Datanode
+      panels:
+        - title: Ingest Rows per Instance
+          type: timeseries
+          description: Ingestion rate by row as in each frontend
+          unit: rowsps
+          queries:
+            - expr: sum by(instance, pod)(rate(greptime_table_operator_ingest_rows{instance=~"$frontend"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+        - title: Region Call QPS per Instance
+          type: timeseries
+          description: Region Call QPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod, request_type) (rate(greptime_grpc_region_request_count{instance=~"$frontend"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{request_type}}]'
+        - title: Region Call P99 per Instance
+          type: timeseries
+          description: Region Call P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, request_type) (rate(greptime_grpc_region_request_bucket{instance=~"$frontend"}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{request_type}}]'
+    - title: Mito Engine
+      panels:
+        - title: Request OPS per Instance
+          type: timeseries
+          description: Request QPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod, type) (rate(greptime_mito_handle_request_elapsed_count{instance=~"$datanode"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{type}}]'
+        - title: Request P99 per Instance
+          type: timeseries
+          description: Request P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, type) (rate(greptime_mito_handle_request_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{type}}]'
+        - title: Write Buffer per Instance
+          type: timeseries
+          description: Write Buffer per Instance.
+          unit: decbytes
+          queries:
+            - expr: greptime_mito_write_buffer_bytes{instance=~"$datanode"}
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+        - title: Write Rows per Instance
+          type: timeseries
+          description: Ingestion size by row counts.
+          unit: rowsps
+          queries:
+            - expr: sum by (instance, pod) (rate(greptime_mito_write_rows_total{instance=~"$datanode"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+        - title: Flush OPS per Instance
+          type: timeseries
+          description: Flush QPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod, reason) (rate(greptime_mito_flush_requests_total{instance=~"$datanode"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{reason}}]'
+        - title: Write Stall per Instance
+          type: timeseries
+          description: Write Stall per Instance.
+          queries:
+            - expr: sum by(instance, pod) (greptime_mito_write_stall_total{instance=~"$datanode"})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+        - title: Read Stage OPS per Instance
+          type: timeseries
+          description: Read Stage OPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod) (rate(greptime_mito_read_stage_elapsed_count{instance=~"$datanode", stage="total"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+        - title: Read Stage P99 per Instance
+          type: timeseries
+          description: Read Stage P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_read_stage_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]'
+        - title: Write Stage P99 per Instance
+          type: timeseries
+          description: Write Stage P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_write_stage_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]'
+        - title: Compaction OPS per Instance
+          type: timeseries
+          description: Compaction OPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod) (rate(greptime_mito_compaction_total_elapsed_count{instance=~"$datanode"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{pod}}]'
+        - title: Compaction P99 per Instance by Stage
+          type: timeseries
+          description: Compaction latency by stage
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-p99'
+        - title: Compaction P99 per Instance
+          type: timeseries
+          description: Compaction P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le,stage) (rate(greptime_mito_compaction_total_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-compaction'
+        - title: WAL write size
+          type: timeseries
+          description: Write-ahead logs write size as bytes. This chart includes stats of p95 and p99 size by instance, total WAL write rate.
+          unit: bytes
+          queries:
+            - expr: histogram_quantile(0.95, sum by(le,instance, pod) (rate(raft_engine_write_size_bucket[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-req-size-p95'
+            - expr: histogram_quantile(0.99, sum by(le,instance,pod) (rate(raft_engine_write_size_bucket[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-req-size-p99'
+            - expr: sum by (instance, pod)(rate(raft_engine_write_size_sum[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-throughput'
+        - title: Cached Bytes per Instance
+          type: timeseries
+          description: Cached Bytes per Instance.
+          unit: decbytes
+          queries:
+            - expr: greptime_mito_cache_bytes{instance=~"$datanode"}
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{type}}]'
+        - title: Inflight Compaction
+          type: timeseries
+          description: Ongoing compaction task count
+          unit: none
+          queries:
+            - expr: greptime_mito_inflight_compaction_count
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+        - title: WAL sync duration seconds
+          type: timeseries
+          description: Raft engine (local disk) log store sync latency, p99
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(le, type, node, instance, pod) (rate(raft_engine_sync_log_duration_seconds_bucket[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-p99'
+        - title: Log Store op duration seconds
+          type: timeseries
+          description: Write-ahead log operations latency at p99
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(le,logstore,optype,instance, pod) (rate(greptime_logstore_op_elapsed_bucket[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{logstore}}]-[{{optype}}]-p99'
+        - title: Inflight Flush
+          type: timeseries
+          description: Ongoing flush task count
+          unit: none
+          queries:
+            - expr: greptime_mito_inflight_flush_count
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+    - title: OpenDAL
+      panels:
+        - title: QPS per Instance
+          type: timeseries
+          description: QPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]'
+        - title: Read QPS per Instance
+          type: timeseries
+          description: Read QPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation="read"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]'
+        - title: Read P99 per Instance
+          type: timeseries
+          description: Read P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode",operation="read"}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-{{scheme}}'
+        - title: Write QPS per Instance
+          type: timeseries
+          description: Write QPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation="write"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-{{scheme}}'
+        - title: Write P99 per Instance
+          type: timeseries
+          description: Write P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode", operation="write"}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]'
+        - title: List QPS per Instance
+          type: timeseries
+          description: List QPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation="list"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]'
+        - title: List P99 per Instance
+          type: timeseries
+          description: List P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode", operation="list"}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]'
+        - title: Other Requests per Instance
+          type: timeseries
+          description: Other Requests per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode",operation!~"read|write|list|stat"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]'
+        - title: Other Request P99 per Instance
+          type: timeseries
+          description: Other Request P99 per Instance.
+          unit: s
+          queries:
+            - expr: 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])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]'
+        - title: Opendal traffic
+          type: timeseries
+          description: Total traffic as in bytes by instance and operation
+          unit: decbytes
+          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
+          type: state-timeline
+          description: Counter of region migration by source and destination
+          unit: none
+          queries:
+            - expr: greptime_meta_region_migration_stat{datanode_type="src"}
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: from-datanode-{{datanode_id}}
+            - expr: greptime_meta_region_migration_stat{datanode_type="desc"}
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: to-datanode-{{datanode_id}}
+        - title: Region migration error
+          type: timeseries
+          description: Counter of region migration error
+          unit: none
+          queries:
+            - expr: greptime_meta_region_migration_error
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: __auto
+        - title: Datanode load
+          type: timeseries
+          description: Gauge of load information of each datanode, collected via heartbeat between datanode and metasrv. This information is for metasrv to schedule workloads.
+          unit: none
+          queries:
+            - expr: greptime_datanode_load
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: __auto
+    - title: Flownode
+      panels:
+        - title: Flow Ingest / Output Rate
+          type: timeseries
+          description: Flow Ingest / Output Rate.
+          queries:
+            - expr: sum by(instance, pod, direction) (rate(greptime_flow_processed_rows[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{pod}}]-[{{instance}}]-[{{direction}}]'
+        - title: Flow Ingest Latency
+          type: timeseries
+          description: Flow Ingest Latency.
+          queries:
+            - expr: histogram_quantile(0.95, sum(rate(greptime_flow_insert_elapsed_bucket[$__rate_interval])) by (le, instance, pod))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-p95'
+            - expr: histogram_quantile(0.99, sum(rate(greptime_flow_insert_elapsed_bucket[$__rate_interval])) by (le, instance, pod))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-p99'
+        - title: Flow Operation Latency
+          type: timeseries
+          description: Flow Operation Latency.
+          queries:
+            - expr: histogram_quantile(0.95, sum(rate(greptime_flow_processing_time_bucket[$__rate_interval])) by (le,instance,pod,type))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{type}}]-p95'
+            - expr: histogram_quantile(0.99, sum(rate(greptime_flow_processing_time_bucket[$__rate_interval])) by (le,instance,pod,type))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{type}}]-p99'
+        - title: Flow Buffer Size per Instance
+          type: timeseries
+          description: Flow Buffer Size per Instance.
+          queries:
+            - expr: greptime_flow_input_buf_size
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}]'
+        - title: Flow Processing Error per Instance
+          type: timeseries
+          description: Flow Processing Error per Instance.
+          queries:
+            - expr: sum by(instance,pod,code) (rate(greptime_flow_errors[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{code}}]'

grafana/dashboards/metrics/standalone/dashboard.md

@@ -0,0 +1,97 @@
+# 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` | -- |
+# 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` |
+# 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` |
+# 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 }}]` |
+# 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` |
+# 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}}]` |
+# 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}}]` |
+# 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}}]` |
+# 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` |
+# 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}}]` |

grafana/dashboards/metrics/standalone/dashboard.yaml

@@ -0,0 +1,769 @@
+groups:
+    - title: Overview
+      panels:
+        - title: Uptime
+          type: stat
+          description: The start time of GreptimeDB.
+          unit: s
+          queries:
+            - expr: time() - process_start_time_seconds
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: __auto
+        - title: Version
+          type: stat
+          description: GreptimeDB version.
+          queries:
+            - expr: SELECT pkg_version FROM information_schema.build_info
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+        - title: Total Ingestion Rate
+          type: stat
+          description: Total ingestion rate.
+          unit: rowsps
+          queries:
+            - expr: sum(rate(greptime_table_operator_ingest_rows[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: __auto
+        - title: Total Storage Size
+          type: stat
+          description: Total number of data file size.
+          unit: decbytes
+          queries:
+            - expr: select SUM(disk_size) from information_schema.region_statistics;
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+        - title: Total Rows
+          type: stat
+          description: Total number of data rows in the cluster. Calculated by sum of rows from each region.
+          unit: sishort
+          queries:
+            - expr: select SUM(region_rows) from information_schema.region_statistics;
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+        - title: Deployment
+          type: stat
+          description: The deployment topology of GreptimeDB.
+          queries:
+            - expr: SELECT count(*) as datanode FROM information_schema.cluster_info WHERE peer_type = 'DATANODE';
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+            - expr: SELECT count(*) as frontend FROM information_schema.cluster_info WHERE peer_type = 'FRONTEND';
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+            - expr: SELECT count(*) as metasrv FROM information_schema.cluster_info WHERE peer_type = 'METASRV';
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+            - expr: SELECT count(*) as flownode FROM information_schema.cluster_info WHERE peer_type = 'FLOWNODE';
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+        - title: Database Resources
+          type: stat
+          description: The number of the key resources in GreptimeDB.
+          queries:
+            - expr: SELECT COUNT(*) as databases FROM information_schema.schemata WHERE schema_name NOT IN ('greptime_private', 'information_schema')
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+            - expr: SELECT COUNT(*) as tables FROM information_schema.tables WHERE table_schema != 'information_schema'
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+            - expr: SELECT COUNT(region_id) as regions FROM information_schema.region_peers
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+            - expr: SELECT COUNT(*) as flows FROM information_schema.flows
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+        - title: Data Size
+          type: stat
+          description: The data size of wal/index/manifest in the GreptimeDB.
+          unit: decbytes
+          queries:
+            - expr: SELECT SUM(memtable_size) * 0.42825 as WAL FROM information_schema.region_statistics;
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+            - expr: SELECT SUM(index_size) as index FROM information_schema.region_statistics;
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+            - expr: SELECT SUM(manifest_size) as manifest FROM information_schema.region_statistics;
+              datasource:
+                type: mysql
+                uid: ${information_schema}
+    - title: Ingestion
+      panels:
+        - title: Total Ingestion Rate
+          type: timeseries
+          description: |
+            Total ingestion rate.
+
+            Here we listed 3 primary protocols:
+
+            - Prometheus remote write
+            - Greptime's gRPC API (when using our ingest SDK)
+            - Log ingestion http API
+          unit: rowsps
+          queries:
+            - expr: sum(rate(greptime_table_operator_ingest_rows{}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: ingestion
+        - title: Ingestion Rate by Type
+          type: timeseries
+          description: |
+            Total ingestion rate.
+
+            Here we listed 3 primary protocols:
+
+            - Prometheus remote write
+            - Greptime's gRPC API (when using our ingest SDK)
+            - Log ingestion http API
+          unit: rowsps
+          queries:
+            - expr: sum(rate(greptime_servers_http_logs_ingestion_counter[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: http-logs
+            - expr: sum(rate(greptime_servers_prometheus_remote_write_samples[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: prometheus-remote-write
+    - title: Queries
+      panels:
+        - title: Total Query Rate
+          type: timeseries
+          description: |-
+            Total rate of query API calls by protocol. This metric is collected from frontends.
+
+            Here we listed 3 main protocols:
+            - MySQL
+            - Postgres
+            - Prometheus API
+
+            Note that there are some other minor query APIs like /sql are not included
+          unit: reqps
+          queries:
+            - expr: sum (rate(greptime_servers_mysql_query_elapsed_count{}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: mysql
+            - expr: sum (rate(greptime_servers_postgres_query_elapsed_count{}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: pg
+            - expr: sum (rate(greptime_servers_http_promql_elapsed_counte{}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: promql
+    - title: Resources
+      panels:
+        - title: Datanode Memory per Instance
+          type: timeseries
+          description: Current memory usage by instance
+          unit: decbytes
+          queries:
+            - expr: sum(process_resident_memory_bytes{}) by (instance, pod)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{ pod }}]'
+        - title: Datanode CPU Usage per Instance
+          type: timeseries
+          description: Current cpu usage by instance
+          unit: none
+          queries:
+            - expr: sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{ pod }}]'
+        - title: Frontend Memory per Instance
+          type: timeseries
+          description: Current memory usage by instance
+          unit: decbytes
+          queries:
+            - expr: sum(process_resident_memory_bytes{}) by (instance, pod)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{ pod }}]'
+        - title: Frontend CPU Usage per Instance
+          type: timeseries
+          description: Current cpu usage by instance
+          unit: none
+          queries:
+            - expr: sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{ pod }}]-cpu'
+        - title: Metasrv Memory per Instance
+          type: timeseries
+          description: Current memory usage by instance
+          unit: decbytes
+          queries:
+            - expr: sum(process_resident_memory_bytes{}) by (instance, pod)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{ pod }}]-resident'
+        - title: Metasrv CPU Usage per Instance
+          type: timeseries
+          description: Current cpu usage by instance
+          unit: none
+          queries:
+            - expr: sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{ pod }}]'
+        - title: Flownode Memory per Instance
+          type: timeseries
+          description: Current memory usage by instance
+          unit: decbytes
+          queries:
+            - expr: sum(process_resident_memory_bytes{}) by (instance, pod)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{ pod }}]'
+        - title: Flownode CPU Usage per Instance
+          type: timeseries
+          description: Current cpu usage by instance
+          unit: none
+          queries:
+            - expr: sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{ pod }}]'
+    - title: Frontend Requests
+      panels:
+        - title: HTTP QPS per Instance
+          type: timeseries
+          description: HTTP QPS per Instance.
+          unit: reqps
+          queries:
+            - expr: sum by(instance, pod, path, method, code) (rate(greptime_servers_http_requests_elapsed_count{path!~"/health|/metrics"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]'
+        - title: HTTP P99 per Instance
+          type: timeseries
+          description: HTTP P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, path, method, code) (rate(greptime_servers_http_requests_elapsed_bucket{path!~"/health|/metrics"}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]-p99'
+        - title: gRPC QPS per Instance
+          type: timeseries
+          description: gRPC QPS per Instance.
+          unit: reqps
+          queries:
+            - expr: sum by(instance, pod, path, code) (rate(greptime_servers_grpc_requests_elapsed_count{}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{path}}]-[{{code}}]'
+        - title: gRPC P99 per Instance
+          type: timeseries
+          description: gRPC P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, path, code) (rate(greptime_servers_grpc_requests_elapsed_bucket{}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{path}}]-[{{method}}]-[{{code}}]-p99'
+        - title: MySQL QPS per Instance
+          type: timeseries
+          description: MySQL QPS per Instance.
+          unit: reqps
+          queries:
+            - expr: sum by(pod, instance)(rate(greptime_servers_mysql_query_elapsed_count{}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+        - title: MySQL P99 per Instance
+          type: timeseries
+          description: MySQL P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(pod, instance, le) (rate(greptime_servers_mysql_query_elapsed_bucket{}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{ pod }}]-p99'
+        - title: PostgreSQL QPS per Instance
+          type: timeseries
+          description: PostgreSQL QPS per Instance.
+          unit: reqps
+          queries:
+            - expr: sum by(pod, instance)(rate(greptime_servers_postgres_query_elapsed_count{}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+        - title: PostgreSQL P99 per Instance
+          type: timeseries
+          description: PostgreSQL P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(pod,instance,le) (rate(greptime_servers_postgres_query_elapsed_bucket{}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-p99'
+    - title: Frontend to Datanode
+      panels:
+        - title: Ingest Rows per Instance
+          type: timeseries
+          description: Ingestion rate by row as in each frontend
+          unit: rowsps
+          queries:
+            - expr: sum by(instance, pod)(rate(greptime_table_operator_ingest_rows{}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+        - title: Region Call QPS per Instance
+          type: timeseries
+          description: Region Call QPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod, request_type) (rate(greptime_grpc_region_request_count{}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{request_type}}]'
+        - title: Region Call P99 per Instance
+          type: timeseries
+          description: Region Call P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, request_type) (rate(greptime_grpc_region_request_bucket{}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{request_type}}]'
+    - title: Mito Engine
+      panels:
+        - title: Request OPS per Instance
+          type: timeseries
+          description: Request QPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod, type) (rate(greptime_mito_handle_request_elapsed_count{}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{type}}]'
+        - title: Request P99 per Instance
+          type: timeseries
+          description: Request P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, type) (rate(greptime_mito_handle_request_elapsed_bucket{}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{type}}]'
+        - title: Write Buffer per Instance
+          type: timeseries
+          description: Write Buffer per Instance.
+          unit: decbytes
+          queries:
+            - expr: greptime_mito_write_buffer_bytes{}
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+        - title: Write Rows per Instance
+          type: timeseries
+          description: Ingestion size by row counts.
+          unit: rowsps
+          queries:
+            - expr: sum by (instance, pod) (rate(greptime_mito_write_rows_total{}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+        - title: Flush OPS per Instance
+          type: timeseries
+          description: Flush QPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod, reason) (rate(greptime_mito_flush_requests_total{}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{reason}}]'
+        - title: Write Stall per Instance
+          type: timeseries
+          description: Write Stall per Instance.
+          queries:
+            - expr: sum by(instance, pod) (greptime_mito_write_stall_total{})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+        - title: Read Stage OPS per Instance
+          type: timeseries
+          description: Read Stage OPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod) (rate(greptime_mito_read_stage_elapsed_count{ stage="total"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+        - title: Read Stage P99 per Instance
+          type: timeseries
+          description: Read Stage P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_read_stage_elapsed_bucket{}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]'
+        - title: Write Stage P99 per Instance
+          type: timeseries
+          description: Write Stage P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_write_stage_elapsed_bucket{}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]'
+        - title: Compaction OPS per Instance
+          type: timeseries
+          description: Compaction OPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod) (rate(greptime_mito_compaction_total_elapsed_count{}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{ instance }}]-[{{pod}}]'
+        - title: Compaction P99 per Instance by Stage
+          type: timeseries
+          description: Compaction latency by stage
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_bucket{}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-p99'
+        - title: Compaction P99 per Instance
+          type: timeseries
+          description: Compaction P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le,stage) (rate(greptime_mito_compaction_total_elapsed_bucket{}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-compaction'
+        - title: WAL write size
+          type: timeseries
+          description: Write-ahead logs write size as bytes. This chart includes stats of p95 and p99 size by instance, total WAL write rate.
+          unit: bytes
+          queries:
+            - expr: histogram_quantile(0.95, sum by(le,instance, pod) (rate(raft_engine_write_size_bucket[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-req-size-p95'
+            - expr: histogram_quantile(0.99, sum by(le,instance,pod) (rate(raft_engine_write_size_bucket[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-req-size-p99'
+            - expr: sum by (instance, pod)(rate(raft_engine_write_size_sum[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-throughput'
+        - title: Cached Bytes per Instance
+          type: timeseries
+          description: Cached Bytes per Instance.
+          unit: decbytes
+          queries:
+            - expr: greptime_mito_cache_bytes{}
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{type}}]'
+        - title: Inflight Compaction
+          type: timeseries
+          description: Ongoing compaction task count
+          unit: none
+          queries:
+            - expr: greptime_mito_inflight_compaction_count
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+        - title: WAL sync duration seconds
+          type: timeseries
+          description: Raft engine (local disk) log store sync latency, p99
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(le, type, node, instance, pod) (rate(raft_engine_sync_log_duration_seconds_bucket[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-p99'
+        - title: Log Store op duration seconds
+          type: timeseries
+          description: Write-ahead log operations latency at p99
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(le,logstore,optype,instance, pod) (rate(greptime_logstore_op_elapsed_bucket[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{logstore}}]-[{{optype}}]-p99'
+        - title: Inflight Flush
+          type: timeseries
+          description: Ongoing flush task count
+          unit: none
+          queries:
+            - expr: greptime_mito_inflight_flush_count
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]'
+    - title: OpenDAL
+      panels:
+        - title: QPS per Instance
+          type: timeseries
+          description: QPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]'
+        - title: Read QPS per Instance
+          type: timeseries
+          description: Read QPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{ operation="read"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]'
+        - title: Read P99 per Instance
+          type: timeseries
+          description: Read P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{operation="read"}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-{{scheme}}'
+        - title: Write QPS per Instance
+          type: timeseries
+          description: Write QPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{ operation="write"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-{{scheme}}'
+        - title: Write P99 per Instance
+          type: timeseries
+          description: Write P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{ operation="write"}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]'
+        - title: List QPS per Instance
+          type: timeseries
+          description: List QPS per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{ operation="list"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]'
+        - title: List P99 per Instance
+          type: timeseries
+          description: List P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{ operation="list"}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]'
+        - title: Other Requests per Instance
+          type: timeseries
+          description: Other Requests per Instance.
+          unit: ops
+          queries:
+            - expr: sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{operation!~"read|write|list|stat"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]'
+        - title: Other Request P99 per Instance
+          type: timeseries
+          description: Other Request P99 per Instance.
+          unit: s
+          queries:
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{ operation!~"read|write|list"}[$__rate_interval])))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]'
+        - title: Opendal traffic
+          type: timeseries
+          description: Total traffic as in bytes by instance and operation
+          unit: decbytes
+          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
+          type: state-timeline
+          description: Counter of region migration by source and destination
+          unit: none
+          queries:
+            - expr: greptime_meta_region_migration_stat{datanode_type="src"}
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: from-datanode-{{datanode_id}}
+            - expr: greptime_meta_region_migration_stat{datanode_type="desc"}
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: to-datanode-{{datanode_id}}
+        - title: Region migration error
+          type: timeseries
+          description: Counter of region migration error
+          unit: none
+          queries:
+            - expr: greptime_meta_region_migration_error
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: __auto
+        - title: Datanode load
+          type: timeseries
+          description: Gauge of load information of each datanode, collected via heartbeat between datanode and metasrv. This information is for metasrv to schedule workloads.
+          unit: none
+          queries:
+            - expr: greptime_datanode_load
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: __auto
+    - title: Flownode
+      panels:
+        - title: Flow Ingest / Output Rate
+          type: timeseries
+          description: Flow Ingest / Output Rate.
+          queries:
+            - expr: sum by(instance, pod, direction) (rate(greptime_flow_processed_rows[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{pod}}]-[{{instance}}]-[{{direction}}]'
+        - title: Flow Ingest Latency
+          type: timeseries
+          description: Flow Ingest Latency.
+          queries:
+            - expr: histogram_quantile(0.95, sum(rate(greptime_flow_insert_elapsed_bucket[$__rate_interval])) by (le, instance, pod))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-p95'
+            - expr: histogram_quantile(0.99, sum(rate(greptime_flow_insert_elapsed_bucket[$__rate_interval])) by (le, instance, pod))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-p99'
+        - title: Flow Operation Latency
+          type: timeseries
+          description: Flow Operation Latency.
+          queries:
+            - expr: histogram_quantile(0.95, sum(rate(greptime_flow_processing_time_bucket[$__rate_interval])) by (le,instance,pod,type))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{type}}]-p95'
+            - expr: histogram_quantile(0.99, sum(rate(greptime_flow_processing_time_bucket[$__rate_interval])) by (le,instance,pod,type))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{type}}]-p99'
+        - title: Flow Buffer Size per Instance
+          type: timeseries
+          description: Flow Buffer Size per Instance.
+          queries:
+            - expr: greptime_flow_input_buf_size
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}]'
+        - title: Flow Processing Error per Instance
+          type: timeseries
+          description: Flow Processing Error per Instance.
+          queries:
+            - expr: sum by(instance,pod,code) (rate(greptime_flow_errors[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{code}}]'

grafana/scripts/check.sh

@@ -0,0 +1,54 @@
+#!/usr/bin/env bash
+
+DASHBOARD_DIR=${1:-grafana/dashboards/metrics}
+
+check_dashboard_description() {
+  for dashboard in $(find $DASHBOARD_DIR -name "*.json"); do
+    echo "Checking $dashboard description"
+
+    # Use jq to check for panels with empty or missing descriptions
+    invalid_panels=$(cat $dashboard | jq -r '
+      .panels[]
+    | select((.type == "stats" or .type == "timeseries") and (.description == "" or .description == null))')
+
+    # Check if any invalid panels were found
+    if [[ -n "$invalid_panels" ]]; then
+      echo "Error: The following panels have empty or missing descriptions:"
+      echo "$invalid_panels"
+      exit 1
+    else
+      echo "All panels with type 'stats' or 'timeseries' have valid descriptions."
+    fi
+  done
+}
+
+check_dashboards_generation() {
+  ./grafana/scripts/gen-dashboards.sh
+
+  if [[ -n "$(git diff --name-only grafana/dashboards/metrics)" ]]; then
+    echo "Error: The dashboards are not generated correctly. You should execute the `make dashboards` command."
+    exit 1
+  fi
+}
+
+check_datasource() {
+  for dashboard in $(find $DASHBOARD_DIR -name "*.json"); do
+    echo "Checking $dashboard datasource"
+    jq -r '.panels[] | select(.type != "row") | .targets[] | [.datasource.type, .datasource.uid] | @tsv' $dashboard | while read -r type uid; do
+    # if the datasource is prometheus, check if the uid is ${metrics}
+    if [[ "$type" == "prometheus" && "$uid" != "\${metrics}" ]]; then
+      echo "Error: The datasource uid of $dashboard is not valid. It should be \${metrics}, got $uid"
+      exit 1
+    fi
+    # if the datasource is mysql, check if the uid is ${information_schema}
+    if [[ "$type" == "mysql" && "$uid" != "\${information_schema}" ]]; then
+      echo "Error: The datasource uid of $dashboard is not valid. It should be \${information_schema}, got $uid"
+      exit 1
+    fi
+    done
+  done
+}
+
+check_dashboards_generation
+check_dashboard_description
+check_datasource

grafana/scripts/gen-dashboards.sh

@@ -0,0 +1,25 @@
+#! /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
+
+remove_instance_filters() {
+  # Remove the instance filters for the standalone dashboards.
+  sed 's/instance=~\\"$datanode\\",//; s/instance=~\\"$datanode\\"//; s/instance=~\\"$frontend\\",//; s/instance=~\\"$frontend\\"//; s/instance=~\\"$metasrv\\",//; s/instance=~\\"$metasrv\\"//; s/instance=~\\"$flownode\\",//; s/instance=~\\"$flownode\\"//;' $CLUSTER_DASHBOARD_DIR/dashboard.json > $STANDALONE_DASHBOARD_DIR/dashboard.json
+}
+
+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
+}
+
+remove_instance_filters
+generate_intermediate_dashboards_and_docs

grafana/summary.sh

@@ -1,11 +0,0 @@
-#!/usr/bin/env bash
-
-BASEDIR=$(dirname "$0")
-echo '| Title | Description | Expressions |
-|---|---|---|'
-
-cat $BASEDIR/greptimedb-cluster.json | jq -r '
-  .panels |
-  map(select(.type == "stat" or .type == "timeseries")) |
-  .[] | "| \(.title) | \(.description | gsub("\n"; "<br>")) | \(.targets | map(.expr // .rawSql | "`\(.|gsub("\n"; "<br>"))`")  | join("<br>")) |"
-'

rust-toolchain.toml

@@ -1,2 +1,2 @@
 [toolchain]
-channel = "nightly-2024-12-25"
+channel = "nightly-2025-04-15"

src/api/src/helper.rs

@@ -514,6 +514,7 @@ fn query_request_type(request: &QueryRequest) -> &'static str {
         Some(Query::Sql(_)) => "query.sql",
         Some(Query::LogicalPlan(_)) => "query.logical_plan",
         Some(Query::PromRangeQuery(_)) => "query.prom_range",
+        Some(Query::InsertIntoPlan(_)) => "query.insert_into_plan",
         None => "query.empty",
     }
 }
@@ -1049,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.as_utf8().to_string())),
+            Value::String(v) => Some(ValueData::StringValue(v.into_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() {

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(':').context(InvalidConfigSnafu {
+    let (name, content) = opt.split_once(':').with_context(|| InvalidConfigSnafu {
         value: opt.to_string(),
         msg: "UserProviderOption must be in format `<option>:<value>`",
     })?;
@@ -57,6 +57,24 @@ 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;
 

src/auth/src/error.rs

@@ -38,6 +38,14 @@ 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)]
@@ -85,7 +93,7 @@ impl ErrorExt for Error {
     fn status_code(&self) -> StatusCode {
         match self {
             Error::InvalidConfig { .. } => StatusCode::InvalidArguments,
-            Error::IllegalParam { .. } => StatusCode::InvalidArguments,
+            Error::IllegalParam { .. } | Error::FromUtf8 { .. } => StatusCode::InvalidArguments,
             Error::FileWatch { .. } => StatusCode::InvalidArguments,
             Error::InternalState { .. } => StatusCode::Unexpected,
             Error::Io { .. } => StatusCode::StorageUnavailable,

src/auth/src/lib.rs

@@ -22,10 +22,12 @@ mod user_provider;
 pub mod tests;
 
 pub use common::{
-    auth_mysql, user_provider_from_option, userinfo_by_name, HashedPassword, Identity, Password,
+    auth_mysql, static_user_provider_from_option, 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

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;
+use snafu::{OptionExt, ResultExt};
 
-use crate::error::{InvalidConfigSnafu, Result};
+use crate::error::{FromUtf8Snafu, 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(crate) struct StaticUserProvider {
+pub struct StaticUserProvider {
     users: HashMap<String, Vec<u8>>,
 }
 
@@ -60,6 +60,18 @@ 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]

src/catalog/src/system_schema/pg_catalog/pg_namespace/oid_map.rs

@@ -84,12 +84,6 @@ 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);

src/catalog/src/table_source.rs

@@ -27,7 +27,7 @@ use session::context::QueryContextRef;
 use snafu::{ensure, OptionExt, ResultExt};
 use table::metadata::TableType;
 use table::table::adapter::DfTableProviderAdapter;
-mod dummy_catalog;
+pub mod dummy_catalog;
 use dummy_catalog::DummyCatalogList;
 use table::TableRef;
 

src/cli/Cargo.toml

@@ -51,7 +51,6 @@ 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

src/cli/src/cmd.rs

@@ -1,154 +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 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(),
-            },
-        );
-    }
-}

src/cli/src/error.rs

@@ -101,9 +101,6 @@ 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,
@@ -254,7 +251,6 @@ impl ErrorExt for Error {
             Error::MissingConfig { .. }
             | Error::LoadLayeredConfig { .. }
             | Error::IllegalConfig { .. }
-            | Error::InvalidReplCommand { .. }
             | Error::InitTimezone { .. }
             | Error::ConnectEtcd { .. }
             | Error::CreateDir { .. }

src/cli/src/helper.rs

@@ -1,112 +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::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![]))
-        }
-    }
-}

src/cli/src/lib.rs

@@ -13,15 +13,9 @@
 // limitations under the License.
 
 mod bench;
-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;
+pub mod error;
+mod export;
 mod import;
 
 use async_trait::async_trait;

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, MetadataValue};
+use tonic::metadata::{AsciiMetadataKey, AsciiMetadataValue, MetadataMap, MetadataValue};
 use tonic::transport::Channel;
 
 use crate::error::{
-    ConvertFlightDataSnafu, Error, FlightGetSnafu, IllegalFlightMessagesSnafu, InvalidAsciiSnafu,
+    ConvertFlightDataSnafu, Error, FlightGetSnafu, IllegalFlightMessagesSnafu,
     InvalidTonicMetadataValueSnafu, ServerSnafu,
 };
 use crate::{from_grpc_response, Client, Result};
@@ -165,26 +165,27 @@ impl Database {
 
         let mut request = tonic::Request::new(request);
         let metadata = request.metadata_mut();
-        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);
-        }
+        Self::put_hints(metadata, hints)?;
+
         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);
@@ -192,6 +193,36 @@ 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 {
@@ -212,39 +243,49 @@ impl Database {
     where
         S: AsRef<str>,
     {
-        self.do_get(Request::Query(QueryRequest {
+        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 {
             query: Some(Query::Sql(sql.as_ref().to_string())),
-        }))
-        .await
+        });
+        self.do_get(request, hints).await
     }
 
     pub async fn logical_plan(&self, logical_plan: Vec<u8>) -> Result<Output> {
-        self.do_get(Request::Query(QueryRequest {
+        let request = Request::Query(QueryRequest {
             query: Some(Query::LogicalPlan(logical_plan)),
-        }))
-        .await
+        });
+        self.do_get(request, &[]).await
     }
 
     pub async fn create(&self, expr: CreateTableExpr) -> Result<Output> {
-        self.do_get(Request::Ddl(DdlRequest {
+        let request = Request::Ddl(DdlRequest {
             expr: Some(DdlExpr::CreateTable(expr)),
-        }))
-        .await
+        });
+        self.do_get(request, &[]).await
     }
 
     pub async fn alter(&self, expr: AlterTableExpr) -> Result<Output> {
-        self.do_get(Request::Ddl(DdlRequest {
+        let request = Request::Ddl(DdlRequest {
             expr: Some(DdlExpr::AlterTable(expr)),
-        }))
-        .await
+        });
+        self.do_get(request, &[]).await
     }
 
-    async fn do_get(&self, request: Request) -> Result<Output> {
+    async fn do_get(&self, request: Request, hints: &[(&str, &str)]) -> 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| {
@@ -368,6 +409,11 @@ 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>,

src/client/src/error.rs

@@ -110,13 +110,6 @@ 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)]
@@ -143,10 +136,7 @@ impl ErrorExt for Error {
             | Error::ConvertFlightData { source, .. }
             | Error::CreateTlsChannel { source, .. } => source.status_code(),
             Error::IllegalGrpcClientState { .. } => StatusCode::Unexpected,
-
-            Error::InvalidAscii { .. } | Error::InvalidTonicMetadataValue { .. } => {
-                StatusCode::InvalidArguments
-            }
+            Error::InvalidTonicMetadataValue { .. } => StatusCode::InvalidArguments,
         }
     }
 

src/cmd/src/bin/greptime.rs

@@ -15,9 +15,11 @@
 #![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;
 
@@ -102,10 +104,10 @@ async fn main_body() -> Result<()> {
 async fn start(cli: Command) -> Result<()> {
     match cli.subcmd {
         SubCommand::Datanode(cmd) => {
-            cmd.build(cmd.load_options(&cli.global_options)?)
-                .await?
-                .run()
-                .await
+            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
         }
         SubCommand::Flownode(cmd) => {
             cmd.build(cmd.load_options(&cli.global_options)?)

src/cmd/src/cli.rs

@@ -58,7 +58,7 @@ impl App for Instance {
         false
     }
 
-    async fn stop(&self) -> Result<()> {
+    async fn stop(&mut self) -> Result<()> {
         Ok(())
     }
 }

src/cmd/src/datanode.rs

@@ -12,33 +12,27 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
+pub mod builder;
+
 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, DatanodeBuilder};
-use datanode::service::DatanodeServiceBuilder;
-use meta_client::{MetaClientOptions, MetaClientType};
-use servers::Mode;
-use snafu::{ensure, OptionExt, ResultExt};
+use datanode::datanode::Datanode;
+use meta_client::MetaClientOptions;
+use snafu::{ensure, ResultExt};
 use tracing_appender::non_blocking::WorkerGuard;
 
+use crate::datanode::builder::InstanceBuilder;
 use crate::error::{
-    LoadLayeredConfigSnafu, MetaClientInitSnafu, MissingConfigSnafu, Result, ShutdownDatanodeSnafu,
-    StartDatanodeSnafu,
+    LoadLayeredConfigSnafu, MissingConfigSnafu, Result, ShutdownDatanodeSnafu, StartDatanodeSnafu,
 };
 use crate::options::{GlobalOptions, GreptimeOptions};
-use crate::{log_versions, App};
+use crate::App;
 
 pub const APP_NAME: &str = "greptime-datanode";
 
@@ -83,7 +77,7 @@ impl App for Instance {
         self.datanode.start().await.context(StartDatanodeSnafu)
     }
 
-    async fn stop(&self) -> Result<()> {
+    async fn stop(&mut self) -> Result<()> {
         self.datanode
             .shutdown()
             .await
@@ -98,8 +92,8 @@ pub struct Command {
 }
 
 impl Command {
-    pub async fn build(&self, opts: DatanodeOptions) -> Result<Instance> {
-        self.subcmd.build(opts).await
+    pub async fn build_with(&self, builder: InstanceBuilder) -> Result<Instance> {
+        self.subcmd.build_with(builder).await
     }
 
     pub fn load_options(&self, global_options: &GlobalOptions) -> Result<DatanodeOptions> {
@@ -115,9 +109,12 @@ enum SubCommand {
 }
 
 impl SubCommand {
-    async fn build(&self, opts: DatanodeOptions) -> Result<Instance> {
+    async fn build_with(&self, builder: InstanceBuilder) -> Result<Instance> {
         match self {
-            SubCommand::Start(cmd) => cmd.build(opts).await,
+            SubCommand::Start(cmd) => {
+                info!("Building datanode with {:#?}", cmd);
+                builder.build().await
+            }
         }
     }
 }
@@ -263,74 +260,6 @@ 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)]
@@ -352,7 +281,6 @@ 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
 
@@ -379,7 +307,6 @@ 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
 
@@ -545,7 +472,6 @@ 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"

src/cmd/src/datanode/builder.rs

@@ -0,0 +1,137 @@
+// 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))
+    }
+}

src/cmd/src/error.rs

@@ -177,9 +177,6 @@ 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,
@@ -331,7 +328,6 @@ impl ErrorExt for Error {
             Error::MissingConfig { .. }
             | Error::LoadLayeredConfig { .. }
             | Error::IllegalConfig { .. }
-            | Error::InvalidReplCommand { .. }
             | Error::InitTimezone { .. }
             | Error::ConnectEtcd { .. }
             | Error::CreateDir { .. }

src/cmd/src/flownode.rs

@@ -33,7 +33,8 @@ use common_telemetry::info;
 use common_telemetry::logging::TracingOptions;
 use common_version::{short_version, version};
 use flow::{
-    FlownodeBuilder, FlownodeInstance, FlownodeServiceBuilder, FrontendClient, FrontendInvoker,
+    get_flow_auth_options, FlownodeBuilder, FlownodeInstance, FlownodeServiceBuilder,
+    FrontendClient, FrontendInvoker,
 };
 use meta_client::{MetaClientOptions, MetaClientType};
 use snafu::{ensure, OptionExt, ResultExt};
@@ -82,10 +83,14 @@ 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(&self) -> Result<()> {
+    async fn stop(&mut self) -> Result<()> {
         self.flownode
             .shutdown()
             .await
@@ -151,6 +156,9 @@ 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 {
@@ -214,6 +222,10 @@ 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 {
@@ -238,9 +250,15 @@ 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'" })?;
@@ -315,10 +333,12 @@ impl StartCommand {
         );
 
         let flow_metadata_manager = Arc::new(FlowMetadataManager::new(cached_meta_backend.clone()));
-        let frontend_client = FrontendClient::from_meta_client(meta_client.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 flownode_builder = FlownodeBuilder::new(
             opts.clone(),
-            Plugins::new(),
+            plugins,
             table_metadata_manager,
             catalog_manager.clone(),
             flow_metadata_manager,
@@ -331,7 +351,6 @@ impl StartCommand {
             .with_grpc_server(flownode.flownode_server().clone())
             .enable_http_service()
             .build()
-            .await
             .context(StartFlownodeSnafu)?;
         flownode.setup_services(services);
         let flownode = flownode;
@@ -345,7 +364,7 @@ impl StartCommand {
         let client = Arc::new(NodeClients::new(channel_config));
 
         let invoker = FrontendInvoker::build_from(
-            flownode.flow_worker_manager().clone(),
+            flownode.flow_engine().streaming_engine(),
             catalog_manager.clone(),
             cached_meta_backend.clone(),
             layered_cache_registry.clone(),
@@ -355,7 +374,9 @@ impl StartCommand {
         .await
         .context(StartFlownodeSnafu)?;
         flownode
-            .flow_worker_manager()
+            .flow_engine()
+            .streaming_engine()
+            // TODO(discord9): refactor and avoid circular reference
             .set_frontend_invoker(invoker)
             .await;
 

src/cmd/src/frontend.rs

@@ -89,7 +89,7 @@ impl App for Instance {
             .context(error::StartFrontendSnafu)
     }
 
-    async fn stop(&self) -> Result<()> {
+    async fn stop(&mut self) -> Result<()> {
         self.frontend
             .shutdown()
             .await
@@ -382,7 +382,6 @@ impl StartCommand {
 
         let servers = Services::new(opts, instance.clone(), plugins)
             .build()
-            .await
             .context(error::StartFrontendSnafu)?;
 
         let frontend = Frontend {
@@ -448,8 +447,6 @@ 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"
@@ -538,8 +535,6 @@ 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"
 

src/cmd/src/lib.rs

@@ -74,7 +74,7 @@ pub trait App: Send {
         true
     }
 
-    async fn stop(&self) -> Result<()>;
+    async fn stop(&mut self) -> Result<()>;
 
     async fn run(&mut self) -> Result<()> {
         info!("Starting app: {}", self.name());

src/cmd/src/metasrv.rs

@@ -12,6 +12,7 @@
 // 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;
@@ -68,7 +69,7 @@ impl App for Instance {
         self.instance.start().await.context(StartMetaServerSnafu)
     }
 
-    async fn stop(&self) -> Result<()> {
+    async fn stop(&mut self) -> Result<()> {
         self.instance
             .shutdown()
             .await
@@ -131,8 +132,8 @@ impl SubCommand {
     }
 }
 
-#[derive(Debug, Default, Parser)]
-struct StartCommand {
+#[derive(Default, Parser)]
+pub struct StartCommand {
     /// The address to bind the gRPC server.
     #[clap(long, alias = "bind-addr")]
     rpc_bind_addr: Option<String>,
@@ -171,8 +172,29 @@ 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 {
-    fn load_options(&self, global_options: &GlobalOptions) -> Result<MetasrvOptions> {
+    pub 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(),
@@ -184,6 +206,15 @@ 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,
@@ -261,7 +292,7 @@ impl StartCommand {
         Ok(())
     }
 
-    async fn build(&self, opts: MetasrvOptions) -> Result<Instance> {
+    pub async fn build(&self, opts: MetasrvOptions) -> Result<Instance> {
         common_runtime::init_global_runtimes(&opts.runtime);
 
         let guard = common_telemetry::init_global_logging(

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, FlowWorkerManager, FlownodeBuilder, FlownodeInstance, FlownodeOptions,
-    FrontendClient, FrontendInvoker,
+    FlowConfig, FlownodeBuilder, FlownodeInstance, FlownodeOptions, FrontendClient,
+    FrontendInvoker, GrpcQueryHandlerWithBoxedError, StreamingEngine,
 };
 use frontend::frontend::{Frontend, FrontendOptions};
 use frontend::instance::builder::FrontendBuilder;
@@ -75,7 +75,6 @@ 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;
@@ -256,8 +255,8 @@ pub struct Instance {
 
 impl Instance {
     /// Find the socket addr of a server by its `name`.
-    pub async fn server_addr(&self, name: &str) -> Option<SocketAddr> {
-        self.frontend.server_handlers().addr(name).await
+    pub fn server_addr(&self, name: &str) -> Option<SocketAddr> {
+        self.frontend.server_handlers().addr(name)
     }
 }
 
@@ -294,7 +293,7 @@ impl App for Instance {
         Ok(())
     }
 
-    async fn stop(&self) -> Result<()> {
+    async fn stop(&mut self) -> Result<()> {
         self.frontend
             .shutdown()
             .await
@@ -497,12 +496,9 @@ impl StartCommand {
             .build(),
         );
 
-        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 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 information_extension = Arc::new(StandaloneInformationExtension::new(
             datanode.region_server(),
@@ -524,17 +520,17 @@ impl StartCommand {
             ..Default::default()
         };
 
-        // TODO(discord9): for standalone not use grpc, but just somehow get a handler to frontend grpc client without
+        // for standalone not use grpc, but get a handler to frontend grpc client without
         // actually make a connection
-        let fe_server_addr = fe_opts.grpc.bind_addr.clone();
-        let frontend_client = FrontendClient::from_static_grpc_addr(fe_server_addr);
+        let (frontend_client, frontend_instance_handler) =
+            FrontendClient::from_empty_grpc_handler();
         let flow_builder = FlownodeBuilder::new(
             flownode_options,
             plugins.clone(),
             table_metadata_manager.clone(),
             catalog_manager.clone(),
             flow_metadata_manager.clone(),
-            Arc::new(frontend_client),
+            Arc::new(frontend_client.clone()),
         );
         let flownode = flow_builder
             .build()
@@ -544,15 +540,15 @@ impl StartCommand {
 
         // set the ref to query for the local flow state
         {
-            let flow_worker_manager = flownode.flow_worker_manager();
+            let flow_streaming_engine = flownode.flow_engine().streaming_engine();
             information_extension
-                .set_flow_worker_manager(flow_worker_manager.clone())
+                .set_flow_streaming_engine(flow_streaming_engine)
                 .await;
         }
 
         let node_manager = Arc::new(StandaloneDatanodeManager {
             region_server: datanode.region_server(),
-            flow_server: flownode.flow_worker_manager(),
+            flow_server: flownode.flow_engine(),
         });
 
         let table_id_sequence = Arc::new(
@@ -606,10 +602,19 @@ impl StartCommand {
         .context(error::StartFrontendSnafu)?;
         let fe_instance = Arc::new(fe_instance);
 
-        let flow_worker_manager = flownode.flow_worker_manager();
+        // set the frontend client for flownode
+        let grpc_handler = fe_instance.clone() as Arc<dyn GrpcQueryHandlerWithBoxedError>;
+        let weak_grpc_handler = Arc::downgrade(&grpc_handler);
+        frontend_instance_handler
+            .lock()
+            .unwrap()
+            .replace(weak_grpc_handler);
+
+        // set the frontend invoker for flownode
+        let flow_streaming_engine = 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_worker_manager.clone(),
+            flow_streaming_engine.clone(),
             catalog_manager.clone(),
             kv_backend.clone(),
             layered_cache_registry.clone(),
@@ -618,14 +623,13 @@ impl StartCommand {
         )
         .await
         .context(error::StartFlownodeSnafu)?;
-        flow_worker_manager.set_frontend_invoker(invoker).await;
+        flow_streaming_engine.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 {
@@ -694,7 +698,7 @@ pub struct StandaloneInformationExtension {
     region_server: RegionServer,
     procedure_manager: ProcedureManagerRef,
     start_time_ms: u64,
-    flow_worker_manager: RwLock<Option<Arc<FlowWorkerManager>>>,
+    flow_streaming_engine: RwLock<Option<Arc<StreamingEngine>>>,
 }
 
 impl StandaloneInformationExtension {
@@ -703,14 +707,14 @@ impl StandaloneInformationExtension {
             region_server,
             procedure_manager,
             start_time_ms: common_time::util::current_time_millis() as u64,
-            flow_worker_manager: RwLock::new(None),
+            flow_streaming_engine: RwLock::new(None),
         }
     }
 
-    /// Set the flow worker manager for the standalone instance.
-    pub async fn set_flow_worker_manager(&self, flow_worker_manager: Arc<FlowWorkerManager>) {
-        let mut guard = self.flow_worker_manager.write().await;
-        *guard = Some(flow_worker_manager);
+    /// 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);
     }
 }
 
@@ -789,7 +793,7 @@ impl InformationExtension for StandaloneInformationExtension {
 
     async fn flow_stats(&self) -> std::result::Result<Option<FlowStat>, Self::Error> {
         Ok(Some(
-            self.flow_worker_manager
+            self.flow_streaming_engine
                 .read()
                 .await
                 .as_ref()
@@ -849,8 +853,6 @@ 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]
@@ -981,8 +983,6 @@ 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"
 

src/cmd/tests/load_config_test.rs

@@ -74,6 +74,7 @@ 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 {
@@ -216,6 +217,7 @@ 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 {

src/common/base/src/plugins.rs

@@ -31,7 +31,8 @@ impl Plugins {
     }
 
     pub fn insert<T: 'static + Send + Sync>(&self, value: T) {
-        let _ = self.write().insert(value);
+        let last = self.write().insert(value);
+        assert!(last.is_none(), "each type of plugins must be one and only");
     }
 
     pub fn get<T: 'static + Send + Sync + Clone>(&self) -> Option<T> {
@@ -137,4 +138,12 @@ mod tests {
         assert_eq!(plugins.len(), 2);
         assert!(!plugins.is_empty());
     }
+
+    #[test]
+    #[should_panic(expected = "each type of plugins must be one and only")]
+    fn test_plugin_uniqueness() {
+        let plugins = Plugins::new();
+        plugins.insert(1i32);
+        plugins.insert(2i32);
+    }
 }

src/common/config/src/config.rs

@@ -111,11 +111,9 @@ mod tests {
     use serde::{Deserialize, Serialize};
 
     use super::*;
-    use crate::Mode;
 
-    #[derive(Debug, Serialize, Deserialize)]
+    #[derive(Debug, Serialize, Deserialize, Default)]
     struct TestDatanodeConfig {
-        mode: Mode,
         node_id: Option<u64>,
         logging: LoggingOptions,
         meta_client: Option<MetaClientOptions>,
@@ -123,19 +121,6 @@ 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"])
@@ -146,7 +131,6 @@ 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"

src/common/config/src/lib.rs

@@ -26,16 +26,6 @@ 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 {

src/common/function/Cargo.toml

@@ -13,7 +13,7 @@ default = ["geo"]
 geo = ["geohash", "h3o", "s2", "wkt", "geo-types", "dep:geo"]
 
 [dependencies]
-ahash = "0.8"
+ahash.workspace = true
 api.workspace = true
 arc-swap = "1.0"
 async-trait.workspace = true

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_STATE_NAME};
+pub use uddsketch_state::{UddSketchState, UDDSKETCH_MERGE_NAME, UDDSKETCH_STATE_NAME};

src/common/function/src/aggr/uddsketch_state.rs

@@ -31,23 +31,28 @@ 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";
 
-#[derive(Debug)]
+pub const UDDSKETCH_MERGE_NAME: &str = "uddsketch_merge";
+
+#[derive(Debug, Serialize, Deserialize)]
 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 udf_impl() -> AggregateUDF {
+    pub fn state_udf_impl() -> AggregateUDF {
         create_udaf(
             UDDSKETCH_STATE_NAME,
             vec![DataType::Int64, DataType::Float64, DataType::Float64],
@@ -61,18 +66,55 @@ 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]) {
-        if let Ok(uddsketch) = bincode::deserialize::<UDDSketch>(raw) {
-            if uddsketch.count() != 0 {
-                self.uddsketch.merge_sketch(&uddsketch);
+    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);
             }
         } else {
             trace!("Warning: Failed to deserialize UDDSketch from {:?}", raw);
+            return Err(DataFusionError::Plan(
+                "Failed to deserialize UDDSketch from binary".to_string(),
+            ));
         }
+
+        Ok(())
     }
 }
 
@@ -113,9 +155,21 @@ 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
-        let f64_array = as_primitive_array::<Float64Type>(array)?;
-        for v in f64_array.iter().flatten() {
-            self.update(v);
+        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()
+                )
+            }
         }
 
         Ok(())
@@ -123,7 +177,7 @@ impl DfAccumulator for UddSketchState {
 
     fn evaluate(&mut self) -> DfResult<ScalarValue> {
         Ok(ScalarValue::Binary(Some(
-            bincode::serialize(&self.uddsketch).map_err(|e| {
+            bincode::serialize(&self).map_err(|e| {
                 DataFusionError::Internal(format!("Failed to serialize UDDSketch: {}", e))
             })?,
         )))
@@ -150,7 +204,7 @@ impl DfAccumulator for UddSketchState {
 
     fn state(&mut self) -> DfResult<Vec<ScalarValue>> {
         Ok(vec![ScalarValue::Binary(Some(
-            bincode::serialize(&self.uddsketch).map_err(|e| {
+            bincode::serialize(&self).map_err(|e| {
                 DataFusionError::Internal(format!("Failed to serialize UDDSketch: {}", e))
             })?,
         ))])
@@ -160,7 +214,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(())
@@ -182,8 +236,8 @@ mod tests {
 
         let result = state.evaluate().unwrap();
         if let ScalarValue::Binary(Some(bytes)) = result {
-            let deserialized: UDDSketch = bincode::deserialize(&bytes).unwrap();
-            assert_eq!(deserialized.count(), 3);
+            let deserialized: UddSketchState = bincode::deserialize(&bytes).unwrap();
+            assert_eq!(deserialized.uddsketch.count(), 3);
         } else {
             panic!("Expected binary scalar value");
         }
@@ -201,13 +255,15 @@ 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);
+            new_state.merge(bytes).unwrap();
 
             // Verify the merged state matches original by comparing deserialized values
-            let original_sketch: UDDSketch = bincode::deserialize(bytes).unwrap();
+            let original_sketch: UddSketchState = bincode::deserialize(bytes).unwrap();
+            let original_sketch = original_sketch.uddsketch;
             let new_result = new_state.evaluate().unwrap();
             if let ScalarValue::Binary(Some(new_bytes)) = new_result {
-                let new_sketch: UDDSketch = bincode::deserialize(&new_bytes).unwrap();
+                let new_sketch: UddSketchState = bincode::deserialize(&new_bytes).unwrap();
+                let new_sketch = new_sketch.uddsketch;
                 assert_eq!(original_sketch.count(), new_sketch.count());
                 assert_eq!(original_sketch.sum(), new_sketch.sum());
                 assert_eq!(original_sketch.mean(), new_sketch.mean());
@@ -244,7 +300,8 @@ mod tests {
 
         let result = state.evaluate().unwrap();
         if let ScalarValue::Binary(Some(bytes)) = result {
-            let deserialized: UDDSketch = bincode::deserialize(&bytes).unwrap();
+            let deserialized: UddSketchState = bincode::deserialize(&bytes).unwrap();
+            let deserialized = deserialized.uddsketch;
             assert_eq!(deserialized.count(), 3);
         } else {
             panic!("Expected binary scalar value");
@@ -273,7 +330,8 @@ mod tests {
 
             let result = merged_state.evaluate().unwrap();
             if let ScalarValue::Binary(Some(bytes)) = result {
-                let deserialized: UDDSketch = bincode::deserialize(&bytes).unwrap();
+                let deserialized: UddSketchState = bincode::deserialize(&bytes).unwrap();
+                let deserialized = deserialized.uddsketch;
                 assert_eq!(deserialized.count(), 2);
             } else {
                 panic!("Expected binary scalar value");

src/common/function/src/scalars/matches_term.rs

@@ -12,8 +12,9 @@
 // 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;
@@ -126,9 +127,10 @@ impl Function for MatchesTermFunction {
             let term = term_column.get_ref(0).as_string().unwrap();
             match term {
                 None => {
-                    return Ok(Arc::new(BooleanVector::from_iter(
-                        iter::repeat(None).take(text_column.len()),
-                    )));
+                    return Ok(Arc::new(BooleanVector::from_iter(repeat_n(
+                        None,
+                        text_column.len(),
+                    ))));
                 }
                 Some(term) => Some(MatchesTermFinder::new(term)),
             }
@@ -217,7 +219,7 @@ impl MatchesTermFinder {
         }
 
         let mut pos = 0;
-        while let Some(found_pos) = self.finder.find(text[pos..].as_bytes()) {
+        while let Some(found_pos) = self.finder.find(&text.as_bytes()[pos..]) {
             let actual_pos = pos + found_pos;
 
             let prev_ok = self.starts_with_non_alnum

src/common/function/src/scalars/timestamp.rs

@@ -13,10 +13,8 @@
 // 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;
@@ -26,6 +24,5 @@ pub(crate) struct TimestampFunction;
 impl TimestampFunction {
     pub fn register(registry: &FunctionRegistry) {
         registry.register(Arc::new(ToUnixtimeFunction));
-        registry.register(Arc::new(GreatestFunction));
     }
 }

src/common/function/src/scalars/timestamp/greatest.rs

@@ -1,328 +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::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);
-}

src/common/function/src/scalars/uddsketch_calc.rs

@@ -115,6 +115,13 @@ 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));

src/common/grpc/src/flight.rs

@@ -192,6 +192,10 @@ impl FlightDecoder {
             }
         }
     }
+
+    pub fn schema(&self) -> Option<&SchemaRef> {
+        self.schema.as_ref()
+    }
 }
 
 pub fn flight_messages_to_recordbatches(messages: Vec<FlightMessage>) -> Result<RecordBatches> {

src/common/grpc/src/lib.rs

@@ -18,4 +18,5 @@ pub mod flight;
 pub mod precision;
 pub mod select;
 
+pub use arrow_flight::FlightData;
 pub use error::Error;

src/common/meta/src/cache/flow/table_flownode.rs

@@ -187,6 +187,7 @@ 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),

src/common/meta/src/ddl/create_flow.rs

@@ -38,7 +38,7 @@ use table::metadata::TableId;
 use crate::cache_invalidator::Context;
 use crate::ddl::utils::{add_peer_context_if_needed, handle_retry_error};
 use crate::ddl::DdlContext;
-use crate::error::{self, Result};
+use crate::error::{self, Result, UnexpectedSnafu};
 use crate::instruction::{CacheIdent, CreateFlow};
 use crate::key::flow::flow_info::FlowInfoValue;
 use crate::key::flow::flow_route::FlowRouteValue;
@@ -171,7 +171,7 @@ impl CreateFlowProcedure {
         }
         self.data.state = CreateFlowState::CreateFlows;
         // determine flow type
-        self.data.flow_type = Some(determine_flow_type(&self.data.task));
+        self.data.flow_type = Some(get_flow_type_from_options(&self.data.task)?);
 
         Ok(Status::executing(true))
     }
@@ -196,8 +196,8 @@ impl CreateFlowProcedure {
             });
         }
         info!(
-            "Creating flow({:?}) on flownodes with peers={:?}",
-            self.data.flow_id, self.data.peers
+            "Creating flow({:?}, type={:?}) on flownodes with peers={:?}",
+            self.data.flow_id, self.data.flow_type, self.data.peers
         );
         join_all(create_flow)
             .await
@@ -306,8 +306,20 @@ impl Procedure for CreateFlowProcedure {
     }
 }
 
-pub fn determine_flow_type(_flow_task: &CreateFlowTask) -> FlowType {
-    FlowType::Batching
+pub fn get_flow_type_from_options(flow_task: &CreateFlowTask) -> Result<FlowType> {
+    let flow_type = flow_task
+        .flow_options
+        .get(FlowType::FLOW_TYPE_KEY)
+        .map(|s| s.as_str());
+    match flow_type {
+        Some(FlowType::BATCHING) => Ok(FlowType::Batching),
+        Some(FlowType::STREAMING) => Ok(FlowType::Streaming),
+        Some(unknown) => UnexpectedSnafu {
+            err_msg: format!("Unknown flow type: {}", unknown),
+        }
+        .fail(),
+        None => Ok(FlowType::Batching),
+    }
 }
 
 /// The state of [CreateFlowProcedure].
@@ -437,6 +449,7 @@ 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,

src/common/meta/src/ddl/tests/create_flow.rs

@@ -46,7 +46,7 @@ pub(crate) fn test_create_flow_task(
         create_if_not_exists,
         expire_after: Some(300),
         comment: "".to_string(),
-        sql: "raw_sql".to_string(),
+        sql: "select 1".to_string(),
         flow_options: Default::default(),
     }
 }

src/common/meta/src/error.rs

@@ -401,6 +401,13 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Invalid flow request body: {:?}", body))]
+    InvalidFlowRequestBody {
+        body: Box<Option<api::v1::flow::flow_request::Body>>,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Failed to get kv cache, err: {}", err_msg))]
     GetKvCache { err_msg: String },
 
@@ -783,6 +790,14 @@ 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>;
@@ -853,7 +868,9 @@ impl ErrorExt for Error {
             | TlsConfig { .. }
             | InvalidSetDatabaseOption { .. }
             | InvalidUnsetDatabaseOption { .. }
-            | InvalidTopicNamePrefix { .. } => StatusCode::InvalidArguments,
+            | InvalidTopicNamePrefix { .. }
+            | InvalidTimeZone { .. } => StatusCode::InvalidArguments,
+            InvalidFlowRequestBody { .. } => StatusCode::InvalidArguments,
 
             FlowNotFound { .. } => StatusCode::FlowNotFound,
             FlowRouteNotFound { .. } => StatusCode::Unexpected,

src/common/meta/src/instruction.rs

@@ -217,7 +217,9 @@ pub enum Instruction {
     /// Invalidates batch cache.
     InvalidateCaches(Vec<CacheIdent>),
     /// Flushes regions.
-    FlushRegion(FlushRegions),
+    FlushRegions(FlushRegions),
+    /// Flushes a single region.
+    FlushRegion(RegionId),
 }
 
 /// The reply of [UpgradeRegion].
@@ -248,6 +250,7 @@ pub enum InstructionReply {
     CloseRegion(SimpleReply),
     UpgradeRegion(UpgradeRegionReply),
     DowngradeRegion(DowngradeRegionReply),
+    FlushRegion(SimpleReply),
 }
 
 impl Display for InstructionReply {
@@ -259,6 +262,7 @@ impl Display for InstructionReply {
             Self::DowngradeRegion(reply) => {
                 write!(f, "InstructionReply::DowngradeRegion({})", reply)
             }
+            Self::FlushRegion(reply) => write!(f, "InstructionReply::FlushRegion({})", reply),
         }
     }
 }

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(crate) mod txn_helper;
+pub mod txn_helper;
 pub mod view_info;
 
 use std::collections::{BTreeMap, HashMap, HashSet};

src/common/meta/src/key/flow.rs

@@ -452,6 +452,7 @@ mod tests {
         };
         FlowInfoValue {
             catalog_name: catalog_name.to_string(),
+            query_context: None,
             flow_name: flow_name.to_string(),
             source_table_ids,
             sink_table_name,
@@ -625,6 +626,7 @@ 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,
@@ -864,6 +866,7 @@ 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,

src/common/meta/src/key/flow/flow_info.rs

@@ -121,6 +121,13 @@ 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.
@@ -155,6 +162,10 @@ 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
     }

src/common/meta/src/key/table_route.rs

@@ -478,10 +478,11 @@ 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(crate) fn build_update_txn(
+    pub fn build_update_txn(
         &self,
         table_id: TableId,
         current_table_route_value: &DeserializedValueWithBytes<TableRouteValue>,

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(crate) fn filter(key: Vec<u8>) -> impl FnMut(&mut TxnOpGetResponseSet) -> Option<Vec<u8>> {
+    pub 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(crate) fn decode_with<F, T>(
+    pub fn decode_with<F, T>(
         mut f: F,
     ) -> impl FnMut(&mut TxnOpGetResponseSet) -> Result<Option<DeserializedValueWithBytes<T>>>
     where

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]

src/common/meta/src/kv_backend/util.rs

@@ -0,0 +1,85 @@
+// 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"
+        );
+    }
+}

src/common/meta/src/lib.rs

@@ -15,8 +15,6 @@
 #![feature(assert_matches)]
 #![feature(btree_extract_if)]
 #![feature(let_chains)]
-#![feature(extract_if)]
-#![feature(hash_extract_if)]
 
 pub mod cache;
 pub mod cache_invalidator;

src/common/meta/src/lock_key.rs

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

src/common/meta/src/region_registry.rs

@@ -113,8 +113,10 @@ impl LeaderRegionManifestInfo {
     pub fn prunable_entry_id(&self) -> u64 {
         match self {
             LeaderRegionManifestInfo::Mito {
-                flushed_entry_id, ..
-            } => *flushed_entry_id,
+                flushed_entry_id,
+                topic_latest_entry_id,
+                ..
+            } => (*flushed_entry_id).max(*topic_latest_entry_id),
             LeaderRegionManifestInfo::Metric {
                 data_flushed_entry_id,
                 data_topic_latest_entry_id,

src/common/meta/src/rpc/ddl.rs

@@ -35,17 +35,20 @@ use api::v1::{
 };
 use base64::engine::general_purpose;
 use base64::Engine as _;
-use common_time::DatabaseTimeToLive;
+use common_time::{DatabaseTimeToLive, Timezone};
 use prost::Message;
 use serde::{Deserialize, Serialize};
 use serde_with::{serde_as, DefaultOnNull};
-use session::context::QueryContextRef;
+use session::context::{QueryContextBuilder, 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, InvalidUnsetDatabaseOptionSnafu, Result};
+use crate::error::{
+    self, InvalidSetDatabaseOptionSnafu, InvalidTimeZoneSnafu, InvalidUnsetDatabaseOptionSnafu,
+    Result,
+};
 use crate::key::FlowId;
 
 /// DDL tasks
@@ -1202,7 +1205,7 @@ impl From<DropFlowTask> for PbDropFlowTask {
     }
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct QueryContext {
     current_catalog: String,
     current_schema: String,
@@ -1223,6 +1226,19 @@ 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 {

src/common/meta/src/rpc/router.rs

@@ -176,15 +176,12 @@ impl TableRoute {
                 })?
                 .into();
 
-            let leader_peer = peers
-                .get(region_route.leader_peer_index as usize)
-                .cloned()
-                .map(Into::into);
+            let leader_peer = peers.get(region_route.leader_peer_index as usize).cloned();
 
             let follower_peers = region_route
                 .follower_peer_indexes
                 .into_iter()
-                .filter_map(|x| peers.get(x as usize).cloned().map(Into::into))
+                .filter_map(|x| peers.get(x as usize).cloned())
                 .collect::<Vec<_>>();
 
             region_routes.push(RegionRoute {

src/common/procedure-test/src/lib.rs

@@ -18,11 +18,13 @@ 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,
+    ProcedureWithId, Result, Status, StringKey,
 };
 
 /// A Mock [ContextProvider].
@@ -30,6 +32,7 @@ use common_procedure::{
 pub struct MockContextProvider {
     states: HashMap<ProcedureId, ProcedureState>,
     poison_manager: InMemoryPoisonStore,
+    dynamic_key_lock: Arc<KeyRwLock<String>>,
 }
 
 impl MockContextProvider {
@@ -38,6 +41,7 @@ impl MockContextProvider {
         MockContextProvider {
             states,
             poison_manager: InMemoryPoisonStore::default(),
+            dynamic_key_lock: Arc::new(KeyRwLock::new()),
         }
     }
 
@@ -58,6 +62,10 @@ 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].

src/common/procedure/src/lib.rs

@@ -20,6 +20,7 @@ pub mod error;
 pub mod local;
 pub mod options;
 mod procedure;
+pub mod rwlock;
 pub mod store;
 pub mod watcher;
 
@@ -28,8 +29,8 @@ pub mod test_util;
 
 pub use crate::error::{Error, Result};
 pub use crate::procedure::{
-    BoxedProcedure, BoxedProcedureLoader, Context, ContextProvider, LockKey, Output, ParseIdError,
-    PoisonKey, PoisonKeys, Procedure, ProcedureId, ProcedureInfo, ProcedureManager,
-    ProcedureManagerRef, ProcedureState, ProcedureWithId, Status, StringKey,
+    BoxedProcedure, BoxedProcedureLoader, Context, ContextProvider, ContextProviderRef, LockKey,
+    Output, ParseIdError, PoisonKey, PoisonKeys, Procedure, ProcedureId, ProcedureInfo,
+    ProcedureManager, ProcedureManagerRef, ProcedureState, ProcedureWithId, Status, StringKey,
 };
 pub use crate::watcher::Watcher;

src/common/procedure/src/local.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 mod runner;
-mod rwlock;
 
 use std::collections::hash_map::Entry;
 use std::collections::{HashMap, HashSet, VecDeque};
@@ -30,7 +29,6 @@ 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,
@@ -38,11 +36,12 @@ 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, Watcher,
+    ProcedureState, ProcedureWithId, StringKey, Watcher,
 };
 
 /// The expired time of a procedure's metadata.
@@ -157,12 +156,80 @@ 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)>>,
@@ -199,6 +266,10 @@ 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 {
@@ -206,6 +277,7 @@ 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()),

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,6 +581,7 @@ impl Runner {
 
 #[cfg(test)]
 mod tests {
+    use std::sync::atomic::{AtomicU64, Ordering};
     use std::sync::Arc;
 
     use async_trait::async_trait;
@@ -588,13 +589,14 @@ 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;
+    use crate::local::{test_util, DynamicKeyLockGuard};
     use crate::procedure::PoisonKeys;
     use crate::store::proc_path;
     use crate::test_util::InMemoryPoisonStore;
@@ -666,6 +668,10 @@ mod tests {
             ) -> Result<()> {
                 unimplemented!()
             }
+
+            async fn acquire_lock(&self, _key: &StringKey) -> DynamicKeyLockGuard {
+                unimplemented!()
+            }
         }
 
         Context {
@@ -1674,4 +1680,66 @@ 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);
+    }
 }

src/common/procedure/src/procedure.rs

@@ -25,6 +25,7 @@ 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>;
@@ -144,6 +145,9 @@ 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].

src/common/procedure/src/rwlock.rs

@@ -18,8 +18,18 @@ 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<()> },
 }
 
@@ -36,7 +46,7 @@ impl From<OwnedRwLockWriteGuard<()>> for OwnedKeyRwLockGuard {
 }
 
 /// Locks based on a key, allowing other keys to lock independently.
-#[derive(Debug)]
+#[derive(Debug, Default)]
 pub struct KeyRwLock<K> {
     /// The inner map of locks for specific keys.
     inner: Mutex<HashMap<K, Arc<RwLock<()>>>>,

src/common/query/src/logical_plan.rs

@@ -18,16 +18,19 @@ mod udaf;
 
 use std::sync::Arc;
 
+use api::v1::TableName;
 use datafusion::catalog::CatalogProviderList;
 use datafusion::error::Result as DatafusionResult;
 use datafusion::logical_expr::{LogicalPlan, LogicalPlanBuilder};
-use datafusion_common::Column;
-use datafusion_expr::col;
+use datafusion_common::{Column, TableReference};
+use datafusion_expr::dml::InsertOp;
+use datafusion_expr::{col, DmlStatement, WriteOp};
 pub use expr::{build_filter_from_timestamp, build_same_type_ts_filter};
+use snafu::ResultExt;
 
 pub use self::accumulator::{Accumulator, AggregateFunctionCreator, AggregateFunctionCreatorRef};
 pub use self::udaf::AggregateFunction;
-use crate::error::Result;
+use crate::error::{GeneralDataFusionSnafu, Result};
 use crate::logical_plan::accumulator::*;
 use crate::signature::{Signature, Volatility};
 
@@ -79,6 +82,74 @@ pub fn rename_logical_plan_columns(
     LogicalPlanBuilder::from(plan).project(projection)?.build()
 }
 
+/// Convert a insert into logical plan to an (table_name, logical_plan)
+/// where table_name is the name of the table to insert into.
+/// logical_plan is the plan to be executed.
+///
+/// if input logical plan is not `insert into table_name <input>`, return None
+///
+/// Returned TableName will use provided catalog and schema if not specified in the logical plan,
+/// if table scan in logical plan have full table name, will **NOT** override it.
+pub fn breakup_insert_plan(
+    plan: &LogicalPlan,
+    default_catalog: &str,
+    default_schema: &str,
+) -> Option<(TableName, Arc<LogicalPlan>)> {
+    if let LogicalPlan::Dml(dml) = plan {
+        if dml.op != WriteOp::Insert(InsertOp::Append) {
+            return None;
+        }
+        let table_name = &dml.table_name;
+        let table_name = match table_name {
+            TableReference::Bare { table } => TableName {
+                catalog_name: default_catalog.to_string(),
+                schema_name: default_schema.to_string(),
+                table_name: table.to_string(),
+            },
+            TableReference::Partial { schema, table } => TableName {
+                catalog_name: default_catalog.to_string(),
+                schema_name: schema.to_string(),
+                table_name: table.to_string(),
+            },
+            TableReference::Full {
+                catalog,
+                schema,
+                table,
+            } => TableName {
+                catalog_name: catalog.to_string(),
+                schema_name: schema.to_string(),
+                table_name: table.to_string(),
+            },
+        };
+        let logical_plan = dml.input.clone();
+        Some((table_name, logical_plan))
+    } else {
+        None
+    }
+}
+
+/// create a `insert into table_name <input>` logical plan
+pub fn add_insert_to_logical_plan(
+    table_name: TableName,
+    table_schema: datafusion_common::DFSchemaRef,
+    input: LogicalPlan,
+) -> Result<LogicalPlan> {
+    let table_name = TableReference::Full {
+        catalog: table_name.catalog_name.into(),
+        schema: table_name.schema_name.into(),
+        table: table_name.table_name.into(),
+    };
+
+    let plan = LogicalPlan::Dml(DmlStatement::new(
+        table_name,
+        table_schema,
+        WriteOp::Insert(InsertOp::Append),
+        Arc::new(input),
+    ));
+    let plan = plan.recompute_schema().context(GeneralDataFusionSnafu)?;
+    Ok(plan)
+}
+
 /// The datafusion `[LogicalPlan]` decoder.
 #[async_trait::async_trait]
 pub trait SubstraitPlanDecoder {

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, Error, FromScalarValueSnafu, IntoVectorSnafu, Result};
+use crate::error::{self, FromScalarValueSnafu, IntoVectorSnafu, Result};
 use crate::prelude::*;
 
 pub type AggregateFunctionCreatorRef = Arc<dyn AggregateFunctionCreator>;
@@ -166,8 +166,7 @@ impl DfAccumulator for DfAccumulatorAdaptor {
         let output_type = self.creator.output_type()?;
         let scalar_value = value
             .try_to_scalar_value(&output_type)
-            .context(error::ToScalarValueSnafu)
-            .map_err(Error::from)?;
+            .context(error::ToScalarValueSnafu)?;
         Ok(scalar_value)
     }
 

src/common/query/src/request.rs

@@ -17,6 +17,7 @@ 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>,

src/common/wal/src/config/kafka/common.rs

@@ -30,10 +30,10 @@ pub const DEFAULT_BACKOFF_CONFIG: BackoffConfig = BackoffConfig {
     deadline: Some(Duration::from_secs(120)),
 };
 
-/// Default interval for active WAL pruning.
-pub const DEFAULT_ACTIVE_PRUNE_INTERVAL: Duration = Duration::ZERO;
-/// Default limit for concurrent active pruning tasks.
-pub const DEFAULT_ACTIVE_PRUNE_TASK_LIMIT: usize = 10;
+/// Default interval for auto WAL pruning.
+pub const DEFAULT_AUTO_PRUNE_INTERVAL: Duration = Duration::ZERO;
+/// Default limit for concurrent auto pruning tasks.
+pub const DEFAULT_AUTO_PRUNE_PARALLELISM: usize = 10;
 /// Default interval for sending flush request to regions when pruning remote WAL.
 pub const DEFAULT_TRIGGER_FLUSH_THRESHOLD: u64 = 0;
 

src/common/wal/src/config/kafka/datanode.rs

@@ -18,8 +18,8 @@ use common_base::readable_size::ReadableSize;
 use serde::{Deserialize, Serialize};
 
 use crate::config::kafka::common::{
-    KafkaConnectionConfig, KafkaTopicConfig, DEFAULT_ACTIVE_PRUNE_INTERVAL,
-    DEFAULT_ACTIVE_PRUNE_TASK_LIMIT, DEFAULT_TRIGGER_FLUSH_THRESHOLD,
+    KafkaConnectionConfig, KafkaTopicConfig, DEFAULT_AUTO_PRUNE_INTERVAL,
+    DEFAULT_AUTO_PRUNE_PARALLELISM, DEFAULT_TRIGGER_FLUSH_THRESHOLD,
 };
 
 /// Kafka wal configurations for datanode.
@@ -47,9 +47,8 @@ pub struct DatanodeKafkaConfig {
     pub dump_index_interval: Duration,
     /// Ignore missing entries during read WAL.
     pub overwrite_entry_start_id: bool,
-    // Active WAL pruning.
-    pub auto_prune_topic_records: bool,
     // Interval of WAL pruning.
+    #[serde(with = "humantime_serde")]
     pub auto_prune_interval: Duration,
     // Threshold for sending flush request when pruning remote WAL.
     // `None` stands for never sending flush request.
@@ -70,10 +69,9 @@ impl Default for DatanodeKafkaConfig {
             create_index: true,
             dump_index_interval: Duration::from_secs(60),
             overwrite_entry_start_id: false,
-            auto_prune_topic_records: false,
-            auto_prune_interval: DEFAULT_ACTIVE_PRUNE_INTERVAL,
+            auto_prune_interval: DEFAULT_AUTO_PRUNE_INTERVAL,
             trigger_flush_threshold: DEFAULT_TRIGGER_FLUSH_THRESHOLD,
-            auto_prune_parallelism: DEFAULT_ACTIVE_PRUNE_TASK_LIMIT,
+            auto_prune_parallelism: DEFAULT_AUTO_PRUNE_PARALLELISM,
         }
     }
 }