chore: bump version to 0.14.3

Signed-off-by: Zhenchi <zhongzc_arch@outlook.com>

.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

@@ -2,13 +2,14 @@ meta:
   configData: |-
     [runtime]
     global_rt_size = 4
-    
+
     [wal]
     provider = "kafka"
     broker_endpoints = ["kafka.kafka-cluster.svc.cluster.local:9092"]
     num_topics = 3
+    auto_prune_interval = "30s"
+    trigger_flush_threshold = 100
 
-        
     [datanode]
     [datanode.client]
     timeout = "120s"
@@ -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

@@ -25,7 +25,7 @@ function create_version() {
   fi
 
   # Reuse $NEXT_RELEASE_VERSION to identify whether it's a nightly build.
-  # It will be like 'nigtly-20230808-7d0d8dc6'.
+  # It will be like 'nightly-20230808-7d0d8dc6'.
   if [ "$NEXT_RELEASE_VERSION" = nightly ]; then
     echo "$NIGHTLY_RELEASE_PREFIX-$(date "+%Y%m%d")-$(git rev-parse --short HEAD)"
     exit 0
@@ -60,9 +60,9 @@ function create_version() {
 }
 
 # You can run as following examples:
-#  GITHUB_EVENT_NAME=push NEXT_RELEASE_VERSION=v0.4.0 NIGHTLY_RELEASE_PREFIX=nigtly GITHUB_REF_NAME=v0.3.0 ./create-version.sh
-#  GITHUB_EVENT_NAME=workflow_dispatch NEXT_RELEASE_VERSION=v0.4.0 NIGHTLY_RELEASE_PREFIX=nigtly ./create-version.sh
-#  GITHUB_EVENT_NAME=schedule NEXT_RELEASE_VERSION=v0.4.0 NIGHTLY_RELEASE_PREFIX=nigtly ./create-version.sh
-#  GITHUB_EVENT_NAME=schedule NEXT_RELEASE_VERSION=nightly NIGHTLY_RELEASE_PREFIX=nigtly ./create-version.sh
-#  GITHUB_EVENT_NAME=workflow_dispatch COMMIT_SHA=f0e7216c4bb6acce9b29a21ec2d683be2e3f984a NEXT_RELEASE_VERSION=dev NIGHTLY_RELEASE_PREFIX=nigtly ./create-version.sh
+#  GITHUB_EVENT_NAME=push NEXT_RELEASE_VERSION=v0.4.0 NIGHTLY_RELEASE_PREFIX=nightly GITHUB_REF_NAME=v0.3.0 ./create-version.sh
+#  GITHUB_EVENT_NAME=workflow_dispatch NEXT_RELEASE_VERSION=v0.4.0 NIGHTLY_RELEASE_PREFIX=nightly ./create-version.sh
+#  GITHUB_EVENT_NAME=schedule NEXT_RELEASE_VERSION=v0.4.0 NIGHTLY_RELEASE_PREFIX=nightly ./create-version.sh
+#  GITHUB_EVENT_NAME=schedule NEXT_RELEASE_VERSION=nightly NIGHTLY_RELEASE_PREFIX=nightly ./create-version.sh
+#  GITHUB_EVENT_NAME=workflow_dispatch COMMIT_SHA=f0e7216c4bb6acce9b29a21ec2d683be2e3f984a NEXT_RELEASE_VERSION=dev NIGHTLY_RELEASE_PREFIX=nightly ./create-version.sh
 create_version

.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/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/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

@@ -317,7 +317,7 @@ jobs:
           image-registry-username: ${{ secrets.DOCKERHUB_USERNAME }}
           image-registry-password: ${{ secrets.DOCKERHUB_TOKEN }}
           version: ${{ needs.allocate-runners.outputs.version }}
-          push-latest-tag: true
+          push-latest-tag: ${{ github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
 
       - name: Set build image result
         id: set-build-image-result
@@ -364,7 +364,7 @@ jobs:
           dev-mode: false
           upload-to-s3: true
           update-version-info: true
-          push-latest-tag: true
+          push-latest-tag: ${{ github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
 
   publish-github-release:
     name: Create GitHub release and upload artifacts

Cargo.toml

@@ -68,7 +68,7 @@ members = [
 resolver = "2"
 
 [workspace.package]
-version = "0.14.0"
+version = "0.14.3"
 edition = "2021"
 license = "Apache-2.0"
 
@@ -77,7 +77,6 @@ clippy.print_stdout = "warn"
 clippy.print_stderr = "warn"
 clippy.dbg_macro = "warn"
 clippy.implicit_clone = "warn"
-clippy.readonly_write_lock = "allow"
 rust.unknown_lints = "deny"
 rust.unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tokio_unstable)'] }
 
@@ -90,11 +89,11 @@ rust.unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tokio_unstable)'] }
 # See for more detaiils: https://github.com/rust-lang/cargo/issues/11329
 ahash = { version = "0.8", features = ["compile-time-rng"] }
 aquamarine = "0.6"
-arrow = { version = "53.0.0", features = ["prettyprint"] }
-arrow-array = { version = "53.0.0", default-features = false, features = ["chrono-tz"] }
-arrow-flight = "53.0"
-arrow-ipc = { version = "53.0.0", default-features = false, features = ["lz4", "zstd"] }
-arrow-schema = { version = "53.0", features = ["serde"] }
+arrow = { version = "54.2", features = ["prettyprint"] }
+arrow-array = { version = "54.2", default-features = false, features = ["chrono-tz"] }
+arrow-flight = "54.2"
+arrow-ipc = { version = "54.2", default-features = false, features = ["lz4", "zstd"] }
+arrow-schema = { version = "54.2", features = ["serde"] }
 async-stream = "0.3"
 async-trait = "0.1"
 # Remember to update axum-extra, axum-macros when updating axum
@@ -113,15 +112,15 @@ clap = { version = "4.4", features = ["derive"] }
 config = "0.13.0"
 crossbeam-utils = "0.8"
 dashmap = "6.1"
-datafusion = { git = "https://github.com/apache/datafusion.git", rev = "2464703c84c400a09cc59277018813f0e797bb4e" }
-datafusion-common = { git = "https://github.com/apache/datafusion.git", rev = "2464703c84c400a09cc59277018813f0e797bb4e" }
-datafusion-expr = { git = "https://github.com/apache/datafusion.git", rev = "2464703c84c400a09cc59277018813f0e797bb4e" }
-datafusion-functions = { git = "https://github.com/apache/datafusion.git", rev = "2464703c84c400a09cc59277018813f0e797bb4e" }
-datafusion-optimizer = { git = "https://github.com/apache/datafusion.git", rev = "2464703c84c400a09cc59277018813f0e797bb4e" }
-datafusion-physical-expr = { git = "https://github.com/apache/datafusion.git", rev = "2464703c84c400a09cc59277018813f0e797bb4e" }
-datafusion-physical-plan = { git = "https://github.com/apache/datafusion.git", rev = "2464703c84c400a09cc59277018813f0e797bb4e" }
-datafusion-sql = { git = "https://github.com/apache/datafusion.git", rev = "2464703c84c400a09cc59277018813f0e797bb4e" }
-datafusion-substrait = { git = "https://github.com/apache/datafusion.git", rev = "2464703c84c400a09cc59277018813f0e797bb4e" }
+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"
@@ -130,7 +129,7 @@ etcd-client = "0.14"
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "dd4a1996982534636734674db66e44464b0c0d83" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "4d4136692fe7fbbd509ebc8c902f6afcc0ce61e4" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"
@@ -148,6 +147,7 @@ moka = "0.12"
 nalgebra = "0.33"
 notify = "8.0"
 num_cpus = "1.16"
+object_store_opendal = "0.50"
 once_cell = "1.18"
 opentelemetry-proto = { version = "0.27", features = [
     "gen-tonic",
@@ -157,12 +157,14 @@ opentelemetry-proto = { version = "0.27", features = [
     "logs",
 ] }
 parking_lot = "0.12"
-parquet = { version = "53.0.0", default-features = false, features = ["arrow", "async", "object_store"] }
+parquet = { version = "54.2", default-features = false, features = ["arrow", "async", "object_store"] }
 paste = "1.0"
 pin-project = "1.0"
 prometheus = { version = "0.13.3", features = ["process"] }
-promql-parser = { version = "0.5", features = ["ser"] }
-prost = "0.13"
+promql-parser = { git = "https://github.com/GreptimeTeam/promql-parser.git", rev = "0410e8b459dda7cb222ce9596f8bf3971bd07bd2", features = [
+    "ser",
+] }
+prost = { version = "0.13", features = ["no-recursion-limit"] }
 raft-engine = { version = "0.4.1", default-features = false }
 rand = "0.9"
 ratelimit = "0.10"
@@ -191,19 +193,18 @@ simd-json = "0.15"
 similar-asserts = "1.6.0"
 smallvec = { version = "1", features = ["serde"] }
 snafu = "0.8"
+sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "0cf6c04490d59435ee965edd2078e8855bd8471e", features = [
+    "visitor",
+    "serde",
+] } # branch = "v0.54.x"
 sqlx = { version = "0.8", features = [
     "runtime-tokio-rustls",
     "mysql",
     "postgres",
     "chrono",
 ] }
-sysinfo = "0.33"
-# on branch v0.52.x
-sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "71dd86058d2af97b9925093d40c4e03360403170", features = [
-    "visitor",
-    "serde",
-] } # on branch v0.44.x
 strum = { version = "0.27", features = ["derive"] }
+sysinfo = "0.33"
 tempfile = "3"
 tokio = { version = "1.40", features = ["full"] }
 tokio-postgres = "0.7"
@@ -270,6 +271,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

@@ -32,6 +32,10 @@ ifneq ($(strip $(BUILD_JOBS)),)
 	NEXTEST_OPTS += --build-jobs=${BUILD_JOBS}
 endif
 
+ifneq ($(strip $(BUILD_JOBS)),)
+	SQLNESS_OPTS += --jobs ${BUILD_JOBS}
+endif
+
 ifneq ($(strip $(CARGO_PROFILE)),)
 	CARGO_BUILD_OPTS += --profile ${CARGO_PROFILE}
 endif
@@ -193,6 +197,7 @@ fix-clippy: ## Fix clippy violations.
 fmt-check: ## Check code format.
 	cargo fmt --all -- --check
 	python3 scripts/check-snafu.py
+	python3 scripts/check-super-imports.py
 
 .PHONY: start-etcd
 start-etcd: ## Start single node etcd for testing purpose.
@@ -217,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,7 @@
   </picture>
 </p>
 
-<h2 align="center">Unified & Cost-Effective Observerability Database for Metrics, Logs, and Events</h2>
+<h2 align="center">Real-Time & Cloud-Native Observability  Database<br/>for metrics, logs, and traces</h2>
 
 <div align="center">
 <h3 align="center">
@@ -62,7 +62,7 @@
 
 ## Introduction
 
-**GreptimeDB** is an open-source unified & cost-effective observerability database for **Metrics**, **Logs**, and **Events** (also **Traces** in plan). You can gain real-time insights from Edge to Cloud at Any Scale.
+**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.
 
 ## News
 
@@ -70,27 +70,27 @@
 
 ## Why GreptimeDB
 
-Our core developers have been building observerability data platforms for years. Based on our best practices, GreptimeDB was born to give you:
+Our core developers have been building observability data platforms for years. Based on our best practices, GreptimeDB was born to give you:
 
-* **Unified Processing of Metrics, Logs, and Events**
+* **Unified Processing of Observability Data**
 
-  GreptimeDB unifies observerability data processing by treating all data - whether metrics, logs, or events - as timestamped events with context. Users can analyze this data using either [SQL](https://docs.greptime.com/user-guide/query-data/sql) or [PromQL](https://docs.greptime.com/user-guide/query-data/promql) and leverage stream processing ([Flow](https://docs.greptime.com/user-guide/flow-computation/overview)) to enable continuous aggregation. [Read more](https://docs.greptime.com/user-guide/concepts/data-model).
+  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.
+
+* **High Performance and Cost-effective**
+
+   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.
 
 * **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.
 
-* **Performance and Cost-effective**
+* **Developer-Friendly**
 
-  Written in pure Rust for superior performance and reliability. GreptimeDB features a distributed query engine with intelligent indexing to handle high cardinality data efficiently. Its optimized columnar storage achieves 50x cost efficiency on cloud object storage through advanced compression. [Benchmark reports](https://www.greptime.com/blogs/2024-09-09-report-summary).
+  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.
 
-* **Cloud-Edge Collaboration**
+* **Flexible Deployment Options**
 
-  GreptimeDB seamlessly operates across cloud and edge (ARM/Android/Linux), providing consistent APIs and control plane for unified data management and efficient synchronization. [Learn how to run on Android](https://docs.greptime.com/user-guide/deployments/run-on-android/).
-
-* **Multi-protocol Ingestion, SQL & PromQL Ready**
-
-  Widely adopted database protocols and APIs, including MySQL, PostgreSQL, InfluxDB, OpenTelemetry, Loki and Prometheus, etc.  Effortless Adoption & Seamless Migration. [Supported Protocols Overview](https://docs.greptime.com/user-guide/protocols/overview).
+  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).
 
@@ -233,3 +233,5 @@ Special thanks to all the contributors who have propelled GreptimeDB forward. Fo
 - 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"/>

config/config.md

@@ -96,6 +96,8 @@
 | `procedure.max_running_procedures` | Integer | `128` | Max running procedures.<br/>The maximum number of procedures that can be running at the same time.<br/>If the number of running procedures exceeds this limit, the procedure will be rejected. |
 | `flow` | -- | -- | flow engine options. |
 | `flow.num_workers` | Integer | `0` | The number of flow worker in flownode.<br/>Not setting(or set to 0) this value will use the number of CPU cores divided by 2. |
+| `query` | -- | -- | The query engine options. |
+| `query.parallelism` | Integer | `0` | Parallelism of the query engine.<br/>Default to 0, which means the number of CPU cores. |
 | `storage` | -- | -- | The data storage options. |
 | `storage.data_home` | String | `./greptimedb_data/` | The working home directory. |
 | `storage.type` | String | `File` | The storage type used to store the data.<br/>- `File`: the data is stored in the local file system.<br/>- `S3`: the data is stored in the S3 object storage.<br/>- `Gcs`: the data is stored in the Google Cloud Storage.<br/>- `Azblob`: the data is stored in the Azure Blob Storage.<br/>- `Oss`: the data is stored in the Aliyun OSS. |
@@ -270,6 +272,8 @@
 | `meta_client.metadata_cache_max_capacity` | Integer | `100000` | The configuration about the cache of the metadata. |
 | `meta_client.metadata_cache_ttl` | String | `10m` | TTL of the metadata cache. |
 | `meta_client.metadata_cache_tti` | String | `5m` | -- |
+| `query` | -- | -- | The query engine options. |
+| `query.parallelism` | Integer | `0` | Parallelism of the query engine.<br/>Default to 0, which means the number of CPU cores. |
 | `datanode` | -- | -- | Datanode options. |
 | `datanode.client` | -- | -- | Datanode client options. |
 | `datanode.client.connect_timeout` | String | `10s` | -- |
@@ -315,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. |
@@ -339,6 +344,9 @@
 | `wal.provider` | String | `raft_engine` | -- |
 | `wal.broker_endpoints` | Array | -- | The broker endpoints of the Kafka cluster. |
 | `wal.auto_create_topics` | Bool | `true` | Automatically create topics for WAL.<br/>Set to `true` to automatically create topics for WAL.<br/>Otherwise, use topics named `topic_name_prefix_[0..num_topics)` |
+| `wal.auto_prune_interval` | String | `0s` | Interval of automatically WAL pruning.<br/>Set to `0s` to disable automatically WAL pruning which delete unused remote WAL entries periodically. |
+| `wal.trigger_flush_threshold` | Integer | `0` | The threshold to trigger a flush operation of a region in automatically WAL pruning.<br/>Metasrv will send a flush request to flush the region when:<br/>`trigger_flush_threshold` + `prunable_entry_id` < `max_prunable_entry_id`<br/>where:<br/>- `prunable_entry_id` is the maximum entry id that can be pruned of the region.<br/>- `max_prunable_entry_id` is the maximum prunable entry id among all regions in the same topic.<br/>Set to `0` to disable the flush operation. |
+| `wal.auto_prune_parallelism` | Integer | `10` | Concurrent task limit for automatically WAL pruning. |
 | `wal.num_topics` | Integer | `64` | Number of topics. |
 | `wal.selector_type` | String | `round_robin` | Topic selector type.<br/>Available selector types:<br/>- `round_robin` (default) |
 | `wal.topic_name_prefix` | String | `greptimedb_wal_topic` | A Kafka topic is constructed by concatenating `topic_name_prefix` and `topic_id`.<br/>Only accepts strings that match the following regular expression pattern:<br/>[a-zA-Z_:-][a-zA-Z0-9_:\-\.@#]*<br/>i.g., greptimedb_wal_topic_0, greptimedb_wal_topic_1. |
@@ -429,6 +437,8 @@
 | `wal.create_index` | Bool | `true` | Whether to enable WAL index creation.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.dump_index_interval` | String | `60s` | The interval for dumping WAL indexes.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.overwrite_entry_start_id` | Bool | `false` | Ignore missing entries during read WAL.<br/>**It's only used when the provider is `kafka`**.<br/><br/>This option ensures that when Kafka messages are deleted, the system<br/>can still successfully replay memtable data without throwing an<br/>out-of-range error.<br/>However, enabling this option might lead to unexpected data loss,<br/>as the system will skip over missing entries instead of treating<br/>them as critical errors. |
+| `query` | -- | -- | The query engine options. |
+| `query.parallelism` | Integer | `0` | Parallelism of the query engine.<br/>Default to 0, which means the number of CPU cores. |
 | `storage` | -- | -- | The data storage options. |
 | `storage.data_home` | String | `./greptimedb_data/` | The working home directory. |
 | `storage.type` | String | `File` | The storage type used to store the data.<br/>- `File`: the data is stored in the local file system.<br/>- `S3`: the data is stored in the S3 object storage.<br/>- `Gcs`: the data is stored in the Google Cloud Storage.<br/>- `Azblob`: the data is stored in the Azure Blob Storage.<br/>- `Oss`: the data is stored in the Aliyun OSS. |

config/datanode.example.toml

@@ -243,6 +243,12 @@ overwrite_entry_start_id = false
 # credential = "base64-credential"
 # endpoint = "https://storage.googleapis.com"
 
+## The query engine options.
+[query]
+## Parallelism of the query engine.
+## Default to 0, which means the number of CPU cores.
+parallelism = 0
+
 ## The data storage options.
 [storage]
 ## The working home directory.

config/frontend.example.toml

@@ -179,6 +179,12 @@ metadata_cache_ttl = "10m"
 # TTI of the metadata cache.
 metadata_cache_tti = "5m"
 
+## The query engine options.
+[query]
+## Parallelism of the query engine.
+## Default to 0, which means the number of CPU cores.
+parallelism = 0
+
 ## Datanode options.
 [datanode]
 ## Datanode client 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"
 
@@ -130,6 +134,22 @@ broker_endpoints = ["127.0.0.1:9092"]
 ## Otherwise, use topics named `topic_name_prefix_[0..num_topics)`
 auto_create_topics = true
 
+## Interval of automatically WAL pruning.
+## Set to `0s` to disable automatically WAL pruning which delete unused remote WAL entries periodically.
+auto_prune_interval = "0s"
+
+## The threshold to trigger a flush operation of a region in automatically WAL pruning.
+## Metasrv will send a flush request to flush the region when:
+## `trigger_flush_threshold` + `prunable_entry_id` < `max_prunable_entry_id`
+## where:
+## - `prunable_entry_id` is the maximum entry id that can be pruned of the region.
+## - `max_prunable_entry_id` is the maximum prunable entry id among all regions in the same topic.
+## Set to `0` to disable the flush operation.
+trigger_flush_threshold = 0
+
+## Concurrent task limit for automatically WAL pruning.
+auto_prune_parallelism = 10
+
 ## Number of topics.
 num_topics = 64
 

config/standalone.example.toml

@@ -334,6 +334,12 @@ max_running_procedures = 128
 # credential = "base64-credential"
 # endpoint = "https://storage.googleapis.com"
 
+## The query engine options.
+[query]
+## Parallelism of the query engine.
+## Default to 0, which means the number of CPU cores.
+parallelism = 0
+
 ## The data storage options.
 [storage]
 ## The working home directory.

docs/how-to/how-to-profile-memory.md

@@ -1,6 +1,6 @@
 # Profile memory usage of GreptimeDB
 
-This crate provides an easy approach to dump memory profiling info.
+This crate provides an easy approach to dump memory profiling info. A set of ready to use scripts is provided in [docs/how-to/memory-profile-scripts](docs/how-to/memory-profile-scripts).
 
 ## Prerequisites
 ### jemalloc

docs/how-to/memory-profile-scripts/scripts/README.md

@@ -0,0 +1,52 @@
+# Memory Analysis Process
+This section will guide you through the process of analyzing memory usage for greptimedb.
+
+1. Get the `jeprof` tool script, see the next section("Getting the `jeprof` tool") for details.
+
+2. After starting `greptimedb`(with env var `MALLOC_CONF=prof:true`), execute the `dump.sh` script with the PID of the `greptimedb` process as an argument. This continuously monitors memory usage and captures profiles when exceeding thresholds (e.g. +20MB within 10 minutes). Outputs `greptime-{timestamp}.gprof` files.
+
+3. With 2-3 gprof files, run `gen_flamegraph.sh` in the same environment to generate flame graphs showing memory allocation call stacks.
+
+4.  **NOTE:** The `gen_flamegraph.sh` script requires `jeprof` and optionally `flamegraph.pl` to be in the current directory. If needed to gen flamegraph now, run the `get_flamegraph_tool.sh` script, which downloads the flame graph generation tool `flamegraph.pl` to the current directory.
+    The usage of `gen_flamegraph.sh` is:
+
+    `Usage: ./gen_flamegraph.sh <binary_path> <gprof_directory>`
+    where `<binary_path>` is the path to the greptimedb binary, `<gprof_directory>` is the directory containing the gprof files(the directory `dump.sh` is dumping profiles to).
+    Example call: `./gen_flamegraph.sh ./greptime .`
+
+    Generating the flame graph might take a few minutes. The generated flame graphs are located in the `<gprof_directory>/flamegraphs` directory. Or if no `flamegraph.pl` is found, it will only contain `.collapse` files which is also fine.
+5.  You can send the generated flame graphs(the entire folder of `<gprof_directory>/flamegraphs`) to developers for further analysis.
+
+
+## Getting the `jeprof` tool
+there are three ways to get `jeprof`, list in here from simple to complex, using any one of those methods is ok, as long as it's the same environment as the `greptimedb` will be running on:
+1. If you are compiling greptimedb from source, then `jeprof` is already produced during compilation. After running `cargo build`, execute `find_compiled_jeprof.sh`. This will copy `jeprof` to the current directory.
+2. Or, if you have the Rust toolchain installed locally, simply follow these commands:
+```bash
+cargo new get_jeprof
+cd get_jeprof
+```
+Then add this line to `Cargo.toml`:
+```toml
+[dependencies]
+tikv-jemalloc-ctl = { version = "0.6", features = ["use_std", "stats"] }
+```
+then run:
+```bash
+cargo build
+```
+after that the `jeprof` tool is produced. Now run `find_compiled_jeprof.sh` in current directory, it will copy the `jeprof` tool to the current directory.
+
+3. compile jemalloc from source
+you can first clone this repo, and checkout to this commit:
+```bash
+git clone https://github.com/tikv/jemalloc.git
+cd jemalloc
+git checkout e13ca993e8ccb9ba9847cc330696e02839f328f7
+```
+then run:
+```bash
+./configure
+make
+```
+and `jeprof` is in `.bin/` directory. Copy it to the current directory.

docs/how-to/memory-profile-scripts/scripts/dump.sh

@@ -0,0 +1,78 @@
+#!/bin/bash
+
+# Monitors greptime process memory usage every 10 minutes
+# Triggers memory profile capture via `curl -X POST localhost:4000/debug/prof/mem > greptime-{timestamp}.gprof`
+# when memory increases by more than 20MB since last check
+# Generated profiles can be analyzed using flame graphs as described in `how-to-profile-memory.md`
+# (jeprof is compiled with the database - see documentation)
+# Alternative: Share binaries + profiles for analysis (Docker images preferred)
+
+# Threshold in Kilobytes (20 MB)
+threshold_kb=$((20 * 1024))
+sleep_interval=$((10 * 60))
+
+# Variable to store the last measured memory usage in KB
+last_mem_kb=0
+
+echo "Starting memory monitoring for 'greptime' process..."
+
+while true; do
+
+    # Check if PID is provided as an argument
+    if [ -z "$1" ]; then
+        echo "$(date): PID must be provided as a command-line argument."
+        exit 1
+    fi
+
+    pid="$1"
+
+    # Validate that the PID is a number
+    if ! [[ "$pid" =~ ^[0-9]+$ ]]; then
+        echo "$(date): Invalid PID: '$pid'. PID must be a number."
+        exit 1
+    fi
+
+    # Get the current Resident Set Size (RSS) in Kilobytes
+    current_mem_kb=$(ps -o rss= -p "$pid")
+
+    # Check if ps command was successful and returned a number
+    if ! [[ "$current_mem_kb" =~ ^[0-9]+$ ]]; then
+        echo "$(date): Failed to get memory usage for PID $pid. Skipping check."
+        # Keep last_mem_kb to avoid false positives if the process briefly becomes unreadable.
+        continue
+    fi
+
+    echo "$(date): Current memory usage for PID $pid: ${current_mem_kb} KB"
+
+    # Compare with the last measurement
+    # if it's the first run, also do a baseline dump just to make sure we can dump
+    
+    diff_kb=$((current_mem_kb - last_mem_kb))
+    echo "$(date): Memory usage change since last check: ${diff_kb} KB"
+
+    if [ "$diff_kb" -gt "$threshold_kb" ]; then
+        echo "$(date): Memory increase (${diff_kb} KB) exceeded threshold (${threshold_kb} KB). Dumping profile..."
+        timestamp=$(date +%Y%m%d%H%M%S)
+        profile_file="greptime-${timestamp}.gprof"
+        # Execute curl and capture output to file
+        if curl -sf -X POST localhost:4000/debug/prof/mem > "$profile_file"; then
+            echo "$(date): Memory profile saved to $profile_file"
+        else
+            echo "$(date): Failed to dump memory profile (curl exit code: $?)."
+            # Remove the potentially empty/failed profile file
+            rm -f "$profile_file"
+        fi
+    else
+            echo "$(date): Memory increase (${diff_kb} KB) is within the threshold (${threshold_kb} KB)."
+    fi
+    
+
+    # Update the last memory usage
+    last_mem_kb=$current_mem_kb
+    
+    # Wait for 5 minutes
+    echo "$(date): Sleeping for $sleep_interval seconds..."
+    sleep $sleep_interval
+done
+
+echo "Memory monitoring script stopped." # This line might not be reached in normal operation

docs/how-to/memory-profile-scripts/scripts/find_compiled_jeprof.sh

@@ -0,0 +1,15 @@
+#!/bin/bash
+
+# Locates compiled jeprof binary (memory analysis tool) after cargo build
+# Copies it to current directory from target/ build directories
+
+JPROF_PATH=$(find . -name 'jeprof' -print -quit)
+if [ -n "$JPROF_PATH" ]; then
+  echo "Found jeprof at $JPROF_PATH"
+  cp "$JPROF_PATH" .
+  chmod +x jeprof
+  echo "Copied jeprof to current directory and made it executable."
+else
+  echo "jeprof not found"
+  exit 1
+fi

docs/how-to/memory-profile-scripts/scripts/gen_flamegraph.sh

@@ -0,0 +1,89 @@
+#!/bin/bash
+
+# Generate flame graphs from a series of `.gprof` files
+# First argument: Path to the binary executable
+# Second argument: Path to directory containing gprof files
+# Requires `jeprof` and `flamegraph.pl` in current directory
+# What this script essentially does is:
+# ./jeprof <binary> <gprof> --collapse | ./flamegraph.pl > <output>
+# For differential analysis between consecutive profiles:
+# ./jeprof <binary> --base <gprof1> <gprof2> --collapse | ./flamegraph.pl > <output_diff>
+
+set -e # Exit immediately if a command exits with a non-zero status.
+
+# Check for required tools
+if [ ! -f "./jeprof" ]; then
+    echo "Error: jeprof not found in the current directory."
+    exit 1
+fi
+
+if [ ! -f "./flamegraph.pl" ]; then
+    echo "Error: flamegraph.pl not found in the current directory."
+    exit 1
+fi
+
+# Check arguments
+if [ "$#" -ne 2 ]; then
+    echo "Usage: $0 <binary_path> <gprof_directory>"
+    exit 1
+fi
+
+BINARY_PATH=$1
+GPROF_DIR=$2
+OUTPUT_DIR="${GPROF_DIR}/flamegraphs" # Store outputs in a subdirectory
+
+if [ ! -f "$BINARY_PATH" ]; then
+    echo "Error: Binary file not found at $BINARY_PATH"
+    exit 1
+fi
+
+if [ ! -d "$GPROF_DIR" ]; then
+    echo "Error: gprof directory not found at $GPROF_DIR"
+    exit 1
+fi
+
+mkdir -p "$OUTPUT_DIR"
+echo "Generating flamegraphs in $OUTPUT_DIR"
+
+# Find and sort gprof files
+# Use find + sort -V for natural sort of version numbers if present in filenames
+# Use null-terminated strings for safety with find/xargs/sort
+mapfile -d $'\0' gprof_files < <(find "$GPROF_DIR" -maxdepth 1 -name '*.gprof' -print0 | sort -zV)
+
+if [ ${#gprof_files[@]} -eq 0 ]; then
+    echo "No .gprof files found in $GPROF_DIR"
+    exit 0
+fi
+
+prev_gprof=""
+
+# Generate flamegraphs
+for gprof_file in "${gprof_files[@]}"; do
+    # Skip empty entries if any
+    if [ -z "$gprof_file" ]; then
+        continue
+    fi
+
+    filename=$(basename "$gprof_file" .gprof)
+    output_collapse="${OUTPUT_DIR}/${filename}.collapse"
+    output_svg="${OUTPUT_DIR}/${filename}.svg"
+    echo "Generating collapse file for $gprof_file -> $output_collapse"
+    ./jeprof "$BINARY_PATH" "$gprof_file" --collapse > "$output_collapse"
+    echo "Generating flamegraph for $gprof_file -> $output_svg"
+    ./flamegraph.pl "$output_collapse" > "$output_svg" || true
+
+    # Generate diff flamegraph if not the first file
+    if [ -n "$prev_gprof" ]; then
+        prev_filename=$(basename "$prev_gprof" .gprof)
+        diff_output_collapse="${OUTPUT_DIR}/${prev_filename}_vs_${filename}_diff.collapse"
+        diff_output_svg="${OUTPUT_DIR}/${prev_filename}_vs_${filename}_diff.svg"
+        echo "Generating diff collapse file for $prev_gprof vs $gprof_file -> $diff_output_collapse"
+        ./jeprof "$BINARY_PATH" --base "$prev_gprof" "$gprof_file" --collapse > "$diff_output_collapse"
+        echo "Generating diff flamegraph for $prev_gprof vs $gprof_file -> $diff_output_svg"
+        ./flamegraph.pl "$diff_output_collapse" > "$diff_output_svg" || true
+    fi
+
+    prev_gprof="$gprof_file"
+done
+
+echo "Flamegraph generation complete."

docs/how-to/memory-profile-scripts/scripts/gen_from_collapse.sh

@@ -0,0 +1,44 @@
+#!/bin/bash
+
+# Generate flame graphs from .collapse files
+# Argument: Path to directory containing collapse files
+# Requires `flamegraph.pl` in current directory
+
+# Check if flamegraph.pl exists
+if [ ! -f "./flamegraph.pl" ]; then
+    echo "Error: flamegraph.pl not found in the current directory."
+    exit 1
+fi
+
+# Check if directory argument is provided
+if [ -z "$1" ]; then
+    echo "Usage: $0 <collapse_directory>"
+    exit 1
+fi
+
+COLLAPSE_DIR=$1
+
+# Check if the provided argument is a directory
+if [ ! -d "$COLLAPSE_DIR" ]; then
+    echo "Error: '$COLLAPSE_DIR' is not a valid directory."
+    exit 1
+fi
+
+echo "Generating flame graphs from collapse files in '$COLLAPSE_DIR'..."
+
+# Find and process each .collapse file
+find "$COLLAPSE_DIR" -maxdepth 1 -name "*.collapse" -print0 | while IFS= read -r -d $'\0' collapse_file; do
+    if [ -f "$collapse_file" ]; then
+        # Construct the output SVG filename
+        svg_file="${collapse_file%.collapse}.svg"
+        echo "Generating $svg_file from $collapse_file..."
+        ./flamegraph.pl "$collapse_file" > "$svg_file"
+        if [ $? -ne 0 ]; then
+            echo "Error generating flame graph for $collapse_file"
+        else
+            echo "Successfully generated $svg_file"
+        fi
+    fi
+done
+
+echo "Flame graph generation complete."

docs/how-to/memory-profile-scripts/scripts/get_flamegraph_tool.sh

@@ -0,0 +1,6 @@
+#!/bin/bash
+
+# Download flamegraph.pl to current directory - this is the flame graph generation tool script
+
+curl https://raw.githubusercontent.com/brendangregg/FlameGraph/master/flamegraph.pl > ./flamegraph.pl
+chmod +x ./flamegraph.pl

grafana/README.md

@@ -1,61 +1,89 @@
-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 maintains the Grafana dashboards for GreptimeDB. It has two types of dashboards:
 
-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:
+- `cluster/dashboard.json`: The Grafana dashboard for the GreptimeDB cluster. Read the [dashboard.md](./dashboards/cluster/dashboard.md) for more details.
+- `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/standalone/dashboard.md) for more details.
+
+As the rapid development of GreptimeDB, the metrics may be changed, and please feel free to submit your feedback and/or contribution to this dashboard 🤗
+
+**NOTE**: 
+
+- The Grafana version should be greater than 9.0.
+
+- If you want to modify the dashboards, you only need to modify the `cluster/dashboard.json` and run the `make dashboards` command to generate the `standalone/dashboard.json` and other related files.
+
+To maintain the dashboards easily, we use the [`dac`](https://github.com/zyy17/dac) tool to generate the intermediate dashboards and markdown documents:
+
+- `cluster/dashboard.yaml`: The intermediate dashboard for the GreptimeDB cluster.
+- `standalone/dashboard.yaml`: The intermediate dashboard for the standalone GreptimeDB instance.
+
+## Data Sources
+
+There are two data sources for the dashboards to fetch the metrics:
+
+- **Prometheus**: Expose the metrics of GreptimeDB.
+- **Information Schema**: It is the MySQL port of the current monitored instance. The `overview` dashboard will use this datasource to show the information schema of the current instance.
+
+## Instance Filters
+
+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
+
+### Helm
+
+If you use the Helm [chart](https://github.com/GreptimeTeam/helm-charts) to deploy a GreptimeDB cluster, you can enable self-monitoring by setting the following values in your Helm chart:
 
 - `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 `cluster/dashboard.json` dashboard.
+   - **Standalone**: Import the `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/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/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/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/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}
+
+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)" ]]; 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/cluster}
+STANDALONE_DASHBOARD_DIR=${2:-grafana/dashboards/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>")) |"
-'

scripts/check-super-imports.py

@@ -0,0 +1,74 @@
+# 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 os
+import re
+from multiprocessing import Pool
+
+
+def find_rust_files(directory):
+    rust_files = []
+    for root, _, files in os.walk(directory):
+        # Skip files with "test" in the path
+        if "test" in root.lower():
+            continue
+
+        for file in files:
+            # Skip files with "test" in the filename
+            if "test" in file.lower():
+                continue
+
+            if file.endswith(".rs"):
+                rust_files.append(os.path.join(root, file))
+    return rust_files
+
+
+def check_file_for_super_import(file_path):
+    with open(file_path, "r") as file:
+        lines = file.readlines()
+
+    violations = []
+    for line_number, line in enumerate(lines, 1):
+        # Check for "use super::" without leading tab
+        if line.startswith("use super::"):
+            violations.append((line_number, line.strip()))
+
+    if violations:
+        return file_path, violations
+    return None
+
+
+def main():
+    rust_files = find_rust_files(".")
+
+    with Pool() as pool:
+        results = pool.map(check_file_for_super_import, rust_files)
+
+    # Filter out None results
+    violations = [result for result in results if result]
+
+    if violations:
+        print("Found 'use super::' without leading tab in the following files:")
+        counter = 1
+        for file_path, file_violations in violations:
+            for line_number, line in file_violations:
+                print(f"{counter:>5} {file_path}:{line_number} - {line}")
+                counter += 1
+        raise SystemExit(1)
+    else:
+        print("No 'use super::' without leading tab found. All files are compliant.")
+
+
+if __name__ == "__main__":
+    main()

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",
     }
 }

src/catalog/src/system_schema/information_schema.rs

@@ -49,7 +49,6 @@ pub use table_names::*;
 use views::InformationSchemaViews;
 
 use self::columns::InformationSchemaColumns;
-use super::{SystemSchemaProviderInner, SystemTable, SystemTableRef};
 use crate::error::{Error, Result};
 use crate::system_schema::information_schema::cluster_info::InformationSchemaClusterInfo;
 use crate::system_schema::information_schema::flows::InformationSchemaFlows;
@@ -63,7 +62,9 @@ use crate::system_schema::information_schema::table_constraints::InformationSche
 use crate::system_schema::information_schema::tables::InformationSchemaTables;
 use crate::system_schema::memory_table::MemoryTable;
 pub(crate) use crate::system_schema::predicate::Predicates;
-use crate::system_schema::SystemSchemaProvider;
+use crate::system_schema::{
+    SystemSchemaProvider, SystemSchemaProviderInner, SystemTable, SystemTableRef,
+};
 use crate::CatalogManager;
 
 lazy_static! {

src/catalog/src/system_schema/information_schema/cluster_info.rs

@@ -36,9 +36,8 @@ use datatypes::vectors::{
 use snafu::ResultExt;
 use store_api::storage::{ScanRequest, TableId};
 
-use super::CLUSTER_INFO;
 use crate::error::{CreateRecordBatchSnafu, InternalSnafu, Result};
-use crate::system_schema::information_schema::{InformationTable, Predicates};
+use crate::system_schema::information_schema::{InformationTable, Predicates, CLUSTER_INFO};
 use crate::system_schema::utils;
 use crate::CatalogManager;
 

src/catalog/src/system_schema/information_schema/columns.rs

@@ -38,11 +38,11 @@ use snafu::{OptionExt, ResultExt};
 use sql::statements;
 use store_api::storage::{ScanRequest, TableId};
 
-use super::{InformationTable, COLUMNS};
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::information_schema::Predicates;
+use crate::system_schema::information_schema::{InformationTable, COLUMNS};
 use crate::CatalogManager;
 
 #[derive(Debug)]

src/catalog/src/system_schema/information_schema/information_memory_table.rs

@@ -18,7 +18,7 @@ use common_catalog::consts::{METRIC_ENGINE, MITO_ENGINE};
 use datatypes::schema::{Schema, SchemaRef};
 use datatypes::vectors::{Int64Vector, StringVector, VectorRef};
 
-use super::table_names::*;
+use crate::system_schema::information_schema::table_names::*;
 use crate::system_schema::utils::tables::{
     bigint_column, string_column, string_columns, timestamp_micro_column,
 };

src/catalog/src/system_schema/information_schema/key_column_usage.rs

@@ -24,18 +24,17 @@ use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatch
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
 use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datatypes::prelude::{ConcreteDataType, MutableVector, ScalarVectorBuilder, VectorRef};
-use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
+use datatypes::schema::{ColumnSchema, FulltextBackend, Schema, SchemaRef};
 use datatypes::value::Value;
 use datatypes::vectors::{ConstantVector, StringVector, StringVectorBuilder, UInt32VectorBuilder};
 use futures_util::TryStreamExt;
 use snafu::{OptionExt, ResultExt};
 use store_api::storage::{ScanRequest, TableId};
 
-use super::KEY_COLUMN_USAGE;
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
-use crate::system_schema::information_schema::{InformationTable, Predicates};
+use crate::system_schema::information_schema::{InformationTable, Predicates, KEY_COLUMN_USAGE};
 use crate::CatalogManager;
 
 pub const CONSTRAINT_SCHEMA: &str = "constraint_schema";
@@ -48,20 +47,38 @@ pub const TABLE_SCHEMA: &str = "table_schema";
 pub const TABLE_NAME: &str = "table_name";
 pub const COLUMN_NAME: &str = "column_name";
 pub const ORDINAL_POSITION: &str = "ordinal_position";
+/// The type of the index.
+pub const GREPTIME_INDEX_TYPE: &str = "greptime_index_type";
 const INIT_CAPACITY: usize = 42;
 
-/// Primary key constraint name
-pub(crate) const PRI_CONSTRAINT_NAME: &str = "PRIMARY";
 /// Time index constraint name
-pub(crate) const TIME_INDEX_CONSTRAINT_NAME: &str = "TIME INDEX";
+pub(crate) const CONSTRAINT_NAME_TIME_INDEX: &str = "TIME INDEX";
+
+/// Primary key constraint name
+pub(crate) const CONSTRAINT_NAME_PRI: &str = "PRIMARY";
+/// Primary key index type
+pub(crate) const INDEX_TYPE_PRI: &str = "greptime-primary-key-v1";
+
 /// Inverted index constraint name
-pub(crate) const INVERTED_INDEX_CONSTRAINT_NAME: &str = "INVERTED INDEX";
+pub(crate) const CONSTRAINT_NAME_INVERTED_INDEX: &str = "INVERTED INDEX";
+/// Inverted index type
+pub(crate) const INDEX_TYPE_INVERTED_INDEX: &str = "greptime-inverted-index-v1";
+
 /// Fulltext index constraint name
-pub(crate) const FULLTEXT_INDEX_CONSTRAINT_NAME: &str = "FULLTEXT INDEX";
+pub(crate) const CONSTRAINT_NAME_FULLTEXT_INDEX: &str = "FULLTEXT INDEX";
+/// Fulltext index v1 type
+pub(crate) const INDEX_TYPE_FULLTEXT_TANTIVY: &str = "greptime-fulltext-index-v1";
+/// Fulltext index bloom type
+pub(crate) const INDEX_TYPE_FULLTEXT_BLOOM: &str = "greptime-fulltext-index-bloom";
+
 /// Skipping index constraint name
-pub(crate) const SKIPPING_INDEX_CONSTRAINT_NAME: &str = "SKIPPING INDEX";
+pub(crate) const CONSTRAINT_NAME_SKIPPING_INDEX: &str = "SKIPPING INDEX";
+/// Skipping index type
+pub(crate) const INDEX_TYPE_SKIPPING_INDEX: &str = "greptime-bloom-filter-v1";
 
 /// The virtual table implementation for `information_schema.KEY_COLUMN_USAGE`.
+///
+/// Provides an extra column `greptime_index_type` for the index type of the key column.
 #[derive(Debug)]
 pub(super) struct InformationSchemaKeyColumnUsage {
     schema: SchemaRef,
@@ -121,6 +138,11 @@ impl InformationSchemaKeyColumnUsage {
                 ConcreteDataType::string_datatype(),
                 true,
             ),
+            ColumnSchema::new(
+                GREPTIME_INDEX_TYPE,
+                ConcreteDataType::string_datatype(),
+                true,
+            ),
         ]))
     }
 
@@ -185,6 +207,7 @@ struct InformationSchemaKeyColumnUsageBuilder {
     column_name: StringVectorBuilder,
     ordinal_position: UInt32VectorBuilder,
     position_in_unique_constraint: UInt32VectorBuilder,
+    greptime_index_type: StringVectorBuilder,
 }
 
 impl InformationSchemaKeyColumnUsageBuilder {
@@ -207,6 +230,7 @@ impl InformationSchemaKeyColumnUsageBuilder {
             column_name: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             ordinal_position: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
             position_in_unique_constraint: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
+            greptime_index_type: StringVectorBuilder::with_capacity(INIT_CAPACITY),
         }
     }
 
@@ -230,34 +254,47 @@ impl InformationSchemaKeyColumnUsageBuilder {
 
                 for (idx, column) in schema.column_schemas().iter().enumerate() {
                     let mut constraints = vec![];
+                    let mut greptime_index_type = vec![];
                     if column.is_time_index() {
                         self.add_key_column_usage(
                             &predicates,
                             &schema_name,
-                            TIME_INDEX_CONSTRAINT_NAME,
+                            CONSTRAINT_NAME_TIME_INDEX,
                             &catalog_name,
                             &schema_name,
                             table_name,
                             &column.name,
                             1, //always 1 for time index
+                            "",
                         );
                     }
                     // TODO(dimbtp): foreign key constraint not supported yet
                     if keys.contains(&idx) {
-                        constraints.push(PRI_CONSTRAINT_NAME);
+                        constraints.push(CONSTRAINT_NAME_PRI);
+                        greptime_index_type.push(INDEX_TYPE_PRI);
                     }
                     if column.is_inverted_indexed() {
-                        constraints.push(INVERTED_INDEX_CONSTRAINT_NAME);
+                        constraints.push(CONSTRAINT_NAME_INVERTED_INDEX);
+                        greptime_index_type.push(INDEX_TYPE_INVERTED_INDEX);
                     }
-                    if column.is_fulltext_indexed() {
-                        constraints.push(FULLTEXT_INDEX_CONSTRAINT_NAME);
+                    if let Ok(Some(options)) = column.fulltext_options() {
+                        if options.enable {
+                            constraints.push(CONSTRAINT_NAME_FULLTEXT_INDEX);
+                            let index_type = match options.backend {
+                                FulltextBackend::Bloom => INDEX_TYPE_FULLTEXT_BLOOM,
+                                FulltextBackend::Tantivy => INDEX_TYPE_FULLTEXT_TANTIVY,
+                            };
+                            greptime_index_type.push(index_type);
+                        }
                     }
                     if column.is_skipping_indexed() {
-                        constraints.push(SKIPPING_INDEX_CONSTRAINT_NAME);
+                        constraints.push(CONSTRAINT_NAME_SKIPPING_INDEX);
+                        greptime_index_type.push(INDEX_TYPE_SKIPPING_INDEX);
                     }
 
                     if !constraints.is_empty() {
                         let aggregated_constraints = constraints.join(", ");
+                        let aggregated_index_types = greptime_index_type.join(", ");
                         self.add_key_column_usage(
                             &predicates,
                             &schema_name,
@@ -267,6 +304,7 @@ impl InformationSchemaKeyColumnUsageBuilder {
                             table_name,
                             &column.name,
                             idx as u32 + 1,
+                            &aggregated_index_types,
                         );
                     }
                 }
@@ -289,6 +327,7 @@ impl InformationSchemaKeyColumnUsageBuilder {
         table_name: &str,
         column_name: &str,
         ordinal_position: u32,
+        index_types: &str,
     ) {
         let row = [
             (CONSTRAINT_SCHEMA, &Value::from(constraint_schema)),
@@ -298,6 +337,7 @@ impl InformationSchemaKeyColumnUsageBuilder {
             (TABLE_NAME, &Value::from(table_name)),
             (COLUMN_NAME, &Value::from(column_name)),
             (ORDINAL_POSITION, &Value::from(ordinal_position)),
+            (GREPTIME_INDEX_TYPE, &Value::from(index_types)),
         ];
 
         if !predicates.eval(&row) {
@@ -314,6 +354,7 @@ impl InformationSchemaKeyColumnUsageBuilder {
         self.column_name.push(Some(column_name));
         self.ordinal_position.push(Some(ordinal_position));
         self.position_in_unique_constraint.push(None);
+        self.greptime_index_type.push(Some(index_types));
     }
 
     fn finish(&mut self) -> Result<RecordBatch> {
@@ -337,6 +378,7 @@ impl InformationSchemaKeyColumnUsageBuilder {
             null_string_vector.clone(),
             null_string_vector.clone(),
             null_string_vector,
+            Arc::new(self.greptime_index_type.finish()),
         ];
         RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
     }

src/catalog/src/system_schema/information_schema/partitions.rs

@@ -39,13 +39,12 @@ use snafu::{OptionExt, ResultExt};
 use store_api::storage::{ScanRequest, TableId};
 use table::metadata::{TableInfo, TableType};
 
-use super::PARTITIONS;
 use crate::error::{
     CreateRecordBatchSnafu, FindPartitionsSnafu, InternalSnafu, PartitionManagerNotFoundSnafu,
     Result, UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::kvbackend::KvBackendCatalogManager;
-use crate::system_schema::information_schema::{InformationTable, Predicates};
+use crate::system_schema::information_schema::{InformationTable, Predicates, PARTITIONS};
 use crate::CatalogManager;
 
 const TABLE_CATALOG: &str = "table_catalog";

src/catalog/src/system_schema/information_schema/procedure_info.rs

@@ -33,9 +33,8 @@ use datatypes::vectors::{StringVectorBuilder, TimestampMillisecondVectorBuilder}
 use snafu::ResultExt;
 use store_api::storage::{ScanRequest, TableId};
 
-use super::PROCEDURE_INFO;
 use crate::error::{CreateRecordBatchSnafu, InternalSnafu, Result};
-use crate::system_schema::information_schema::{InformationTable, Predicates};
+use crate::system_schema::information_schema::{InformationTable, Predicates, PROCEDURE_INFO};
 use crate::system_schema::utils;
 use crate::CatalogManager;
 

src/catalog/src/system_schema/information_schema/region_peers.rs

@@ -35,13 +35,12 @@ use snafu::{OptionExt, ResultExt};
 use store_api::storage::{RegionId, ScanRequest, TableId};
 use table::metadata::TableType;
 
-use super::REGION_PEERS;
 use crate::error::{
     CreateRecordBatchSnafu, FindRegionRoutesSnafu, InternalSnafu, Result,
     UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::kvbackend::KvBackendCatalogManager;
-use crate::system_schema::information_schema::{InformationTable, Predicates};
+use crate::system_schema::information_schema::{InformationTable, Predicates, REGION_PEERS};
 use crate::CatalogManager;
 
 pub const TABLE_CATALOG: &str = "table_catalog";

src/catalog/src/system_schema/information_schema/region_statistics.rs

@@ -30,9 +30,9 @@ use datatypes::vectors::{StringVectorBuilder, UInt32VectorBuilder, UInt64VectorB
 use snafu::ResultExt;
 use store_api::storage::{ScanRequest, TableId};
 
-use super::{InformationTable, REGION_STATISTICS};
 use crate::error::{CreateRecordBatchSnafu, InternalSnafu, Result};
 use crate::information_schema::Predicates;
+use crate::system_schema::information_schema::{InformationTable, REGION_STATISTICS};
 use crate::system_schema::utils;
 use crate::CatalogManager;
 

src/catalog/src/system_schema/information_schema/runtime_metrics.rs

@@ -35,8 +35,8 @@ use itertools::Itertools;
 use snafu::ResultExt;
 use store_api::storage::{ScanRequest, TableId};
 
-use super::{InformationTable, RUNTIME_METRICS};
 use crate::error::{CreateRecordBatchSnafu, InternalSnafu, Result};
+use crate::system_schema::information_schema::{InformationTable, RUNTIME_METRICS};
 
 #[derive(Debug)]
 pub(super) struct InformationSchemaMetrics {

src/catalog/src/system_schema/information_schema/schemata.rs

@@ -31,12 +31,11 @@ use datatypes::vectors::StringVectorBuilder;
 use snafu::{OptionExt, ResultExt};
 use store_api::storage::{ScanRequest, TableId};
 
-use super::SCHEMATA;
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, TableMetadataManagerSnafu,
     UpgradeWeakCatalogManagerRefSnafu,
 };
-use crate::system_schema::information_schema::{InformationTable, Predicates};
+use crate::system_schema::information_schema::{InformationTable, Predicates, SCHEMATA};
 use crate::system_schema::utils;
 use crate::CatalogManager;
 

src/catalog/src/system_schema/information_schema/table_constraints.rs

@@ -32,14 +32,14 @@ use futures::TryStreamExt;
 use snafu::{OptionExt, ResultExt};
 use store_api::storage::{ScanRequest, TableId};
 
-use super::{InformationTable, TABLE_CONSTRAINTS};
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::information_schema::key_column_usage::{
-    PRI_CONSTRAINT_NAME, TIME_INDEX_CONSTRAINT_NAME,
+    CONSTRAINT_NAME_PRI, CONSTRAINT_NAME_TIME_INDEX,
 };
 use crate::information_schema::Predicates;
+use crate::system_schema::information_schema::{InformationTable, TABLE_CONSTRAINTS};
 use crate::CatalogManager;
 
 /// The `TABLE_CONSTRAINTS` table describes which tables have constraints.
@@ -188,7 +188,7 @@ impl InformationSchemaTableConstraintsBuilder {
                     self.add_table_constraint(
                         &predicates,
                         &schema_name,
-                        TIME_INDEX_CONSTRAINT_NAME,
+                        CONSTRAINT_NAME_TIME_INDEX,
                         &schema_name,
                         &table.table_info().name,
                         TIME_INDEX_CONSTRAINT_TYPE,
@@ -199,7 +199,7 @@ impl InformationSchemaTableConstraintsBuilder {
                     self.add_table_constraint(
                         &predicates,
                         &schema_name,
-                        PRI_CONSTRAINT_NAME,
+                        CONSTRAINT_NAME_PRI,
                         &schema_name,
                         &table.table_info().name,
                         PRI_KEY_CONSTRAINT_TYPE,

src/catalog/src/system_schema/information_schema/tables.rs

@@ -38,11 +38,10 @@ use snafu::{OptionExt, ResultExt};
 use store_api::storage::{RegionId, ScanRequest, TableId};
 use table::metadata::{TableInfo, TableType};
 
-use super::TABLES;
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
-use crate::system_schema::information_schema::{InformationTable, Predicates};
+use crate::system_schema::information_schema::{InformationTable, Predicates, TABLES};
 use crate::system_schema::utils;
 use crate::CatalogManager;
 

src/catalog/src/system_schema/information_schema/views.rs

@@ -32,13 +32,12 @@ use snafu::{OptionExt, ResultExt};
 use store_api::storage::{ScanRequest, TableId};
 use table::metadata::TableType;
 
-use super::VIEWS;
 use crate::error::{
     CastManagerSnafu, CreateRecordBatchSnafu, GetViewCacheSnafu, InternalSnafu, Result,
     UpgradeWeakCatalogManagerRefSnafu, ViewInfoNotFoundSnafu,
 };
 use crate::kvbackend::KvBackendCatalogManager;
-use crate::system_schema::information_schema::{InformationTable, Predicates};
+use crate::system_schema::information_schema::{InformationTable, Predicates, VIEWS};
 use crate::CatalogManager;
 const INIT_CAPACITY: usize = 42;
 

src/catalog/src/system_schema/memory_table.rs

@@ -29,8 +29,8 @@ use datatypes::vectors::VectorRef;
 use snafu::ResultExt;
 use store_api::storage::{ScanRequest, TableId};
 
-use super::SystemTable;
 use crate::error::{CreateRecordBatchSnafu, InternalSnafu, Result};
+use crate::system_schema::SystemTable;
 
 /// A memory table with specified schema and columns.
 #[derive(Debug)]

src/catalog/src/system_schema/pg_catalog.rs

@@ -34,9 +34,9 @@ use table::TableRef;
 pub use table_names::*;
 
 use self::pg_namespace::oid_map::{PGNamespaceOidMap, PGNamespaceOidMapRef};
-use super::memory_table::MemoryTable;
-use super::utils::tables::u32_column;
-use super::{SystemSchemaProvider, SystemSchemaProviderInner, SystemTableRef};
+use crate::system_schema::memory_table::MemoryTable;
+use crate::system_schema::utils::tables::u32_column;
+use crate::system_schema::{SystemSchemaProvider, SystemSchemaProviderInner, SystemTableRef};
 use crate::CatalogManager;
 
 lazy_static! {

src/catalog/src/system_schema/pg_catalog/pg_catalog_memory_table.rs

@@ -17,9 +17,9 @@ use std::sync::Arc;
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
 use datatypes::vectors::{Int16Vector, StringVector, UInt32Vector, VectorRef};
 
-use super::oid_column;
-use super::table_names::PG_TYPE;
 use crate::memory_table_cols;
+use crate::system_schema::pg_catalog::oid_column;
+use crate::system_schema::pg_catalog::table_names::PG_TYPE;
 use crate::system_schema::utils::tables::{i16_column, string_column};
 
 fn pg_type_schema_columns() -> (Vec<ColumnSchema>, Vec<VectorRef>) {

src/catalog/src/system_schema/pg_catalog/pg_class.rs

@@ -32,12 +32,12 @@ use snafu::{OptionExt, ResultExt};
 use store_api::storage::ScanRequest;
 use table::metadata::TableType;
 
-use super::pg_namespace::oid_map::PGNamespaceOidMapRef;
-use super::{query_ctx, OID_COLUMN_NAME, PG_CLASS};
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::information_schema::Predicates;
+use crate::system_schema::pg_catalog::pg_namespace::oid_map::PGNamespaceOidMapRef;
+use crate::system_schema::pg_catalog::{query_ctx, OID_COLUMN_NAME, PG_CLASS};
 use crate::system_schema::utils::tables::{string_column, u32_column};
 use crate::system_schema::SystemTable;
 use crate::CatalogManager;

src/catalog/src/system_schema/pg_catalog/pg_database.rs

@@ -29,12 +29,12 @@ use datatypes::vectors::{StringVectorBuilder, UInt32VectorBuilder, VectorRef};
 use snafu::{OptionExt, ResultExt};
 use store_api::storage::ScanRequest;
 
-use super::pg_namespace::oid_map::PGNamespaceOidMapRef;
-use super::{query_ctx, OID_COLUMN_NAME, PG_DATABASE};
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::information_schema::Predicates;
+use crate::system_schema::pg_catalog::pg_namespace::oid_map::PGNamespaceOidMapRef;
+use crate::system_schema::pg_catalog::{query_ctx, OID_COLUMN_NAME, PG_DATABASE};
 use crate::system_schema::utils::tables::{string_column, u32_column};
 use crate::system_schema::SystemTable;
 use crate::CatalogManager;

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

@@ -35,11 +35,13 @@ use datatypes::vectors::{StringVectorBuilder, UInt32VectorBuilder, VectorRef};
 use snafu::{OptionExt, ResultExt};
 use store_api::storage::ScanRequest;
 
-use super::{query_ctx, PGNamespaceOidMapRef, OID_COLUMN_NAME, PG_NAMESPACE};
 use crate::error::{
     CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::information_schema::Predicates;
+use crate::system_schema::pg_catalog::{
+    query_ctx, PGNamespaceOidMapRef, OID_COLUMN_NAME, PG_NAMESPACE,
+};
 use crate::system_schema::utils::tables::{string_column, u32_column};
 use crate::system_schema::SystemTable;
 use crate::CatalogManager;

src/catalog/src/system_schema/predicate.rs

@@ -437,10 +437,7 @@ mod tests {
     }
 
     fn column(name: &str) -> Expr {
-        Expr::Column(Column {
-            relation: None,
-            name: name.to_string(),
-        })
+        Expr::Column(Column::from_name(name))
     }
 
     fn string_literal(v: &str) -> Expr {

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/src/error.rs

@@ -17,7 +17,6 @@ use std::any::Any;
 use common_error::ext::{BoxedError, ErrorExt};
 use common_error::status_code::StatusCode;
 use common_macro::stack_trace_debug;
-use rustyline::error::ReadlineError;
 use snafu::{Location, Snafu};
 
 #[derive(Snafu)]
@@ -105,52 +104,6 @@ pub enum Error {
     #[snafu(display("Invalid REPL command: {reason}"))]
     InvalidReplCommand { reason: String },
 
-    #[snafu(display("Cannot create REPL"))]
-    ReplCreation {
-        #[snafu(source)]
-        error: ReadlineError,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
-    #[snafu(display("Error reading command"))]
-    Readline {
-        #[snafu(source)]
-        error: ReadlineError,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
-    #[snafu(display("Failed to request database, sql: {sql}"))]
-    RequestDatabase {
-        sql: String,
-        #[snafu(source)]
-        source: client::Error,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
-    #[snafu(display("Failed to collect RecordBatches"))]
-    CollectRecordBatches {
-        #[snafu(implicit)]
-        location: Location,
-        source: common_recordbatch::error::Error,
-    },
-
-    #[snafu(display("Failed to pretty print Recordbatches"))]
-    PrettyPrintRecordBatches {
-        #[snafu(implicit)]
-        location: Location,
-        source: common_recordbatch::error::Error,
-    },
-
-    #[snafu(display("Failed to start Meta client"))]
-    StartMetaClient {
-        #[snafu(implicit)]
-        location: Location,
-        source: meta_client::error::Error,
-    },
-
     #[snafu(display("Failed to parse SQL: {}", sql))]
     ParseSql {
         sql: String,
@@ -166,13 +119,6 @@ pub enum Error {
         source: query::error::Error,
     },
 
-    #[snafu(display("Failed to encode logical plan in substrait"))]
-    SubstraitEncodeLogicalPlan {
-        #[snafu(implicit)]
-        location: Location,
-        source: substrait::error::Error,
-    },
-
     #[snafu(display("Failed to load layered config"))]
     LoadLayeredConfig {
         #[snafu(source(from(common_config::error::Error, Box::new)))]
@@ -318,17 +264,10 @@ impl ErrorExt for Error {
             Error::StartProcedureManager { source, .. }
             | Error::StopProcedureManager { source, .. } => source.status_code(),
             Error::StartWalOptionsAllocator { source, .. } => source.status_code(),
-            Error::ReplCreation { .. } | Error::Readline { .. } | Error::HttpQuerySql { .. } => {
-                StatusCode::Internal
-            }
-            Error::RequestDatabase { source, .. } => source.status_code(),
-            Error::CollectRecordBatches { source, .. }
-            | Error::PrettyPrintRecordBatches { source, .. } => source.status_code(),
-            Error::StartMetaClient { source, .. } => source.status_code(),
+            Error::HttpQuerySql { .. } => StatusCode::Internal,
             Error::ParseSql { source, .. } | Error::PlanStatement { source, .. } => {
                 source.status_code()
             }
-            Error::SubstraitEncodeLogicalPlan { source, .. } => source.status_code(),
 
             Error::SerdeJson { .. }
             | Error::FileIo { .. }

src/cli/src/lib.rs

@@ -23,15 +23,12 @@ mod helper;
 // Wait for https://github.com/GreptimeTeam/greptimedb/issues/2373
 mod database;
 mod import;
-#[allow(unused)]
-mod repl;
 
 use async_trait::async_trait;
 use clap::Parser;
 use common_error::ext::BoxedError;
 pub use database::DatabaseClient;
 use error::Result;
-pub use repl::Repl;
 
 pub use crate::bench::BenchTableMetadataCommand;
 pub use crate::export::ExportCommand;

src/cli/src/repl.rs

@@ -1,299 +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::path::PathBuf;
-use std::sync::Arc;
-use std::time::Instant;
-
-use cache::{
-    build_fundamental_cache_registry, with_default_composite_cache_registry, TABLE_CACHE_NAME,
-    TABLE_ROUTE_CACHE_NAME,
-};
-use catalog::information_extension::DistributedInformationExtension;
-use catalog::kvbackend::{
-    CachedKvBackend, CachedKvBackendBuilder, KvBackendCatalogManager, MetaKvBackend,
-};
-use client::{Client, Database, OutputData, DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
-use common_base::Plugins;
-use common_config::Mode;
-use common_error::ext::ErrorExt;
-use common_meta::cache::{CacheRegistryBuilder, LayeredCacheRegistryBuilder};
-use common_meta::kv_backend::KvBackendRef;
-use common_query::Output;
-use common_recordbatch::RecordBatches;
-use common_telemetry::debug;
-use either::Either;
-use meta_client::client::{ClusterKvBackend, MetaClientBuilder};
-use query::datafusion::DatafusionQueryEngine;
-use query::parser::QueryLanguageParser;
-use query::query_engine::{DefaultSerializer, QueryEngineState};
-use query::QueryEngine;
-use rustyline::error::ReadlineError;
-use rustyline::Editor;
-use session::context::QueryContext;
-use snafu::{OptionExt, ResultExt};
-use substrait::{DFLogicalSubstraitConvertor, SubstraitPlan};
-
-use crate::cmd::ReplCommand;
-use crate::error::{
-    CollectRecordBatchesSnafu, ParseSqlSnafu, PlanStatementSnafu, PrettyPrintRecordBatchesSnafu,
-    ReadlineSnafu, ReplCreationSnafu, RequestDatabaseSnafu, Result, StartMetaClientSnafu,
-    SubstraitEncodeLogicalPlanSnafu,
-};
-use crate::helper::RustylineHelper;
-use crate::{error, AttachCommand};
-
-/// Captures the state of the repl, gathers commands and executes them one by one
-pub struct Repl {
-    /// Rustyline editor for interacting with user on command line
-    rl: Editor<RustylineHelper>,
-
-    /// Current prompt
-    prompt: String,
-
-    /// Client for interacting with GreptimeDB
-    database: Database,
-
-    query_engine: Option<DatafusionQueryEngine>,
-}
-
-#[allow(clippy::print_stdout)]
-impl Repl {
-    fn print_help(&self) {
-        println!("{}", ReplCommand::help())
-    }
-
-    pub(crate) async fn try_new(cmd: &AttachCommand) -> Result<Self> {
-        let mut rl = Editor::new().context(ReplCreationSnafu)?;
-
-        if !cmd.disable_helper {
-            rl.set_helper(Some(RustylineHelper::default()));
-
-            let history_file = history_file();
-            if let Err(e) = rl.load_history(&history_file) {
-                debug!(
-                    "failed to load history file on {}, error: {e}",
-                    history_file.display()
-                );
-            }
-        }
-
-        let client = Client::with_urls([&cmd.grpc_addr]);
-        let database = Database::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, client);
-
-        let query_engine = if let Some(meta_addr) = &cmd.meta_addr {
-            create_query_engine(meta_addr).await.map(Some)?
-        } else {
-            None
-        };
-
-        Ok(Self {
-            rl,
-            prompt: "> ".to_string(),
-            database,
-            query_engine,
-        })
-    }
-
-    /// Parse the next command
-    fn next_command(&mut self) -> Result<ReplCommand> {
-        match self.rl.readline(&self.prompt) {
-            Ok(ref line) => {
-                let request = line.trim();
-
-                let _ = self.rl.add_history_entry(request.to_string());
-
-                request.try_into()
-            }
-            Err(ReadlineError::Eof) | Err(ReadlineError::Interrupted) => Ok(ReplCommand::Exit),
-            // Some sort of real underlying error
-            Err(e) => Err(e).context(ReadlineSnafu),
-        }
-    }
-
-    /// Read Evaluate Print Loop (interactive command line) for GreptimeDB
-    ///
-    /// Inspired / based on repl.rs from InfluxDB IOX
-    pub(crate) async fn run(&mut self) -> Result<()> {
-        println!("Ready for commands. (Hint: try 'help')");
-
-        loop {
-            match self.next_command()? {
-                ReplCommand::Help => {
-                    self.print_help();
-                }
-                ReplCommand::UseDatabase { db_name } => {
-                    if self.execute_sql(format!("USE {db_name}")).await {
-                        println!("Using {db_name}");
-                        self.database.set_schema(&db_name);
-                        self.prompt = format!("[{db_name}] > ");
-                    }
-                }
-                ReplCommand::Sql { sql } => {
-                    let _ = self.execute_sql(sql).await;
-                }
-                ReplCommand::Exit => {
-                    return Ok(());
-                }
-            }
-        }
-    }
-
-    async fn execute_sql(&self, sql: String) -> bool {
-        self.do_execute_sql(sql)
-            .await
-            .map_err(|e| {
-                let status_code = e.status_code();
-                let root_cause = e.output_msg();
-                println!("Error: {}({status_code}), {root_cause}", status_code as u32)
-            })
-            .is_ok()
-    }
-
-    async fn do_execute_sql(&self, sql: String) -> Result<()> {
-        let start = Instant::now();
-
-        let output = if let Some(query_engine) = &self.query_engine {
-            let query_ctx = Arc::new(QueryContext::with(
-                self.database.catalog(),
-                self.database.schema(),
-            ));
-
-            let stmt = QueryLanguageParser::parse_sql(&sql, &query_ctx)
-                .with_context(|_| ParseSqlSnafu { sql: sql.clone() })?;
-
-            let plan = query_engine
-                .planner()
-                .plan(&stmt, query_ctx.clone())
-                .await
-                .context(PlanStatementSnafu)?;
-
-            let plan = query_engine
-                .optimize(&query_engine.engine_context(query_ctx), &plan)
-                .context(PlanStatementSnafu)?;
-
-            let plan = DFLogicalSubstraitConvertor {}
-                .encode(&plan, DefaultSerializer)
-                .context(SubstraitEncodeLogicalPlanSnafu)?;
-
-            self.database.logical_plan(plan.to_vec()).await
-        } else {
-            self.database.sql(&sql).await
-        }
-        .context(RequestDatabaseSnafu { sql: &sql })?;
-
-        let either = match output.data {
-            OutputData::Stream(s) => {
-                let x = RecordBatches::try_collect(s)
-                    .await
-                    .context(CollectRecordBatchesSnafu)?;
-                Either::Left(x)
-            }
-            OutputData::RecordBatches(x) => Either::Left(x),
-            OutputData::AffectedRows(rows) => Either::Right(rows),
-        };
-
-        let end = Instant::now();
-
-        match either {
-            Either::Left(recordbatches) => {
-                let total_rows: usize = recordbatches.iter().map(|x| x.num_rows()).sum();
-                if total_rows > 0 {
-                    println!(
-                        "{}",
-                        recordbatches
-                            .pretty_print()
-                            .context(PrettyPrintRecordBatchesSnafu)?
-                    );
-                }
-                println!("Total Rows: {total_rows}")
-            }
-            Either::Right(rows) => println!("Affected Rows: {rows}"),
-        };
-
-        println!("Cost {} ms", (end - start).as_millis());
-        Ok(())
-    }
-}
-
-impl Drop for Repl {
-    fn drop(&mut self) {
-        if self.rl.helper().is_some() {
-            let history_file = history_file();
-            if let Err(e) = self.rl.save_history(&history_file) {
-                debug!(
-                    "failed to save history file on {}, error: {e}",
-                    history_file.display()
-                );
-            }
-        }
-    }
-}
-
-/// Return the location of the history file (defaults to $HOME/".greptimedb_cli_history")
-fn history_file() -> PathBuf {
-    let mut buf = match std::env::var("HOME") {
-        Ok(home) => PathBuf::from(home),
-        Err(_) => PathBuf::new(),
-    };
-    buf.push(".greptimedb_cli_history");
-    buf
-}
-
-async fn create_query_engine(meta_addr: &str) -> Result<DatafusionQueryEngine> {
-    let mut meta_client = MetaClientBuilder::default().enable_store().build();
-    meta_client
-        .start([meta_addr])
-        .await
-        .context(StartMetaClientSnafu)?;
-    let meta_client = Arc::new(meta_client);
-
-    let cached_meta_backend = Arc::new(
-        CachedKvBackendBuilder::new(Arc::new(MetaKvBackend::new(meta_client.clone()))).build(),
-    );
-    let layered_cache_builder = LayeredCacheRegistryBuilder::default().add_cache_registry(
-        CacheRegistryBuilder::default()
-            .add_cache(cached_meta_backend.clone())
-            .build(),
-    );
-    let fundamental_cache_registry =
-        build_fundamental_cache_registry(Arc::new(MetaKvBackend::new(meta_client.clone())));
-    let layered_cache_registry = Arc::new(
-        with_default_composite_cache_registry(
-            layered_cache_builder.add_cache_registry(fundamental_cache_registry),
-        )
-        .context(error::BuildCacheRegistrySnafu)?
-        .build(),
-    );
-
-    let information_extension = Arc::new(DistributedInformationExtension::new(meta_client.clone()));
-    let catalog_manager = KvBackendCatalogManager::new(
-        information_extension,
-        cached_meta_backend.clone(),
-        layered_cache_registry,
-        None,
-    );
-    let plugins: Plugins = Default::default();
-    let state = Arc::new(QueryEngineState::new(
-        catalog_manager,
-        None,
-        None,
-        None,
-        None,
-        false,
-        plugins.clone(),
-    ));
-
-    Ok(DatafusionQueryEngine::new(state, plugins))
-}

src/client/Cargo.toml

@@ -16,6 +16,7 @@ arc-swap = "1.6"
 arrow-flight.workspace = true
 async-stream.workspace = true
 async-trait.workspace = true
+base64.workspace = true
 common-catalog.workspace = true
 common-error.workspace = true
 common-grpc.workspace = true
@@ -25,6 +26,7 @@ common-query.workspace = true
 common-recordbatch.workspace = true
 common-telemetry.workspace = true
 enum_dispatch = "0.3"
+futures.workspace = true
 futures-util.workspace = true
 lazy_static.workspace = true
 moka = { workspace = true, features = ["future"] }

src/client/src/database.rs

@@ -12,36 +12,49 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::pin::Pin;
+use std::str::FromStr;
+
 use api::v1::auth_header::AuthScheme;
 use api::v1::ddl_request::Expr as DdlExpr;
 use api::v1::greptime_database_client::GreptimeDatabaseClient;
 use api::v1::greptime_request::Request;
 use api::v1::query_request::Query;
 use api::v1::{
-    AlterTableExpr, AuthHeader, CreateTableExpr, DdlRequest, GreptimeRequest, InsertRequests,
-    QueryRequest, RequestHeader,
+    AlterTableExpr, AuthHeader, Basic, CreateTableExpr, DdlRequest, GreptimeRequest,
+    InsertRequests, QueryRequest, RequestHeader,
 };
-use arrow_flight::Ticket;
+use arrow_flight::{FlightData, Ticket};
 use async_stream::stream;
+use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
+use common_catalog::build_db_string;
+use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_error::ext::{BoxedError, ErrorExt};
+use common_grpc::flight::do_put::DoPutResponse;
 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 futures_util::StreamExt;
+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;
+use tonic::metadata::{AsciiMetadataKey, MetadataValue};
 use tonic::transport::Channel;
 
 use crate::error::{
     ConvertFlightDataSnafu, Error, FlightGetSnafu, IllegalFlightMessagesSnafu, InvalidAsciiSnafu,
-    ServerSnafu,
+    InvalidTonicMetadataValueSnafu, ServerSnafu,
 };
 use crate::{from_grpc_response, Client, Result};
 
+type FlightDataStream = Pin<Box<dyn Stream<Item = FlightData> + Send>>;
+
+type DoPutResponseStream = Pin<Box<dyn Stream<Item = Result<DoPutResponse>>>>;
+
 #[derive(Clone, Debug, Default)]
 pub struct Database {
     // The "catalog" and "schema" to be used in processing the requests at the server side.
@@ -108,16 +121,24 @@ impl Database {
         self.catalog = catalog.into();
     }
 
-    pub fn catalog(&self) -> &String {
-        &self.catalog
+    fn catalog_or_default(&self) -> &str {
+        if self.catalog.is_empty() {
+            DEFAULT_CATALOG_NAME
+        } else {
+            &self.catalog
+        }
     }
 
     pub fn set_schema(&mut self, schema: impl Into<String>) {
         self.schema = schema.into();
     }
 
-    pub fn schema(&self) -> &String {
-        &self.schema
+    fn schema_or_default(&self) -> &str {
+        if self.schema.is_empty() {
+            DEFAULT_SCHEMA_NAME
+        } else {
+            &self.schema
+        }
     }
 
     pub fn set_timezone(&mut self, timezone: impl Into<String>) {
@@ -164,13 +185,43 @@ impl Database {
         from_grpc_response(response)
     }
 
-    async fn handle(&self, request: Request) -> Result<u32> {
+    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);
         let response = client.handle(request).await?.into_inner();
         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 {
@@ -310,6 +361,46 @@ impl Database {
             }
         }
     }
+
+    /// Ingest a stream of [RecordBatch]es that belong to a table, using Arrow Flight's "`DoPut`"
+    /// method. The return value is also a stream, produces [DoPutResponse]s.
+    pub async fn do_put(&self, stream: FlightDataStream) -> Result<DoPutResponseStream> {
+        let mut request = tonic::Request::new(stream);
+
+        if let Some(AuthHeader {
+            auth_scheme: Some(AuthScheme::Basic(Basic { username, password })),
+        }) = &self.ctx.auth_header
+        {
+            let encoded = BASE64_STANDARD.encode(format!("{username}:{password}"));
+            let value =
+                MetadataValue::from_str(&encoded).context(InvalidTonicMetadataValueSnafu)?;
+            request.metadata_mut().insert("x-greptime-auth", value);
+        }
+
+        let db_to_put = if !self.dbname.is_empty() {
+            &self.dbname
+        } else {
+            &build_db_string(self.catalog_or_default(), self.schema_or_default())
+        };
+        request.metadata_mut().insert(
+            "x-greptime-db-name",
+            MetadataValue::from_str(db_to_put).context(InvalidTonicMetadataValueSnafu)?,
+        );
+
+        let mut client = self.client.make_flight_client()?;
+        let response = client.mut_inner().do_put(request).await?;
+        let response = response
+            .into_inner()
+            .map_err(Into::into)
+            .and_then(|x| future::ready(DoPutResponse::try_from(x).context(ConvertFlightDataSnafu)))
+            .boxed();
+        Ok(response)
+    }
+}
+
+/// 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)]

src/client/src/error.rs

@@ -15,10 +15,11 @@
 use std::any::Any;
 
 use common_error::ext::{BoxedError, ErrorExt};
-use common_error::status_code::StatusCode;
+use common_error::status_code::{convert_tonic_code_to_status_code, StatusCode};
 use common_error::{GREPTIME_DB_HEADER_ERROR_CODE, GREPTIME_DB_HEADER_ERROR_MSG};
 use common_macro::stack_trace_debug;
 use snafu::{location, Location, Snafu};
+use tonic::metadata::errors::InvalidMetadataValue;
 use tonic::{Code, Status};
 
 #[derive(Snafu)]
@@ -115,6 +116,14 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(display("Invalid Tonic metadata value"))]
+    InvalidTonicMetadataValue {
+        #[snafu(source)]
+        error: InvalidMetadataValue,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -135,7 +144,9 @@ impl ErrorExt for Error {
             | Error::CreateTlsChannel { source, .. } => source.status_code(),
             Error::IllegalGrpcClientState { .. } => StatusCode::Unexpected,
 
-            Error::InvalidAscii { .. } => StatusCode::InvalidArguments,
+            Error::InvalidAscii { .. } | Error::InvalidTonicMetadataValue { .. } => {
+                StatusCode::InvalidArguments
+            }
         }
     }
 
@@ -152,15 +163,15 @@ impl From<Status> for Error {
                 .and_then(|v| String::from_utf8(v.as_bytes().to_vec()).ok())
         }
 
-        let code = get_metadata_value(&e, GREPTIME_DB_HEADER_ERROR_CODE)
-            .and_then(|s| {
-                if let Ok(code) = s.parse::<u32>() {
-                    StatusCode::from_u32(code)
-                } else {
-                    None
-                }
-            })
-            .unwrap_or(StatusCode::Unknown);
+        let code = get_metadata_value(&e, GREPTIME_DB_HEADER_ERROR_CODE).and_then(|s| {
+            if let Ok(code) = s.parse::<u32>() {
+                StatusCode::from_u32(code)
+            } else {
+                None
+            }
+        });
+        let tonic_code = e.code();
+        let code = code.unwrap_or_else(|| convert_tonic_code_to_status_code(tonic_code));
 
         let msg = get_metadata_value(&e, GREPTIME_DB_HEADER_ERROR_MSG)
             .unwrap_or_else(|| e.message().to_string());
@@ -187,9 +198,6 @@ impl Error {
             } | Self::RegionServer {
                 code: Code::Unavailable,
                 ..
-            } | Self::RegionServer {
-                code: Code::Unknown,
-                ..
             }
         )
     }

src/client/src/lib.rs

@@ -16,7 +16,7 @@
 
 mod client;
 pub mod client_manager;
-mod database;
+pub mod database;
 pub mod error;
 pub mod flow;
 pub mod load_balance;

src/client/src/region.rs

@@ -201,12 +201,11 @@ impl RegionRequester {
             .await
             .map_err(|e| {
                 let code = e.code();
-                let err: error::Error = e.into();
                 // Uses `Error::RegionServer` instead of `Error::Server`
                 error::Error::RegionServer {
                     addr,
                     code,
-                    source: BoxedError::new(err),
+                    source: BoxedError::new(error::Error::from(e)),
                     location: location!(),
                 }
             })?

src/cmd/Cargo.toml

@@ -68,7 +68,6 @@ query.workspace = true
 rand.workspace = true
 regex.workspace = true
 reqwest.workspace = true
-rustyline = "10.1"
 serde.workspace = true
 serde_json.workspace = true
 servers.workspace = true

src/cmd/src/error.rs

@@ -17,7 +17,6 @@ use std::any::Any;
 use common_error::ext::{BoxedError, ErrorExt};
 use common_error::status_code::StatusCode;
 use common_macro::stack_trace_debug;
-use rustyline::error::ReadlineError;
 use snafu::{Location, Snafu};
 
 #[derive(Snafu)]
@@ -181,52 +180,6 @@ pub enum Error {
     #[snafu(display("Invalid REPL command: {reason}"))]
     InvalidReplCommand { reason: String },
 
-    #[snafu(display("Cannot create REPL"))]
-    ReplCreation {
-        #[snafu(source)]
-        error: ReadlineError,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
-    #[snafu(display("Error reading command"))]
-    Readline {
-        #[snafu(source)]
-        error: ReadlineError,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
-    #[snafu(display("Failed to request database, sql: {sql}"))]
-    RequestDatabase {
-        sql: String,
-        #[snafu(source)]
-        source: client::Error,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
-    #[snafu(display("Failed to collect RecordBatches"))]
-    CollectRecordBatches {
-        #[snafu(implicit)]
-        location: Location,
-        source: common_recordbatch::error::Error,
-    },
-
-    #[snafu(display("Failed to pretty print Recordbatches"))]
-    PrettyPrintRecordBatches {
-        #[snafu(implicit)]
-        location: Location,
-        source: common_recordbatch::error::Error,
-    },
-
-    #[snafu(display("Failed to start Meta client"))]
-    StartMetaClient {
-        #[snafu(implicit)]
-        location: Location,
-        source: meta_client::error::Error,
-    },
-
     #[snafu(display("Failed to parse SQL: {}", sql))]
     ParseSql {
         sql: String,
@@ -242,13 +195,6 @@ pub enum Error {
         source: query::error::Error,
     },
 
-    #[snafu(display("Failed to encode logical plan in substrait"))]
-    SubstraitEncodeLogicalPlan {
-        #[snafu(implicit)]
-        location: Location,
-        source: substrait::error::Error,
-    },
-
     #[snafu(display("Failed to load layered config"))]
     LoadLayeredConfig {
         #[snafu(source(from(common_config::error::Error, Box::new)))]
@@ -395,17 +341,10 @@ impl ErrorExt for Error {
             | Error::StopProcedureManager { source, .. } => source.status_code(),
             Error::BuildWalOptionsAllocator { source, .. }
             | Error::StartWalOptionsAllocator { source, .. } => source.status_code(),
-            Error::ReplCreation { .. } | Error::Readline { .. } | Error::HttpQuerySql { .. } => {
-                StatusCode::Internal
-            }
-            Error::RequestDatabase { source, .. } => source.status_code(),
-            Error::CollectRecordBatches { source, .. }
-            | Error::PrettyPrintRecordBatches { source, .. } => source.status_code(),
-            Error::StartMetaClient { source, .. } => source.status_code(),
+            Error::HttpQuerySql { .. } => StatusCode::Internal,
             Error::ParseSql { source, .. } | Error::PlanStatement { source, .. } => {
                 source.status_code()
             }
-            Error::SubstraitEncodeLogicalPlan { source, .. } => source.status_code(),
 
             Error::SerdeJson { .. }
             | Error::FileIo { .. }

src/cmd/src/flownode.rs

@@ -32,7 +32,9 @@ use common_meta::key::TableMetadataManager;
 use common_telemetry::info;
 use common_telemetry::logging::TracingOptions;
 use common_version::{short_version, version};
-use flow::{FlownodeBuilder, FlownodeInstance, FrontendInvoker};
+use flow::{
+    FlownodeBuilder, FlownodeInstance, FlownodeServiceBuilder, FrontendClient, FrontendInvoker,
+};
 use meta_client::{MetaClientOptions, MetaClientType};
 use snafu::{ensure, OptionExt, ResultExt};
 use tracing_appender::non_blocking::WorkerGuard;
@@ -313,16 +315,26 @@ 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 flownode_builder = FlownodeBuilder::new(
-            opts,
+            opts.clone(),
             Plugins::new(),
             table_metadata_manager,
             catalog_manager.clone(),
             flow_metadata_manager,
+            Arc::new(frontend_client),
         )
         .with_heartbeat_task(heartbeat_task);
 
-        let flownode = flownode_builder.build().await.context(StartFlownodeSnafu)?;
+        let mut flownode = flownode_builder.build().await.context(StartFlownodeSnafu)?;
+        let services = FlownodeServiceBuilder::new(&opts)
+            .with_grpc_server(flownode.flownode_server().clone())
+            .enable_http_service()
+            .build()
+            .await
+            .context(StartFlownodeSnafu)?;
+        flownode.setup_services(services);
+        let flownode = flownode;
 
         // flownode's frontend to datanode need not timeout.
         // Some queries are expected to take long time.
@@ -333,7 +345,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(),
@@ -343,7 +355,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/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;
@@ -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

@@ -55,7 +55,10 @@ use datanode::config::{DatanodeOptions, ProcedureConfig, RegionEngineConfig, Sto
 use datanode::datanode::{Datanode, DatanodeBuilder};
 use datanode::region_server::RegionServer;
 use file_engine::config::EngineConfig as FileEngineConfig;
-use flow::{FlowConfig, FlowWorkerManager, FlownodeBuilder, FlownodeOptions, FrontendInvoker};
+use flow::{
+    FlowConfig, FlownodeBuilder, FlownodeInstance, FlownodeOptions, FrontendClient,
+    FrontendInvoker, GrpcQueryHandlerWithBoxedError, StreamingEngine,
+};
 use frontend::frontend::{Frontend, FrontendOptions};
 use frontend::instance::builder::FrontendBuilder;
 use frontend::instance::{Instance as FeInstance, StandaloneDatanodeManager};
@@ -74,10 +77,10 @@ use servers::http::HttpOptions;
 use servers::tls::{TlsMode, TlsOption};
 use servers::Mode;
 use snafu::ResultExt;
-use tokio::sync::{broadcast, RwLock};
+use tokio::sync::RwLock;
 use tracing_appender::non_blocking::WorkerGuard;
 
-use crate::error::Result;
+use crate::error::{Result, StartFlownodeSnafu};
 use crate::options::{GlobalOptions, GreptimeOptions};
 use crate::{error, log_versions, App};
 
@@ -244,9 +247,7 @@ impl StandaloneOptions {
 pub struct Instance {
     datanode: Datanode,
     frontend: Frontend,
-    // TODO(discord9): wrapped it in flownode instance instead
-    flow_worker_manager: Arc<FlowWorkerManager>,
-    flow_shutdown: broadcast::Sender<()>,
+    flownode: FlownodeInstance,
     procedure_manager: ProcedureManagerRef,
     wal_options_allocator: WalOptionsAllocatorRef,
     // Keep the logging guard to prevent the worker from being dropped.
@@ -288,9 +289,7 @@ impl App for Instance {
             .await
             .context(error::StartFrontendSnafu)?;
 
-        self.flow_worker_manager
-            .clone()
-            .run_background(Some(self.flow_shutdown.subscribe()));
+        self.flownode.start().await.context(StartFlownodeSnafu)?;
 
         Ok(())
     }
@@ -311,14 +310,9 @@ impl App for Instance {
             .await
             .context(error::ShutdownDatanodeSnafu)?;
 
-        self.flow_shutdown
-            .send(())
-            .map_err(|_e| {
-                flow::error::InternalSnafu {
-                    reason: "Failed to send shutdown signal to flow worker manager, all receiver end already closed".to_string(),
-                }
-                .build()
-            })
+        self.flownode
+            .shutdown()
+            .await
             .context(error::ShutdownFlownodeSnafu)?;
 
         info!("Datanode instance stopped.");
@@ -529,32 +523,36 @@ impl StartCommand {
             flow: opts.flow.clone(),
             ..Default::default()
         };
+
+        // for standalone not use grpc, but get a handler to frontend grpc client without
+        // actually make a connection
+        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.clone()),
         );
-        let flownode = Arc::new(
-            flow_builder
-                .build()
-                .await
-                .map_err(BoxedError::new)
-                .context(error::OtherSnafu)?,
-        );
+        let flownode = flow_builder
+            .build()
+            .await
+            .map_err(BoxedError::new)
+            .context(error::OtherSnafu)?;
 
         // 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(
@@ -608,10 +606,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(),
@@ -620,9 +627,7 @@ impl StartCommand {
         )
         .await
         .context(error::StartFlownodeSnafu)?;
-        flow_worker_manager.set_frontend_invoker(invoker).await;
-
-        let (tx, _rx) = broadcast::channel(1);
+        flow_streaming_engine.set_frontend_invoker(invoker).await;
 
         let export_metrics_task = ExportMetricsTask::try_new(&opts.export_metrics, Some(&plugins))
             .context(error::ServersSnafu)?;
@@ -642,8 +647,7 @@ impl StartCommand {
         Ok(Instance {
             datanode,
             frontend,
-            flow_worker_manager,
-            flow_shutdown: tx,
+            flownode,
             procedure_manager,
             wal_options_allocator,
             _guard: guard,
@@ -699,7 +703,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 {
@@ -708,14 +712,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);
     }
 }
 
@@ -784,6 +788,8 @@ impl InformationExtension for StandaloneInformationExtension {
                     sst_size: region_stat.sst_size,
                     index_size: region_stat.index_size,
                     region_manifest: region_stat.manifest.into(),
+                    data_topic_latest_entry_id: region_stat.data_topic_latest_entry_id,
+                    metadata_topic_latest_entry_id: region_stat.metadata_topic_latest_entry_id,
                 }
             })
             .collect::<Vec<_>>();
@@ -792,7 +798,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()

src/cmd/tests/cli.rs

@@ -1,148 +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.
-
-#[cfg(target_os = "macos")]
-mod tests {
-    use std::path::PathBuf;
-    use std::process::{Command, Stdio};
-    use std::time::Duration;
-
-    use common_test_util::temp_dir::create_temp_dir;
-    use rexpect::session::PtyReplSession;
-
-    struct Repl {
-        repl: PtyReplSession,
-    }
-
-    impl Repl {
-        fn send_line(&mut self, line: &str) {
-            let _ = self.repl.send_line(line).unwrap();
-
-            // read a line to consume the prompt
-            let _ = self.read_line();
-        }
-
-        fn read_line(&mut self) -> String {
-            self.repl.read_line().unwrap()
-        }
-
-        fn read_expect(&mut self, expect: &str) {
-            assert_eq!(self.read_line(), expect);
-        }
-
-        fn read_contains(&mut self, pat: &str) {
-            assert!(self.read_line().contains(pat));
-        }
-    }
-
-    // TODO(LFC): Un-ignore this REPL test.
-    // Ignore this REPL test because some logical plans like create database are not supported yet in Datanode.
-    #[ignore]
-    #[test]
-    fn test_repl() {
-        let data_home = create_temp_dir("data");
-        let wal_dir = create_temp_dir("wal");
-
-        let mut bin_path = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
-        bin_path.push("../../target/debug");
-        let bin_path = bin_path.to_str().unwrap();
-
-        let mut datanode = Command::new("./greptime")
-            .current_dir(bin_path)
-            .args([
-                "datanode",
-                "start",
-                "--rpc-bind-addr=0.0.0.0:4321",
-                "--node-id=1",
-                &format!("--data-home={}", data_home.path().display()),
-                &format!("--wal-dir={}", wal_dir.path().display()),
-            ])
-            .stdout(Stdio::null())
-            .spawn()
-            .unwrap();
-
-        // wait for Datanode actually started
-        std::thread::sleep(Duration::from_secs(3));
-
-        let mut repl_cmd = Command::new("./greptime");
-        let _ = repl_cmd.current_dir(bin_path).args([
-            "--log-level=off",
-            "cli",
-            "attach",
-            "--grpc-bind-addr=0.0.0.0:4321",
-            // history commands can sneaky into stdout and mess up our tests, so disable it
-            "--disable-helper",
-        ]);
-        let pty_session = rexpect::session::spawn_command(repl_cmd, Some(5_000)).unwrap();
-        let repl = PtyReplSession {
-            prompt: "> ".to_string(),
-            pty_session,
-            quit_command: None,
-            echo_on: false,
-        };
-        let repl = &mut Repl { repl };
-        repl.read_expect("Ready for commands. (Hint: try 'help')");
-
-        test_create_database(repl);
-
-        test_use_database(repl);
-
-        test_create_table(repl);
-
-        test_insert(repl);
-
-        test_select(repl);
-
-        datanode.kill().unwrap();
-        let _ = datanode.wait().unwrap();
-    }
-
-    fn test_create_database(repl: &mut Repl) {
-        repl.send_line("CREATE DATABASE db;");
-        repl.read_expect("Affected Rows: 1");
-        repl.read_contains("Cost");
-    }
-
-    fn test_use_database(repl: &mut Repl) {
-        repl.send_line("USE db");
-        repl.read_expect("Total Rows: 0");
-        repl.read_contains("Cost");
-        repl.read_expect("Using db");
-    }
-
-    fn test_create_table(repl: &mut Repl) {
-        repl.send_line("CREATE TABLE t(x STRING, ts TIMESTAMP TIME INDEX);");
-        repl.read_expect("Affected Rows: 0");
-        repl.read_contains("Cost");
-    }
-
-    fn test_insert(repl: &mut Repl) {
-        repl.send_line("INSERT INTO t(x, ts) VALUES ('hello', 1676895812239);");
-        repl.read_expect("Affected Rows: 1");
-        repl.read_contains("Cost");
-    }
-
-    fn test_select(repl: &mut Repl) {
-        repl.send_line("SELECT * FROM t;");
-
-        repl.read_expect("+-------+-------------------------+");
-        repl.read_expect("| x     | ts                      |");
-        repl.read_expect("+-------+-------------------------+");
-        repl.read_expect("| hello | 2023-02-20T12:23:32.239 |");
-        repl.read_expect("+-------+-------------------------+");
-        repl.read_expect("Total Rows: 1");
-
-        repl.read_contains("Cost");
-    }
-}

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/datasource/Cargo.toml

@@ -31,7 +31,8 @@ derive_builder.workspace = true
 futures.workspace = true
 lazy_static.workspace = true
 object-store.workspace = true
-orc-rust = { version = "0.5", default-features = false, features = [
+object_store_opendal.workspace = true
+orc-rust = { git = "https://github.com/datafusion-contrib/orc-rust", rev = "3134cab581a8e91b942d6a23aca2916ea965f6bb", default-features = false, features = [
     "async",
 ] }
 parquet.workspace = true

src/common/datasource/src/compression.rs

@@ -19,6 +19,7 @@ use std::str::FromStr;
 use async_compression::tokio::bufread::{BzDecoder, GzipDecoder, XzDecoder, ZstdDecoder};
 use async_compression::tokio::write;
 use bytes::Bytes;
+use datafusion::datasource::file_format::file_compression_type::FileCompressionType;
 use futures::Stream;
 use serde::{Deserialize, Serialize};
 use strum::EnumIter;
@@ -192,3 +193,15 @@ macro_rules! impl_compression_type {
 }
 
 impl_compression_type!((Gzip, Gzip), (Bzip2, Bz), (Xz, Xz), (Zstd, Zstd));
+
+impl From<CompressionType> for FileCompressionType {
+    fn from(t: CompressionType) -> Self {
+        match t {
+            CompressionType::Gzip => FileCompressionType::GZIP,
+            CompressionType::Bzip2 => FileCompressionType::BZIP2,
+            CompressionType::Xz => FileCompressionType::XZ,
+            CompressionType::Zstd => FileCompressionType::ZSTD,
+            CompressionType::Uncompressed => FileCompressionType::UNCOMPRESSED,
+        }
+    }
+}

src/common/datasource/src/file_format/csv.rs

@@ -14,28 +14,23 @@
 
 use std::collections::HashMap;
 use std::str::FromStr;
-use std::sync::Arc;
 
 use arrow::csv;
 use arrow::csv::reader::Format;
 use arrow::record_batch::RecordBatch;
-use arrow_schema::{Schema, SchemaRef};
+use arrow_schema::Schema;
 use async_trait::async_trait;
 use common_runtime;
-use datafusion::datasource::physical_plan::{FileMeta, FileOpenFuture, FileOpener};
-use datafusion::error::Result as DataFusionResult;
 use datafusion::physical_plan::SendableRecordBatchStream;
-use derive_builder::Builder;
 use object_store::ObjectStore;
 use snafu::ResultExt;
 use tokio_util::compat::FuturesAsyncReadCompatExt;
 use tokio_util::io::SyncIoBridge;
 
-use super::stream_to_file;
 use crate::buffered_writer::DfRecordBatchEncoder;
 use crate::compression::CompressionType;
 use crate::error::{self, Result};
-use crate::file_format::{self, open_with_decoder, FileFormat};
+use crate::file_format::{self, stream_to_file, FileFormat};
 use crate::share_buffer::SharedBuffer;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -100,66 +95,6 @@ impl Default for CsvFormat {
     }
 }
 
-#[derive(Debug, Clone, Builder)]
-pub struct CsvConfig {
-    batch_size: usize,
-    file_schema: SchemaRef,
-    #[builder(default = "None")]
-    file_projection: Option<Vec<usize>>,
-    #[builder(default = "true")]
-    has_header: bool,
-    #[builder(default = "b','")]
-    delimiter: u8,
-}
-
-impl CsvConfig {
-    fn builder(&self) -> csv::ReaderBuilder {
-        let mut builder = csv::ReaderBuilder::new(self.file_schema.clone())
-            .with_delimiter(self.delimiter)
-            .with_batch_size(self.batch_size)
-            .with_header(self.has_header);
-
-        if let Some(proj) = &self.file_projection {
-            builder = builder.with_projection(proj.clone());
-        }
-
-        builder
-    }
-}
-
-#[derive(Debug, Clone)]
-pub struct CsvOpener {
-    config: Arc<CsvConfig>,
-    object_store: Arc<ObjectStore>,
-    compression_type: CompressionType,
-}
-
-impl CsvOpener {
-    /// Return a new [`CsvOpener`]. The caller must ensure [`CsvConfig`].file_schema must correspond to the opening file.
-    pub fn new(
-        config: CsvConfig,
-        object_store: ObjectStore,
-        compression_type: CompressionType,
-    ) -> Self {
-        CsvOpener {
-            config: Arc::new(config),
-            object_store: Arc::new(object_store),
-            compression_type,
-        }
-    }
-}
-
-impl FileOpener for CsvOpener {
-    fn open(&self, meta: FileMeta) -> DataFusionResult<FileOpenFuture> {
-        open_with_decoder(
-            self.object_store.clone(),
-            meta.location().to_string(),
-            self.compression_type,
-            || Ok(self.config.builder().build_decoder()),
-        )
-    }
-}
-
 #[async_trait]
 impl FileFormat for CsvFormat {
     async fn infer_schema(&self, store: &ObjectStore, path: &str) -> Result<Schema> {

src/common/datasource/src/file_format/json.rs

@@ -15,29 +15,24 @@
 use std::collections::HashMap;
 use std::io::BufReader;
 use std::str::FromStr;
-use std::sync::Arc;
 
-use arrow::datatypes::SchemaRef;
+use arrow::json;
 use arrow::json::reader::{infer_json_schema_from_iterator, ValueIter};
 use arrow::json::writer::LineDelimited;
-use arrow::json::{self, ReaderBuilder};
 use arrow::record_batch::RecordBatch;
 use arrow_schema::Schema;
 use async_trait::async_trait;
 use common_runtime;
-use datafusion::datasource::physical_plan::{FileMeta, FileOpenFuture, FileOpener};
-use datafusion::error::{DataFusionError, Result as DataFusionResult};
 use datafusion::physical_plan::SendableRecordBatchStream;
 use object_store::ObjectStore;
 use snafu::ResultExt;
 use tokio_util::compat::FuturesAsyncReadCompatExt;
 use tokio_util::io::SyncIoBridge;
 
-use super::stream_to_file;
 use crate::buffered_writer::DfRecordBatchEncoder;
 use crate::compression::CompressionType;
 use crate::error::{self, Result};
-use crate::file_format::{self, open_with_decoder, FileFormat};
+use crate::file_format::{self, stream_to_file, FileFormat};
 use crate::share_buffer::SharedBuffer;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -114,47 +109,6 @@ impl FileFormat for JsonFormat {
     }
 }
 
-#[derive(Debug, Clone)]
-pub struct JsonOpener {
-    batch_size: usize,
-    projected_schema: SchemaRef,
-    object_store: Arc<ObjectStore>,
-    compression_type: CompressionType,
-}
-
-impl JsonOpener {
-    /// Return a new [`JsonOpener`]. Any fields not present in `projected_schema` will be ignored.
-    pub fn new(
-        batch_size: usize,
-        projected_schema: SchemaRef,
-        object_store: ObjectStore,
-        compression_type: CompressionType,
-    ) -> Self {
-        Self {
-            batch_size,
-            projected_schema,
-            object_store: Arc::new(object_store),
-            compression_type,
-        }
-    }
-}
-
-impl FileOpener for JsonOpener {
-    fn open(&self, meta: FileMeta) -> DataFusionResult<FileOpenFuture> {
-        open_with_decoder(
-            self.object_store.clone(),
-            meta.location().to_string(),
-            self.compression_type,
-            || {
-                ReaderBuilder::new(self.projected_schema.clone())
-                    .with_batch_size(self.batch_size)
-                    .build_decoder()
-                    .map_err(DataFusionError::from)
-            },
-        )
-    }
-}
-
 pub async fn stream_to_json(
     stream: SendableRecordBatchStream,
     store: ObjectStore,

src/common/datasource/src/file_format/tests.rs

@@ -19,7 +19,10 @@ use std::vec;
 
 use common_test_util::find_workspace_path;
 use datafusion::assert_batches_eq;
-use datafusion::datasource::physical_plan::{FileOpener, FileScanConfig, FileStream, ParquetExec};
+use datafusion::datasource::file_format::file_compression_type::FileCompressionType;
+use datafusion::datasource::physical_plan::{
+    CsvConfig, CsvOpener, FileOpener, FileScanConfig, FileStream, JsonOpener, ParquetExec,
+};
 use datafusion::execution::context::TaskContext;
 use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
 use datafusion::physical_plan::ExecutionPlan;
@@ -27,14 +30,11 @@ use datafusion::prelude::SessionContext;
 use futures::StreamExt;
 
 use super::FORMAT_TYPE;
-use crate::compression::CompressionType;
-use crate::error;
-use crate::file_format::csv::{CsvConfigBuilder, CsvOpener};
-use crate::file_format::json::JsonOpener;
 use crate::file_format::orc::{OrcFormat, OrcOpener};
 use crate::file_format::parquet::DefaultParquetFileReaderFactory;
 use crate::file_format::{FileFormat, Format};
-use crate::test_util::{self, scan_config, test_basic_schema, test_store};
+use crate::test_util::{scan_config, test_basic_schema, test_store};
+use crate::{error, test_util};
 
 struct Test<'a, T: FileOpener> {
     config: FileScanConfig,
@@ -62,15 +62,18 @@ impl<T: FileOpener> Test<'_, T> {
 #[tokio::test]
 async fn test_json_opener() {
     let store = test_store("/");
+    let store = Arc::new(object_store_opendal::OpendalStore::new(store));
 
     let schema = test_basic_schema();
 
-    let json_opener = JsonOpener::new(
-        100,
-        schema.clone(),
-        store.clone(),
-        CompressionType::Uncompressed,
-    );
+    let json_opener = || {
+        JsonOpener::new(
+            test_util::TEST_BATCH_SIZE,
+            schema.clone(),
+            FileCompressionType::UNCOMPRESSED,
+            store.clone(),
+        )
+    };
 
     let path = &find_workspace_path("/src/common/datasource/tests/json/basic.json")
         .display()
@@ -78,7 +81,7 @@ async fn test_json_opener() {
     let tests = [
         Test {
             config: scan_config(schema.clone(), None, path),
-            opener: json_opener.clone(),
+            opener: json_opener(),
             expected: vec![
                 "+-----+-------+",
                 "| num | str   |",
@@ -91,7 +94,7 @@ async fn test_json_opener() {
         },
         Test {
             config: scan_config(schema.clone(), Some(1), path),
-            opener: json_opener.clone(),
+            opener: json_opener(),
             expected: vec![
                 "+-----+------+",
                 "| num | str  |",
@@ -110,23 +113,30 @@ async fn test_json_opener() {
 #[tokio::test]
 async fn test_csv_opener() {
     let store = test_store("/");
+    let store = Arc::new(object_store_opendal::OpendalStore::new(store));
 
     let schema = test_basic_schema();
     let path = &find_workspace_path("/src/common/datasource/tests/csv/basic.csv")
         .display()
         .to_string();
-    let csv_conf = CsvConfigBuilder::default()
-        .batch_size(test_util::TEST_BATCH_SIZE)
-        .file_schema(schema.clone())
-        .build()
-        .unwrap();
+    let csv_config = Arc::new(CsvConfig::new(
+        test_util::TEST_BATCH_SIZE,
+        schema.clone(),
+        None,
+        true,
+        b',',
+        b'"',
+        None,
+        store,
+        None,
+    ));
 
-    let csv_opener = CsvOpener::new(csv_conf, store, CompressionType::Uncompressed);
+    let csv_opener = || CsvOpener::new(csv_config.clone(), FileCompressionType::UNCOMPRESSED);
 
     let tests = [
         Test {
             config: scan_config(schema.clone(), None, path),
-            opener: csv_opener.clone(),
+            opener: csv_opener(),
             expected: vec![
                 "+-----+-------+",
                 "| num | str   |",
@@ -139,7 +149,7 @@ async fn test_csv_opener() {
         },
         Test {
             config: scan_config(schema.clone(), Some(1), path),
-            opener: csv_opener.clone(),
+            opener: csv_opener(),
             expected: vec![
                 "+-----+------+",
                 "| num | str  |",

src/common/datasource/src/test_util.rs

@@ -16,17 +16,19 @@ use std::sync::Arc;
 
 use arrow_schema::{DataType, Field, Schema, SchemaRef};
 use common_test_util::temp_dir::{create_temp_dir, TempDir};
-use datafusion::common::Statistics;
+use datafusion::common::{Constraints, Statistics};
+use datafusion::datasource::file_format::file_compression_type::FileCompressionType;
 use datafusion::datasource::listing::PartitionedFile;
 use datafusion::datasource::object_store::ObjectStoreUrl;
-use datafusion::datasource::physical_plan::{FileScanConfig, FileStream};
+use datafusion::datasource::physical_plan::{
+    CsvConfig, CsvOpener, FileScanConfig, FileStream, JsonOpener,
+};
 use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
 use object_store::services::Fs;
 use object_store::ObjectStore;
 
-use crate::compression::CompressionType;
-use crate::file_format::csv::{stream_to_csv, CsvConfigBuilder, CsvOpener};
-use crate::file_format::json::{stream_to_json, JsonOpener};
+use crate::file_format::csv::stream_to_csv;
+use crate::file_format::json::stream_to_json;
 use crate::test_util;
 
 pub const TEST_BATCH_SIZE: usize = 100;
@@ -74,6 +76,7 @@ pub fn scan_config(file_schema: SchemaRef, limit: Option<usize>, filename: &str)
         object_store_url: ObjectStoreUrl::parse("empty://").unwrap(), // won't be used
         file_schema,
         file_groups: vec![vec![PartitionedFile::new(filename.to_string(), 10)]],
+        constraints: Constraints::empty(),
         statistics,
         projection: None,
         limit,
@@ -90,8 +93,8 @@ pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usi
     let json_opener = JsonOpener::new(
         test_util::TEST_BATCH_SIZE,
         schema.clone(),
-        store.clone(),
-        CompressionType::Uncompressed,
+        FileCompressionType::UNCOMPRESSED,
+        Arc::new(object_store_opendal::OpendalStore::new(store.clone())),
     );
 
     let size = store.read(origin_path).await.unwrap().len();
@@ -124,13 +127,19 @@ pub async fn setup_stream_to_csv_test(origin_path: &str, threshold: impl Fn(usiz
 
     let schema = test_basic_schema();
 
-    let csv_conf = CsvConfigBuilder::default()
-        .batch_size(test_util::TEST_BATCH_SIZE)
-        .file_schema(schema.clone())
-        .build()
-        .unwrap();
+    let csv_config = Arc::new(CsvConfig::new(
+        TEST_BATCH_SIZE,
+        schema.clone(),
+        None,
+        true,
+        b',',
+        b'"',
+        None,
+        Arc::new(object_store_opendal::OpendalStore::new(store.clone())),
+        None,
+    ));
 
-    let csv_opener = CsvOpener::new(csv_conf, store.clone(), CompressionType::Uncompressed);
+    let csv_opener = CsvOpener::new(csv_config, FileCompressionType::UNCOMPRESSED);
 
     let size = store.read(origin_path).await.unwrap().len();
 

src/common/error/Cargo.toml

@@ -12,3 +12,6 @@ http.workspace = true
 snafu.workspace = true
 strum.workspace = true
 tonic.workspace = true
+
+[dev-dependencies]
+common-macro.workspace = true

src/common/error/src/ext.rs

@@ -42,7 +42,7 @@ pub trait ErrorExt: StackError {
                 if let Some(external_error) = error.source() {
                     let external_root = external_error.sources().last().unwrap();
 
-                    if error.to_string().is_empty() {
+                    if error.transparent() {
                         format!("{external_root}")
                     } else {
                         format!("{error}: {external_root}")
@@ -86,6 +86,14 @@ pub trait StackError: std::error::Error {
         }
         result
     }
+
+    /// Indicates whether this error is "transparent", that it delegates its "display" and "source"
+    /// to the underlying error. Could be useful when you are just wrapping some external error,
+    /// **AND** can not or would not provide meaningful contextual info. For example, the
+    /// `DataFusionError`.
+    fn transparent(&self) -> bool {
+        false
+    }
 }
 
 impl<T: ?Sized + StackError> StackError for Arc<T> {

src/common/error/src/status_code.rs

@@ -34,12 +34,14 @@ pub enum StatusCode {
     Internal = 1003,
     /// Invalid arguments.
     InvalidArguments = 1004,
-    /// The task is cancelled.
+    /// The task is cancelled (typically caller-side).
     Cancelled = 1005,
     /// Illegal state, can be exposed to users.
     IllegalState = 1006,
     /// Caused by some error originated from external system.
     External = 1007,
+    /// The request is deadline exceeded (typically server-side).
+    DeadlineExceeded = 1008,
     // ====== End of common status code ================
 
     // ====== Begin of SQL related status code =========
@@ -142,6 +144,7 @@ impl StatusCode {
             | StatusCode::Unexpected
             | StatusCode::InvalidArguments
             | StatusCode::Cancelled
+            | StatusCode::DeadlineExceeded
             | StatusCode::InvalidSyntax
             | StatusCode::DatabaseAlreadyExists
             | StatusCode::PlanQuery
@@ -177,6 +180,7 @@ impl StatusCode {
             | StatusCode::Unexpected
             | StatusCode::Internal
             | StatusCode::Cancelled
+            | StatusCode::DeadlineExceeded
             | StatusCode::IllegalState
             | StatusCode::EngineExecuteQuery
             | StatusCode::StorageUnavailable
@@ -272,6 +276,7 @@ pub fn status_to_tonic_code(status_code: StatusCode) -> Code {
             Code::InvalidArgument
         }
         StatusCode::Cancelled => Code::Cancelled,
+        StatusCode::DeadlineExceeded => Code::DeadlineExceeded,
         StatusCode::TableAlreadyExists
         | StatusCode::TableColumnExists
         | StatusCode::RegionAlreadyExists
@@ -299,6 +304,15 @@ pub fn status_to_tonic_code(status_code: StatusCode) -> Code {
     }
 }
 
+/// Converts tonic [Code] to [StatusCode].
+pub fn convert_tonic_code_to_status_code(code: Code) -> StatusCode {
+    match code {
+        Code::Cancelled => StatusCode::Cancelled,
+        Code::DeadlineExceeded => StatusCode::DeadlineExceeded,
+        _ => StatusCode::Internal,
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use strum::IntoEnumIterator;

src/common/error/tests/ext.rs

@@ -0,0 +1,115 @@
+// 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::any::Any;
+
+use common_error::ext::{ErrorExt, PlainError, StackError};
+use common_error::status_code::StatusCode;
+use common_macro::stack_trace_debug;
+use snafu::{Location, ResultExt, Snafu};
+
+#[derive(Snafu)]
+#[stack_trace_debug]
+enum MyError {
+    #[snafu(display(r#"A normal error with "display" attribute, message "{}""#, message))]
+    Normal {
+        message: String,
+        #[snafu(source)]
+        error: PlainError,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(transparent)]
+    Transparent {
+        #[snafu(source)]
+        error: PlainError,
+        #[snafu(implicit)]
+        location: Location,
+    },
+}
+
+impl ErrorExt for MyError {
+    fn status_code(&self) -> StatusCode {
+        StatusCode::Unexpected
+    }
+
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+}
+
+fn normal_error() -> Result<(), MyError> {
+    let plain_error = PlainError::new("<root cause>".to_string(), StatusCode::Unexpected);
+    Err(plain_error).context(NormalSnafu { message: "blabla" })
+}
+
+fn transparent_error() -> Result<(), MyError> {
+    let plain_error = PlainError::new("<root cause>".to_string(), StatusCode::Unexpected);
+    Err(plain_error)?
+}
+
+#[test]
+fn test_output_msg() {
+    let result = normal_error();
+    assert_eq!(
+        result.unwrap_err().output_msg(),
+        r#"A normal error with "display" attribute, message "blabla": <root cause>"#
+    );
+
+    let result = transparent_error();
+    assert_eq!(result.unwrap_err().output_msg(), "<root cause>");
+}
+
+#[test]
+fn test_to_string() {
+    let result = normal_error();
+    assert_eq!(
+        result.unwrap_err().to_string(),
+        r#"A normal error with "display" attribute, message "blabla""#
+    );
+
+    let result = transparent_error();
+    assert_eq!(result.unwrap_err().to_string(), "<root cause>");
+}
+
+#[test]
+fn test_debug_format() {
+    let result = normal_error();
+    let debug_output = format!("{:?}", result.unwrap_err());
+    let normalized_output = debug_output.replace('\\', "/");
+    assert_eq!(
+        normalized_output,
+        r#"0: A normal error with "display" attribute, message "blabla", at src/common/error/tests/ext.rs:55:22
+1: PlainError { msg: "<root cause>", status_code: Unexpected }"#
+    );
+
+    let result = transparent_error();
+    let debug_output = format!("{:?}", result.unwrap_err());
+    let normalized_output = debug_output.replace('\\', "/");
+    assert_eq!(
+        normalized_output,
+        r#"0: <transparent>, at src/common/error/tests/ext.rs:60:5
+1: PlainError { msg: "<root cause>", status_code: Unexpected }"#
+    );
+}
+
+#[test]
+fn test_transparent_flag() {
+    let result = normal_error();
+    assert!(!result.unwrap_err().transparent());
+
+    let result = transparent_error();
+    assert!(result.unwrap_err().transparent());
+}

src/common/function/Cargo.toml

@@ -8,6 +8,7 @@ license.workspace = true
 workspace = true
 
 [features]
+testing = []
 default = ["geo"]
 geo = ["geohash", "h3o", "s2", "wkt", "geo-types", "dep:geo"]
 
@@ -17,6 +18,7 @@ api.workspace = true
 arc-swap = "1.0"
 async-trait.workspace = true
 bincode = "1.3"
+catalog.workspace = true
 chrono.workspace = true
 common-base.workspace = true
 common-catalog.workspace = true

src/common/function/src/admin/migrate_region.rs

@@ -25,12 +25,13 @@ use session::context::QueryContextRef;
 use crate::handlers::ProcedureServiceHandlerRef;
 use crate::helper::cast_u64;
 
-const DEFAULT_TIMEOUT_SECS: u64 = 30;
+/// The default timeout for migrate region procedure.
+const DEFAULT_TIMEOUT_SECS: u64 = 300;
 
 /// A function to migrate a region from source peer to target peer.
 /// Returns the submitted procedure id if success. Only available in cluster mode.
 ///
-/// - `migrate_region(region_id, from_peer, to_peer)`, with timeout(30 seconds).
+/// - `migrate_region(region_id, from_peer, to_peer)`, with timeout(300 seconds).
 /// - `migrate_region(region_id, from_peer, to_peer, timeout(secs))`.
 ///
 /// The parameters:

src/common/function/src/function.rs

@@ -32,7 +32,7 @@ pub struct FunctionContext {
 
 impl FunctionContext {
     /// Create a mock [`FunctionContext`] for test.
-    #[cfg(test)]
+    #[cfg(any(test, feature = "testing"))]
     pub fn mock() -> Self {
         Self {
             query_ctx: QueryContextBuilder::default().build().into(),

src/common/function/src/handlers.rs

@@ -15,6 +15,7 @@
 use std::sync::Arc;
 
 use async_trait::async_trait;
+use catalog::CatalogManagerRef;
 use common_base::AffectedRows;
 use common_meta::rpc::procedure::{
     AddRegionFollowerRequest, MigrateRegionRequest, ProcedureStateResponse,
@@ -72,6 +73,9 @@ pub trait ProcedureServiceHandler: Send + Sync {
 
     /// Remove a region follower from a region.
     async fn remove_region_follower(&self, request: RemoveRegionFollowerRequest) -> Result<()>;
+
+    /// Get the catalog manager
+    fn catalog_manager(&self) -> &CatalogManagerRef;
 }
 
 /// This flow service handler is only use for flush flow for now.

src/common/function/src/scalars/geo/encoding.rs

@@ -27,7 +27,7 @@ use datatypes::value::{ListValue, Value};
 use datatypes::vectors::VectorRef;
 use snafu::{ensure, ResultExt};
 
-use super::helpers::{ensure_columns_len, ensure_columns_n};
+use crate::scalars::geo::helpers::{ensure_columns_len, ensure_columns_n};
 
 /// Accumulator of lat, lng, timestamp tuples
 #[derive(Debug)]

src/common/function/src/scalars/geo/h3.rs

@@ -31,8 +31,8 @@ use h3o::{CellIndex, LatLng, Resolution};
 use once_cell::sync::Lazy;
 use snafu::ResultExt;
 
-use super::helpers::{ensure_and_coerce, ensure_columns_len, ensure_columns_n};
 use crate::function::{Function, FunctionContext};
+use crate::scalars::geo::helpers::{ensure_and_coerce, ensure_columns_len, ensure_columns_n};
 
 static CELL_TYPES: Lazy<Vec<ConcreteDataType>> = Lazy::new(|| {
     vec![

src/common/function/src/scalars/geo/measure.rs

@@ -26,9 +26,9 @@ use geo::{Area, Distance, Haversine};
 use geo_types::Geometry;
 use snafu::ResultExt;
 
-use super::helpers::{ensure_columns_len, ensure_columns_n};
-use super::wkt::parse_wkt;
 use crate::function::{Function, FunctionContext};
+use crate::scalars::geo::helpers::{ensure_columns_len, ensure_columns_n};
+use crate::scalars::geo::wkt::parse_wkt;
 
 /// Return WGS84(SRID: 4326) euclidean distance between two geometry object, in degree
 #[derive(Clone, Debug, Default, Display)]

src/common/function/src/scalars/geo/relation.rs

@@ -23,9 +23,9 @@ use geo::algorithm::contains::Contains;
 use geo::algorithm::intersects::Intersects;
 use geo::algorithm::within::Within;
 
-use super::helpers::{ensure_columns_len, ensure_columns_n};
-use super::wkt::parse_wkt;
 use crate::function::{Function, FunctionContext};
+use crate::scalars::geo::helpers::{ensure_columns_len, ensure_columns_n};
+use crate::scalars::geo::wkt::parse_wkt;
 
 /// Test if spatial relationship: contains
 #[derive(Clone, Debug, Default, Display)]

src/common/function/src/scalars/geo/wkt.rs

@@ -26,8 +26,8 @@ use once_cell::sync::Lazy;
 use snafu::ResultExt;
 use wkt::{ToWkt, TryFromWkt};
 
-use super::helpers::{ensure_columns_len, ensure_columns_n};
 use crate::function::{Function, FunctionContext};
+use crate::scalars::geo::helpers::{ensure_columns_len, ensure_columns_n};
 
 static COORDINATE_TYPES: Lazy<Vec<ConcreteDataType>> = Lazy::new(|| {
     vec![

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/function/src/scalars/udf.rs

@@ -163,7 +163,7 @@ mod tests {
         ];
 
         let args = ScalarFunctionArgs {
-            args: &args,
+            args,
             number_rows: 4,
             return_type: &ConcreteDataType::boolean_datatype().as_arrow_type(),
         };

src/common/function/src/state.rs

@@ -28,12 +28,13 @@ pub struct FunctionState {
 
 impl FunctionState {
     /// Create a mock [`FunctionState`] for test.
-    #[cfg(test)]
+    #[cfg(any(test, feature = "testing"))]
     pub fn mock() -> Self {
         use std::sync::Arc;
 
         use api::v1::meta::ProcedureStatus;
         use async_trait::async_trait;
+        use catalog::CatalogManagerRef;
         use common_base::AffectedRows;
         use common_meta::rpc::procedure::{
             AddRegionFollowerRequest, MigrateRegionRequest, ProcedureStateResponse,
@@ -80,6 +81,10 @@ impl FunctionState {
             ) -> Result<()> {
                 Ok(())
             }
+
+            fn catalog_manager(&self) -> &CatalogManagerRef {
+                unimplemented!()
+            }
         }
 
         #[async_trait]