chore: comment

Signed-off-by: discord9 <discord9@163.com>
chore: per review
2025-12-25 07:30:02 +00:00 · 2025-06-12 17:44:54 +08:00 · 2025-06-12 17:44:54 +08:00 · 2025-06-12 17:44:54 +08:00 · 2025-06-12 06:55:22 +00:00 · 2025-06-12 03:18:26 +00:00
172 changed files with 3730 additions and 1531 deletions
--- a/.github/actions/setup-greptimedb-cluster/action.yml
+++ b/.github/actions/setup-greptimedb-cluster/action.yml
@@ -10,13 +10,13 @@ inputs:
  meta-replicas:
    default: 2
    description: "Number of Metasrv replicas"
-  image-registry: 
+  image-registry:
    default: "docker.io"
    description: "Image registry"
-  image-repository: 
+  image-repository:
    default: "greptime/greptimedb"
    description: "Image repository"
-  image-tag: 
+  image-tag:
    default: "latest"
    description: 'Image tag'
  etcd-endpoints:
@@ -32,12 +32,12 @@ runs:
  steps:
  - name: Install GreptimeDB operator
    uses: nick-fields/retry@v3
-    with: 
+    with:
      timeout_minutes: 3
      max_attempts: 3
      shell: bash
      command: |
-        helm repo add greptime https://greptimeteam.github.io/helm-charts/ 
+        helm repo add greptime https://greptimeteam.github.io/helm-charts/
        helm repo update
        helm upgrade \
          --install \
@@ -48,10 +48,10 @@ runs:
          --wait-for-jobs
  - name: Install GreptimeDB cluster
    shell: bash
-    run: | 
+    run: |
      helm upgrade \
        --install my-greptimedb \
-        --set meta.etcdEndpoints=${{ inputs.etcd-endpoints }} \
+        --set meta.backendStorage.etcd.endpoints=${{ inputs.etcd-endpoints }} \
        --set meta.enableRegionFailover=${{ inputs.enable-region-failover }} \
        --set image.registry=${{ inputs.image-registry }} \
        --set image.repository=${{ inputs.image-repository }}  \
@@ -59,7 +59,7 @@ runs:
        --set base.podTemplate.main.resources.requests.cpu=50m \
        --set base.podTemplate.main.resources.requests.memory=256Mi \
        --set base.podTemplate.main.resources.limits.cpu=2000m \
-        --set base.podTemplate.main.resources.limits.memory=2Gi \
+        --set base.podTemplate.main.resources.limits.memory=3Gi \
        --set frontend.replicas=${{ inputs.frontend-replicas }} \
        --set datanode.replicas=${{ inputs.datanode-replicas }} \
        --set meta.replicas=${{ inputs.meta-replicas }} \
@@ -72,7 +72,7 @@ runs:
  - name: Wait for GreptimeDB
    shell: bash
    run: |
-      while true; do 
+      while true; do
        PHASE=$(kubectl -n my-greptimedb get gtc my-greptimedb -o jsonpath='{.status.clusterPhase}')
        if [ "$PHASE" == "Running" ]; then
          echo "Cluster is ready"
@@ -86,10 +86,10 @@ runs:
  - name: Print GreptimeDB info
    if: always()
    shell: bash
-    run: | 
+    run: |
      kubectl get all --show-labels -n my-greptimedb
  - name: Describe Nodes
    if: always()
    shell: bash
-    run: | 
+    run: |
      kubectl describe nodes
--- a/.github/scripts/deploy-greptimedb.sh
+++ b/.github/scripts/deploy-greptimedb.sh
@@ -68,7 +68,7 @@ function deploy_greptimedb_cluster() {

  helm install "$cluster_name" greptime/greptimedb-cluster \
    --set image.tag="$GREPTIMEDB_IMAGE_TAG" \
-    --set meta.etcdEndpoints="etcd.$install_namespace:2379" \
+    --set meta.backendStorage.etcd.endpoints="etcd.$install_namespace:2379" \
    -n "$install_namespace"

  # Wait for greptimedb cluster to be ready.
@@ -103,7 +103,7 @@ function deploy_greptimedb_cluster_with_s3_storage() {

  helm install "$cluster_name" greptime/greptimedb-cluster -n "$install_namespace" \
    --set image.tag="$GREPTIMEDB_IMAGE_TAG" \
-    --set meta.etcdEndpoints="etcd.$install_namespace:2379" \
+    --set meta.backendStorage.etcd.endpoints="etcd.$install_namespace:2379" \
    --set storage.s3.bucket="$AWS_CI_TEST_BUCKET" \
    --set storage.s3.region="$AWS_REGION" \
    --set storage.s3.root="$DATA_ROOT" \
--- a/.github/scripts/update-helm-charts-version.sh
+++ b/.github/scripts/update-helm-charts-version.sh
@@ -30,7 +30,7 @@ update_helm_charts_version() {

  # Commit the changes.
  git add .
-  git commit -m "chore: Update GreptimeDB version to ${VERSION}"
+  git commit -s -m "chore: Update GreptimeDB version to ${VERSION}"
  git push origin $BRANCH_NAME

  # Create a Pull Request.
--- a/.github/scripts/update-homebrew-greptme-version.sh
+++ b/.github/scripts/update-homebrew-greptme-version.sh
@@ -26,7 +26,7 @@ update_homebrew_greptime_version() {

  # Commit the changes.
  git add .
-  git commit -m "chore: Update GreptimeDB version to ${VERSION}"
+  git commit -s -m "chore: Update GreptimeDB version to ${VERSION}"
  git push origin $BRANCH_NAME

  # Create a Pull Request.
--- a/.github/workflows/develop.yml
+++ b/.github/workflows/develop.yml
@@ -250,6 +250,11 @@ jobs:
          name: unstable-fuzz-logs
          path: /tmp/unstable-greptime/
          retention-days: 3
+      - name: Describe pods
+        if: failure()
+        shell: bash
+        run: |
+          kubectl describe pod -n my-greptimedb

  build-greptime-ci:
    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
@@ -405,6 +410,11 @@ jobs:
        shell: bash
        run: |
          kubectl describe nodes
+      - name: Describe pod
+        if: failure()
+        shell: bash
+        run: |
+          kubectl describe pod -n my-greptimedb
      - name: Export kind logs
        if: failure()
        shell: bash
@@ -554,6 +564,11 @@ jobs:
        shell: bash
        run: |
          kubectl describe nodes
+      - name: Describe pods
+        if: failure()
+        shell: bash
+        run: |
+          kubectl describe pod -n my-greptimedb
      - name: Export kind logs
        if: failure()
        shell: bash
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1623,6 +1623,7 @@ dependencies = [
 "chrono",
 "common-catalog",
 "common-error",
+ "common-frontend",
 "common-macro",
 "common-meta",
 "common-procedure",
@@ -2293,8 +2294,14 @@ version = "0.15.0"
 dependencies = [
 "async-trait",
 "common-error",
+ "common-grpc",
 "common-macro",
+ "common-meta",
+ "greptime-proto",
+ "meta-client",
 "snafu 0.8.5",
+ "tokio",
+ "tonic 0.12.3",
 ]

 [[package]]
@@ -3252,7 +3259,7 @@ dependencies = [
 [[package]]
 name = "datafusion"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "arrow 54.2.1",
 "arrow-array 54.2.1",
@@ -3303,7 +3310,7 @@ dependencies = [
 [[package]]
 name = "datafusion-catalog"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "arrow 54.2.1",
 "async-trait",
@@ -3323,7 +3330,7 @@ dependencies = [
 [[package]]
 name = "datafusion-catalog-listing"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "arrow 54.2.1",
 "arrow-schema 54.3.1",
@@ -3346,7 +3353,7 @@ dependencies = [
 [[package]]
 name = "datafusion-common"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "ahash 0.8.11",
 "arrow 54.2.1",
@@ -3371,7 +3378,7 @@ dependencies = [
 [[package]]
 name = "datafusion-common-runtime"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "log",
 "tokio",
@@ -3380,12 +3387,12 @@ dependencies = [
 [[package]]
 name = "datafusion-doc"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"

 [[package]]
 name = "datafusion-execution"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "arrow 54.2.1",
 "dashmap",
@@ -3403,7 +3410,7 @@ dependencies = [
 [[package]]
 name = "datafusion-expr"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "arrow 54.2.1",
 "chrono",
@@ -3423,7 +3430,7 @@ dependencies = [
 [[package]]
 name = "datafusion-expr-common"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "arrow 54.2.1",
 "datafusion-common",
@@ -3434,7 +3441,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "arrow 54.2.1",
 "arrow-buffer 54.3.1",
@@ -3463,7 +3470,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-aggregate"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "ahash 0.8.11",
 "arrow 54.2.1",
@@ -3484,7 +3491,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-aggregate-common"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "ahash 0.8.11",
 "arrow 54.2.1",
@@ -3496,7 +3503,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-nested"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "arrow 54.2.1",
 "arrow-array 54.2.1",
@@ -3518,7 +3525,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-table"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "arrow 54.2.1",
 "async-trait",
@@ -3533,7 +3540,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-window"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "datafusion-common",
 "datafusion-doc",
@@ -3549,7 +3556,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-window-common"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "datafusion-common",
 "datafusion-physical-expr-common",
@@ -3558,7 +3565,7 @@ dependencies = [
 [[package]]
 name = "datafusion-macros"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "datafusion-expr",
 "quote",
@@ -3568,7 +3575,7 @@ dependencies = [
 [[package]]
 name = "datafusion-optimizer"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "arrow 54.2.1",
 "chrono",
@@ -3586,7 +3593,7 @@ dependencies = [
 [[package]]
 name = "datafusion-physical-expr"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "ahash 0.8.11",
 "arrow 54.2.1",
@@ -3609,7 +3616,7 @@ dependencies = [
 [[package]]
 name = "datafusion-physical-expr-common"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "ahash 0.8.11",
 "arrow 54.2.1",
@@ -3622,7 +3629,7 @@ dependencies = [
 [[package]]
 name = "datafusion-physical-optimizer"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "arrow 54.2.1",
 "arrow-schema 54.3.1",
@@ -3643,7 +3650,7 @@ dependencies = [
 [[package]]
 name = "datafusion-physical-plan"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "ahash 0.8.11",
 "arrow 54.2.1",
@@ -3673,7 +3680,7 @@ dependencies = [
 [[package]]
 name = "datafusion-sql"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "arrow 54.2.1",
 "arrow-array 54.2.1",
@@ -3691,7 +3698,7 @@ dependencies = [
 [[package]]
 name = "datafusion-substrait"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
 dependencies = [
 "async-recursion",
 "async-trait",
@@ -4511,9 +4518,9 @@ dependencies = [

 [[package]]
 name = "flate2"
-version = "1.1.1"
+version = "1.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7ced92e76e966ca2fd84c8f7aa01a4aea65b0eb6648d72f7c8f3e2764a67fece"
+checksum = "4a3d7db9596fecd151c5f638c0ee5d5bd487b6e0ea232e5dc96d5250f6f94b1d"
 dependencies = [
 "crc32fast",
 "libz-rs-sys",
@@ -4701,6 +4708,7 @@ dependencies = [
 "common-config",
 "common-datasource",
 "common-error",
+ "common-frontend",
 "common-function",
 "common-grpc",
 "common-macro",
@@ -5133,7 +5141,7 @@ dependencies = [
 [[package]]
 name = "greptime-proto"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=454c52634c3bac27de10bf0d85d5533eed1cf03f#454c52634c3bac27de10bf0d85d5533eed1cf03f"
+source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=5f6119ac7952878d39dcde0343c4bf828d18ffc8#5f6119ac7952878d39dcde0343c4bf828d18ffc8"
 dependencies = [
 "prost 0.13.5",
 "serde",
@@ -5146,9 +5154,9 @@ dependencies = [

 [[package]]
 name = "grok"
-version = "2.0.0"
+version = "2.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "273797968160270573071022613fc4aa28b91fe68f3eef6c96a1b2a1947ddfbd"
+checksum = "6c52724b609896f661a3f4641dd3a44dc602958ef615857c12d00756b4e9355b"
 dependencies = [
 "glob",
 "onig",
@@ -6716,9 +6724,9 @@ dependencies = [

 [[package]]
 name = "libz-rs-sys"
-version = "0.5.0"
+version = "0.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6489ca9bd760fe9642d7644e827b0c9add07df89857b0416ee15c1cc1a3b8c5a"
+checksum = "172a788537a2221661b480fee8dc5f96c580eb34fa88764d3205dc356c7e4221"
 dependencies = [
 "zlib-rs",
 ]
@@ -8885,9 +8893,9 @@ dependencies = [

 [[package]]
 name = "pgwire"
-version = "0.30.1"
+version = "0.30.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ec79ee18e6cafde8698885646780b967ecc905120798b8359dd0da64f9688e89"
+checksum = "4ca6c26b25be998208a13ff2f0c55b567363f34675410e6d6f1c513a150583fd"
 dependencies = [
 "async-trait",
 "bytes",
@@ -9664,9 +9672,9 @@ dependencies = [

 [[package]]
 name = "psl"
-version = "2.1.112"
+version = "2.1.119"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c6b4c497a0c6bfb466f75167c728b1a861b0cdc39de9c35b877208a270a9590"
+checksum = "d0e49aa528239f2ca13ad87387977c208e59c3fb8c437609f95f1b3898ec6ef1"
 dependencies = [
 "psl-types",
 ]
@@ -11135,6 +11143,7 @@ dependencies = [
 "common-catalog",
 "common-config",
 "common-error",
+ "common-frontend",
 "common-grpc",
 "common-macro",
 "common-mem-prof",
@@ -14701,9 +14710,9 @@ dependencies = [

 [[package]]
 name = "zlib-rs"
-version = "0.5.0"
+version = "0.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "868b928d7949e09af2f6086dfc1e01936064cc7a819253bce650d4e2a2d63ba8"
+checksum = "626bd9fa9734751fc50d6060752170984d7053f5a39061f524cda68023d4db8a"

 [[package]]
 name = "zstd"
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -116,15 +116,15 @@ clap = { version = "4.4", features = ["derive"] }
 config = "0.13.0"
 crossbeam-utils = "0.8"
 dashmap = "6.1"
-datafusion = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
-datafusion-common = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
-datafusion-expr = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
-datafusion-functions = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
-datafusion-optimizer = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
-datafusion-physical-expr = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
-datafusion-physical-plan = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
-datafusion-sql = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
-datafusion-substrait = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
+datafusion = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
+datafusion-common = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
+datafusion-expr = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
+datafusion-functions = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
+datafusion-optimizer = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
+datafusion-physical-expr = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
+datafusion-physical-plan = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
+datafusion-sql = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
+datafusion-substrait = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
 deadpool = "0.12"
 deadpool-postgres = "0.14"
 derive_builder = "0.20"
@@ -133,7 +133,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 = "454c52634c3bac27de10bf0d85d5533eed1cf03f" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "5f6119ac7952878d39dcde0343c4bf828d18ffc8" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"
--- a/config/config.md
+++ b/config/config.md
@@ -195,13 +195,13 @@
 | `slow_query.record_type` | String | Unset | The record type of slow queries. It can be `system_table` or `log`. |
 | `slow_query.threshold` | String | Unset | The threshold of slow query. |
 | `slow_query.sample_ratio` | Float | Unset | The sampling ratio of slow query log. The value should be in the range of (0, 1]. |
-| `export_metrics` | -- | -- | The datanode can export its metrics and send to Prometheus compatible service (e.g. send to `greptimedb` itself) from remote-write API.<br/>This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape. |
+| `export_metrics` | -- | -- | The standalone can export its metrics and send to Prometheus compatible service (e.g. `greptimedb`) from remote-write API.<br/>This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape. |
 | `export_metrics.enable` | Bool | `false` | whether enable export metrics. |
 | `export_metrics.write_interval` | String | `30s` | The interval of export metrics. |
 | `export_metrics.self_import` | -- | -- | For `standalone` mode, `self_import` is recommended to collect metrics generated by itself<br/>You must create the database before enabling it. |
 | `export_metrics.self_import.db` | String | Unset | -- |
 | `export_metrics.remote_write` | -- | -- | -- |
-| `export_metrics.remote_write.url` | String | `""` | The url the metrics send to. The url example can be: `http://127.0.0.1:4000/v1/prometheus/write?db=greptime_metrics`. |
+| `export_metrics.remote_write.url` | String | `""` | The prometheus remote write endpoint that the metrics send to. The url example can be: `http://127.0.0.1:4000/v1/prometheus/write?db=greptime_metrics`. |
 | `export_metrics.remote_write.headers` | InlineTable | -- | HTTP headers of Prometheus remote-write carry. |
 | `tracing` | -- | -- | The tracing options. Only effect when compiled with `tokio-console` feature. |
 | `tracing.tokio_console_addr` | String | Unset | The tokio console address. |
@@ -232,6 +232,7 @@
 | `grpc.bind_addr` | String | `127.0.0.1:4001` | The address to bind the gRPC server. |
 | `grpc.server_addr` | String | `127.0.0.1:4001` | The address advertised to the metasrv, and used for connections from outside the host.<br/>If left empty or unset, the server will automatically use the IP address of the first network interface<br/>on the host, with the same port number as the one specified in `grpc.bind_addr`. |
 | `grpc.runtime_size` | Integer | `8` | The number of server worker threads. |
+| `grpc.flight_compression` | String | `arrow_ipc` | Compression mode for frontend side Arrow IPC service. Available options:<br/>- `none`: disable all compression<br/>- `transport`: only enable gRPC transport compression (zstd)<br/>- `arrow_ipc`: only enable Arrow IPC compression (lz4)<br/>- `all`: enable all compression.<br/>Default to `none` |
 | `grpc.tls` | -- | -- | gRPC server TLS options, see `mysql.tls` section. |
 | `grpc.tls.mode` | String | `disable` | TLS mode. |
 | `grpc.tls.cert_path` | String | Unset | Certificate file path. |
@@ -298,13 +299,11 @@
 | `slow_query.threshold` | String | `30s` | The threshold of slow query. It can be human readable time string, for example: `10s`, `100ms`, `1s`. |
 | `slow_query.sample_ratio` | Float | `1.0` | The sampling ratio of slow query log. The value should be in the range of (0, 1]. For example, `0.1` means 10% of the slow queries will be logged and `1.0` means all slow queries will be logged. |
 | `slow_query.ttl` | String | `30d` | The TTL of the `slow_queries` system table. Default is `30d` when `record_type` is `system_table`. |
-| `export_metrics` | -- | -- | The datanode can export its metrics and send to Prometheus compatible service (e.g. send to `greptimedb` itself) from remote-write API.<br/>This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape. |
+| `export_metrics` | -- | -- | The frontend can export its metrics and send to Prometheus compatible service (e.g. `greptimedb` itself) from remote-write API.<br/>This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape. |
 | `export_metrics.enable` | Bool | `false` | whether enable export metrics. |
 | `export_metrics.write_interval` | String | `30s` | The interval of export metrics. |
-| `export_metrics.self_import` | -- | -- | For `standalone` mode, `self_import` is recommend to collect metrics generated by itself<br/>You must create the database before enabling it. |
-| `export_metrics.self_import.db` | String | Unset | -- |
 | `export_metrics.remote_write` | -- | -- | -- |
-| `export_metrics.remote_write.url` | String | `""` | The url the metrics send to. The url example can be: `http://127.0.0.1:4000/v1/prometheus/write?db=greptime_metrics`. |
+| `export_metrics.remote_write.url` | String | `""` | The prometheus remote write endpoint that the metrics send to. The url example can be: `http://127.0.0.1:4000/v1/prometheus/write?db=greptime_metrics`. |
 | `export_metrics.remote_write.headers` | InlineTable | -- | HTTP headers of Prometheus remote-write carry. |
 | `tracing` | -- | -- | The tracing options. Only effect when compiled with `tokio-console` feature. |
 | `tracing.tokio_console_addr` | String | Unset | The tokio console address. |
@@ -315,11 +314,9 @@
 | Key | Type | Default | Descriptions |
 | --- | -----| ------- | ----------- |
 | `data_home` | String | `./greptimedb_data` | The working home directory. |
-| `bind_addr` | String | `127.0.0.1:3002` | The bind address of metasrv. |
-| `server_addr` | String | `127.0.0.1:3002` | The communication server address for the frontend and datanode to connect to metasrv.<br/>If left empty or unset, the server will automatically use the IP address of the first network interface<br/>on the host, with the same port number as the one specified in `bind_addr`. |
 | `store_addrs` | Array | -- | Store server address default to etcd store.<br/>For postgres store, the format is:<br/>"password=password dbname=postgres user=postgres host=localhost port=5432"<br/>For etcd store, the format is:<br/>"127.0.0.1:2379" |
 | `store_key_prefix` | String | `""` | If it's not empty, the metasrv will store all data with this key prefix. |
-| `backend` | String | `etcd_store` | The datastore for meta server.<br/>Available values:<br/>- `etcd_store` (default value)<br/>- `memory_store`<br/>- `postgres_store` |
+| `backend` | String | `etcd_store` | The datastore for meta server.<br/>Available values:<br/>- `etcd_store` (default value)<br/>- `memory_store`<br/>- `postgres_store`<br/>- `mysql_store` |
 | `meta_table_name` | String | `greptime_metakv` | Table name in RDS to store metadata. Effect when using a RDS kvbackend.<br/>**Only used when backend is `postgres_store`.** |
 | `meta_election_lock_id` | Integer | `1` | Advisory lock id in PostgreSQL for election. Effect when using PostgreSQL as kvbackend<br/>Only used when backend is `postgres_store`. |
 | `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". |
@@ -331,6 +328,12 @@
 | `runtime` | -- | -- | The runtime options. |
 | `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
 | `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute the runtime for global write operations. |
+| `grpc` | -- | -- | The gRPC server options. |
+| `grpc.bind_addr` | String | `127.0.0.1:3002` | The address to bind the gRPC server. |
+| `grpc.server_addr` | String | `127.0.0.1:3002` | The communication server address for the frontend and datanode to connect to metasrv.<br/>If left empty or unset, the server will automatically use the IP address of the first network interface<br/>on the host, with the same port number as the one specified in `bind_addr`. |
+| `grpc.runtime_size` | Integer | `8` | The number of server worker threads. |
+| `grpc.max_recv_message_size` | String | `512MB` | The maximum receive message size for gRPC server. |
+| `grpc.max_send_message_size` | String | `512MB` | The maximum send message size for gRPC server. |
 | `http` | -- | -- | The HTTP server options. |
 | `http.addr` | String | `127.0.0.1:4000` | The address to bind the HTTP server. |
 | `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout. |
@@ -372,13 +375,11 @@
 | `logging.max_log_files` | Integer | `720` | The maximum amount of log files. |
 | `logging.tracing_sample_ratio` | -- | -- | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
 | `logging.tracing_sample_ratio.default_ratio` | Float | `1.0` | -- |
-| `export_metrics` | -- | -- | The datanode can export its metrics and send to Prometheus compatible service (e.g. send to `greptimedb` itself) from remote-write API.<br/>This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape. |
+| `export_metrics` | -- | -- | The metasrv can export its metrics and send to Prometheus compatible service (e.g. `greptimedb` itself) from remote-write API.<br/>This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape. |
 | `export_metrics.enable` | Bool | `false` | whether enable export metrics. |
 | `export_metrics.write_interval` | String | `30s` | The interval of export metrics. |
-| `export_metrics.self_import` | -- | -- | For `standalone` mode, `self_import` is recommend to collect metrics generated by itself<br/>You must create the database before enabling it. |
-| `export_metrics.self_import.db` | String | Unset | -- |
 | `export_metrics.remote_write` | -- | -- | -- |
-| `export_metrics.remote_write.url` | String | `""` | The url the metrics send to. The url example can be: `http://127.0.0.1:4000/v1/prometheus/write?db=greptime_metrics`. |
+| `export_metrics.remote_write.url` | String | `""` | The prometheus remote write endpoint that the metrics send to. The url example can be: `http://127.0.0.1:4000/v1/prometheus/write?db=greptime_metrics`. |
 | `export_metrics.remote_write.headers` | InlineTable | -- | HTTP headers of Prometheus remote-write carry. |
 | `tracing` | -- | -- | The tracing options. Only effect when compiled with `tokio-console` feature. |
 | `tracing.tokio_console_addr` | String | Unset | The tokio console address. |
@@ -404,6 +405,7 @@
 | `grpc.runtime_size` | Integer | `8` | The number of server worker threads. |
 | `grpc.max_recv_message_size` | String | `512MB` | The maximum receive message size for gRPC server. |
 | `grpc.max_send_message_size` | String | `512MB` | The maximum send message size for gRPC server. |
+| `grpc.flight_compression` | String | `arrow_ipc` | Compression mode for datanode side Arrow IPC service. Available options:<br/>- `none`: disable all compression<br/>- `transport`: only enable gRPC transport compression (zstd)<br/>- `arrow_ipc`: only enable Arrow IPC compression (lz4)<br/>- `all`: enable all compression.<br/>Default to `none` |
 | `grpc.tls` | -- | -- | gRPC server TLS options, see `mysql.tls` section. |
 | `grpc.tls.mode` | String | `disable` | TLS mode. |
 | `grpc.tls.cert_path` | String | Unset | Certificate file path. |
@@ -536,13 +538,11 @@
 | `logging.max_log_files` | Integer | `720` | The maximum amount of log files. |
 | `logging.tracing_sample_ratio` | -- | -- | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
 | `logging.tracing_sample_ratio.default_ratio` | Float | `1.0` | -- |
-| `export_metrics` | -- | -- | The datanode can export its metrics and send to Prometheus compatible service (e.g. send to `greptimedb` itself) from remote-write API.<br/>This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape. |
+| `export_metrics` | -- | -- | The datanode can export its metrics and send to Prometheus compatible service (e.g. `greptimedb` itself) from remote-write API.<br/>This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape. |
 | `export_metrics.enable` | Bool | `false` | whether enable export metrics. |
 | `export_metrics.write_interval` | String | `30s` | The interval of export metrics. |
-| `export_metrics.self_import` | -- | -- | For `standalone` mode, `self_import` is recommend to collect metrics generated by itself<br/>You must create the database before enabling it. |
-| `export_metrics.self_import.db` | String | Unset | -- |
 | `export_metrics.remote_write` | -- | -- | -- |
-| `export_metrics.remote_write.url` | String | `""` | The url the metrics send to. The url example can be: `http://127.0.0.1:4000/v1/prometheus/write?db=greptime_metrics`. |
+| `export_metrics.remote_write.url` | String | `""` | The prometheus remote write endpoint that the metrics send to. The url example can be: `http://127.0.0.1:4000/v1/prometheus/write?db=greptime_metrics`. |
 | `export_metrics.remote_write.headers` | InlineTable | -- | HTTP headers of Prometheus remote-write carry. |
 | `tracing` | -- | -- | The tracing options. Only effect when compiled with `tokio-console` feature. |
 | `tracing.tokio_console_addr` | String | Unset | The tokio console address. |
--- a/config/datanode.example.toml
+++ b/config/datanode.example.toml
@@ -44,6 +44,13 @@ runtime_size = 8
 max_recv_message_size = "512MB"
 ## The maximum send message size for gRPC server.
 max_send_message_size = "512MB"
+## Compression mode for datanode side Arrow IPC service. Available options:
+## - `none`: disable all compression
+## - `transport`: only enable gRPC transport compression (zstd)
+## - `arrow_ipc`: only enable Arrow IPC compression (lz4)
+## - `all`: enable all compression.
+## Default to `none`
+flight_compression = "arrow_ipc"

 ## gRPC server TLS options, see `mysql.tls` section.
 [grpc.tls]
@@ -635,24 +642,16 @@ max_log_files = 720
 [logging.tracing_sample_ratio]
 default_ratio = 1.0

-## The datanode can export its metrics and send to Prometheus compatible service (e.g. send to `greptimedb` itself) from remote-write API.
+## The datanode can export its metrics and send to Prometheus compatible service (e.g. `greptimedb` itself) from remote-write API.
 ## This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape.
 [export_metrics]
-
 ## whether enable export metrics.
 enable = false
-
 ## The interval of export metrics.
 write_interval = "30s"

-## For `standalone` mode, `self_import` is recommend to collect metrics generated by itself
-## You must create the database before enabling it.
-[export_metrics.self_import]
-## @toml2docs:none-default
-db = "greptime_metrics"
-
 [export_metrics.remote_write]
-## The url the metrics send to. The url example can be: `http://127.0.0.1:4000/v1/prometheus/write?db=greptime_metrics`.
+## The prometheus remote write endpoint that the metrics send to. The url example can be: `http://127.0.0.1:4000/v1/prometheus/write?db=greptime_metrics`.
 url = ""

 ## HTTP headers of Prometheus remote-write carry.
--- a/config/frontend.example.toml
+++ b/config/frontend.example.toml
@@ -54,6 +54,13 @@ bind_addr = "127.0.0.1:4001"
 server_addr = "127.0.0.1:4001"
 ## The number of server worker threads.
 runtime_size = 8
+## Compression mode for frontend side Arrow IPC service. Available options:
+## - `none`: disable all compression
+## - `transport`: only enable gRPC transport compression (zstd)
+## - `arrow_ipc`: only enable Arrow IPC compression (lz4)
+## - `all`: enable all compression.
+## Default to `none`
+flight_compression = "arrow_ipc"

 ## gRPC server TLS options, see `mysql.tls` section.
 [grpc.tls]
@@ -247,24 +254,16 @@ sample_ratio = 1.0
 ## The TTL of the `slow_queries` system table. Default is `30d` when `record_type` is `system_table`.
 ttl = "30d"

-## The datanode can export its metrics and send to Prometheus compatible service (e.g. send to `greptimedb` itself) from remote-write API.
+## The frontend can export its metrics and send to Prometheus compatible service (e.g. `greptimedb` itself) from remote-write API.
 ## This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape.
 [export_metrics]
-
 ## whether enable export metrics.
 enable = false
-
 ## The interval of export metrics.
 write_interval = "30s"

-## For `standalone` mode, `self_import` is recommend to collect metrics generated by itself
-## You must create the database before enabling it.
-[export_metrics.self_import]
-## @toml2docs:none-default
-db = "greptime_metrics"
-
 [export_metrics.remote_write]
-## The url the metrics send to. The url example can be: `http://127.0.0.1:4000/v1/prometheus/write?db=greptime_metrics`.
+## The prometheus remote write endpoint that the metrics send to. The url example can be: `http://127.0.0.1:4000/v1/prometheus/write?db=greptime_metrics`.
 url = ""

 ## HTTP headers of Prometheus remote-write carry.
--- a/config/metasrv.example.toml
+++ b/config/metasrv.example.toml
@@ -1,14 +1,6 @@
 ## The working home directory.
 data_home = "./greptimedb_data"

-## The bind address of metasrv.
-bind_addr = "127.0.0.1:3002"
-
-## The communication server address for the frontend and datanode to connect to metasrv.
-## If left empty or unset, the server will automatically use the IP address of the first network interface
-## on the host, with the same port number as the one specified in `bind_addr`.
-server_addr = "127.0.0.1:3002"
-
 ## Store server address default to etcd store.
 ## For postgres store, the format is:
 ## "password=password dbname=postgres user=postgres host=localhost port=5432"
@@ -24,6 +16,7 @@ store_key_prefix = ""
 ## - `etcd_store` (default value)
 ## - `memory_store`
 ## - `postgres_store`
+## - `mysql_store`
 backend = "etcd_store"

 ## Table name in RDS to store metadata. Effect when using a RDS kvbackend.
@@ -67,6 +60,21 @@ node_max_idle_time = "24hours"
 ## The number of threads to execute the runtime for global write operations.
 #+ compact_rt_size = 4

+## The gRPC server options.
+[grpc]
+## The address to bind the gRPC server.
+bind_addr = "127.0.0.1:3002"
+## The communication server address for the frontend and datanode to connect to metasrv.
+## If left empty or unset, the server will automatically use the IP address of the first network interface
+## on the host, with the same port number as the one specified in `bind_addr`.
+server_addr = "127.0.0.1:3002"
+## The number of server worker threads.
+runtime_size = 8
+## The maximum receive message size for gRPC server.
+max_recv_message_size = "512MB"
+## The maximum send message size for gRPC server.
+max_send_message_size = "512MB"
+
 ## The HTTP server options.
 [http]
 ## The address to bind the HTTP server.
@@ -229,24 +237,16 @@ max_log_files = 720
 [logging.tracing_sample_ratio]
 default_ratio = 1.0

-## The datanode can export its metrics and send to Prometheus compatible service (e.g. send to `greptimedb` itself) from remote-write API.
+## The metasrv can export its metrics and send to Prometheus compatible service (e.g. `greptimedb` itself) from remote-write API.
 ## This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape.
 [export_metrics]
-
 ## whether enable export metrics.
 enable = false
-
 ## The interval of export metrics.
 write_interval = "30s"

-## For `standalone` mode, `self_import` is recommend to collect metrics generated by itself
-## You must create the database before enabling it.
-[export_metrics.self_import]
-## @toml2docs:none-default
-db = "greptime_metrics"
-
 [export_metrics.remote_write]
-## The url the metrics send to. The url example can be: `http://127.0.0.1:4000/v1/prometheus/write?db=greptime_metrics`.
+## The prometheus remote write endpoint that the metrics send to. The url example can be: `http://127.0.0.1:4000/v1/prometheus/write?db=greptime_metrics`.
 url = ""

 ## HTTP headers of Prometheus remote-write carry.
--- a/config/standalone.example.toml
+++ b/config/standalone.example.toml
@@ -750,13 +750,11 @@ default_ratio = 1.0
 ## @toml2docs:none-default
 #+ sample_ratio = 1.0

-## The datanode can export its metrics and send to Prometheus compatible service (e.g. send to `greptimedb` itself) from remote-write API.
+## The standalone can export its metrics and send to Prometheus compatible service (e.g. `greptimedb`) from remote-write API.
 ## This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape.
 [export_metrics]
-
 ## whether enable export metrics.
 enable = false
-
 ## The interval of export metrics.
 write_interval = "30s"

@@ -767,7 +765,7 @@ write_interval = "30s"
 db = "greptime_metrics"

 [export_metrics.remote_write]
-## The url the metrics send to. The url example can be: `http://127.0.0.1:4000/v1/prometheus/write?db=greptime_metrics`.
+## The prometheus remote write endpoint that the metrics send to. The url example can be: `http://127.0.0.1:4000/v1/prometheus/write?db=greptime_metrics`.
 url = ""

 ## HTTP headers of Prometheus remote-write carry.
--- a/grafana/README.md
+++ b/grafana/README.md
@@ -9,7 +9,7 @@ We highly recommend using the self-monitoring feature provided by [GreptimeDB Op
 - **Metrics Dashboards**

  - `dashboards/metrics/cluster/dashboard.json`: The Grafana dashboard for the GreptimeDB cluster. Read the [dashboard.md](./dashboards/metrics/cluster/dashboard.md) for more details.
-  
+
  - `dashboards/metrics/standalone/dashboard.json`: The Grafana dashboard for the standalone GreptimeDB instance. **It's generated from the `cluster/dashboard.json` by removing the instance filter through the `make dashboards` command**. Read the [dashboard.md](./dashboards/metrics/standalone/dashboard.md) for more details.

 - **Logs Dashboard**
@@ -83,7 +83,7 @@ If you use the [Helm Chart](https://github.com/GreptimeTeam/helm-charts) to depl
 - `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/user-guide/deployments-administration/deploy-on-kubernetes/getting-started).

 ### Self-host Prometheus and import dashboards manually

--- a/src/catalog/Cargo.toml
+++ b/src/catalog/Cargo.toml
@@ -19,6 +19,7 @@ async-trait.workspace = true
 bytes.workspace = true
 common-catalog.workspace = true
 common-error.workspace = true
+common-frontend.workspace = true
 common-macro.workspace = true
 common-meta.workspace = true
 common-procedure.workspace = true
--- a/src/catalog/src/error.rs
+++ b/src/catalog/src/error.rs
@@ -277,6 +277,13 @@ pub enum Error {
        #[snafu(implicit)]
        location: Location,
    },
+
+    #[snafu(display("Failed to list frontend nodes"))]
+    ListProcess {
+        source: common_frontend::error::Error,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }

 impl Error {
@@ -345,6 +352,7 @@ impl ErrorExt for Error {
            Error::GetViewCache { source, .. } | Error::GetTableCache { source, .. } => {
                source.status_code()
            }
+            Error::ListProcess { source, .. } => source.status_code(),
        }
    }

--- a/src/catalog/src/kvbackend/manager.rs
+++ b/src/catalog/src/kvbackend/manager.rs
@@ -51,6 +51,7 @@ use crate::error::{
 };
 use crate::information_schema::{InformationExtensionRef, InformationSchemaProvider};
 use crate::kvbackend::TableCacheRef;
+use crate::process_manager::ProcessManagerRef;
 use crate::system_schema::pg_catalog::PGCatalogProvider;
 use crate::system_schema::SystemSchemaProvider;
 use crate::CatalogManager;
@@ -84,6 +85,7 @@ impl KvBackendCatalogManager {
        backend: KvBackendRef,
        cache_registry: LayeredCacheRegistryRef,
        procedure_manager: Option<ProcedureManagerRef>,
+        process_manager: Option<ProcessManagerRef>,
    ) -> Arc<Self> {
        Arc::new_cyclic(|me| Self {
            information_extension,
@@ -102,12 +104,14 @@ impl KvBackendCatalogManager {
                    DEFAULT_CATALOG_NAME.to_string(),
                    me.clone(),
                    Arc::new(FlowMetadataManager::new(backend.clone())),
+                    process_manager.clone(),
                )),
                pg_catalog_provider: Arc::new(PGCatalogProvider::new(
                    DEFAULT_CATALOG_NAME.to_string(),
                    me.clone(),
                )),
                backend,
+                process_manager,
            },
            cache_registry,
            procedure_manager,
@@ -419,6 +423,7 @@ struct SystemCatalog {
    information_schema_provider: Arc<InformationSchemaProvider>,
    pg_catalog_provider: Arc<PGCatalogProvider>,
    backend: KvBackendRef,
+    process_manager: Option<ProcessManagerRef>,
 }

 impl SystemCatalog {
@@ -486,6 +491,7 @@ impl SystemCatalog {
                        catalog.to_string(),
                        self.catalog_manager.clone(),
                        Arc::new(FlowMetadataManager::new(self.backend.clone())),
+                        self.process_manager.clone(),
                    ))
                });
            information_schema_provider.table(table_name)
--- a/src/catalog/src/lib.rs
+++ b/src/catalog/src/lib.rs
@@ -40,6 +40,7 @@ pub mod information_schema {
    pub use crate::system_schema::information_schema::*;
 }

+pub mod process_manager;
 pub mod table_source;

 #[async_trait::async_trait]
--- a/src/catalog/src/memory/manager.rs
+++ b/src/catalog/src/memory/manager.rs
@@ -356,6 +356,7 @@ impl MemoryCatalogManager {
            catalog,
            Arc::downgrade(self) as Weak<dyn CatalogManager>,
            Arc::new(FlowMetadataManager::new(Arc::new(MemoryKvBackend::new()))),
+            None, // we don't need ProcessManager on regions server.
        );
        let information_schema = information_schema_provider.tables().clone();

--- a/src/catalog/src/process_manager.rs
+++ b/src/catalog/src/process_manager.rs
@@ -0,0 +1,188 @@
+// 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::collections::hash_map::Entry;
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, RwLock};
+
+use api::v1::frontend::{ListProcessRequest, ProcessInfo};
+use common_frontend::selector::{FrontendSelector, MetaClientSelector};
+use common_telemetry::{debug, info};
+use common_time::util::current_time_millis;
+use meta_client::MetaClientRef;
+use snafu::ResultExt;
+
+use crate::error;
+
+pub type ProcessManagerRef = Arc<ProcessManager>;
+
+pub struct ProcessManager {
+    server_addr: String,
+    next_id: AtomicU64,
+    catalogs: RwLock<HashMap<String, HashMap<u64, ProcessInfo>>>,
+    frontend_selector: Option<MetaClientSelector>,
+}
+
+impl ProcessManager {
+    /// Create a [ProcessManager] instance with server address and kv client.
+    pub fn new(server_addr: String, meta_client: Option<MetaClientRef>) -> Self {
+        let frontend_selector = meta_client.map(MetaClientSelector::new);
+        Self {
+            server_addr,
+            next_id: Default::default(),
+            catalogs: Default::default(),
+            frontend_selector,
+        }
+    }
+}
+
+impl ProcessManager {
+    /// Registers a submitted query. Use the provided id if present.
+    pub fn register_query(
+        self: &Arc<Self>,
+        catalog: String,
+        schemas: Vec<String>,
+        query: String,
+        client: String,
+        id: Option<u64>,
+    ) -> Ticket {
+        let id = id.unwrap_or_else(|| self.next_id.fetch_add(1, Ordering::Relaxed));
+        let process = ProcessInfo {
+            id,
+            catalog: catalog.clone(),
+            schemas,
+            query,
+            start_timestamp: current_time_millis(),
+            client,
+            frontend: self.server_addr.clone(),
+        };
+        self.catalogs
+            .write()
+            .unwrap()
+            .entry(catalog.clone())
+            .or_default()
+            .insert(id, process);
+        Ticket {
+            catalog,
+            manager: self.clone(),
+            id,
+        }
+    }
+
+    /// Generates the next process id.
+    pub fn next_id(&self) -> u64 {
+        self.next_id.fetch_add(1, Ordering::Relaxed)
+    }
+
+    /// De-register a query from process list.
+    pub fn deregister_query(&self, catalog: String, id: u64) {
+        if let Entry::Occupied(mut o) = self.catalogs.write().unwrap().entry(catalog) {
+            let process = o.get_mut().remove(&id);
+            debug!("Deregister process: {:?}", process);
+            if o.get_mut().is_empty() {
+                o.remove();
+            }
+        }
+    }
+
+    pub fn deregister_all_queries(&self) {
+        self.catalogs.write().unwrap().clear();
+        info!("All queries on {} has been deregistered", self.server_addr);
+    }
+
+    /// List local running processes in given catalog.
+    pub fn local_processes(&self, catalog: Option<&str>) -> error::Result<Vec<ProcessInfo>> {
+        let catalogs = self.catalogs.read().unwrap();
+        let result = if let Some(catalog) = catalog {
+            if let Some(catalogs) = catalogs.get(catalog) {
+                catalogs.values().cloned().collect()
+            } else {
+                vec![]
+            }
+        } else {
+            catalogs
+                .values()
+                .flat_map(|v| v.values().cloned())
+                .collect()
+        };
+        Ok(result)
+    }
+
+    pub async fn list_all_processes(
+        &self,
+        catalog: Option<&str>,
+    ) -> error::Result<Vec<ProcessInfo>> {
+        let mut processes = vec![];
+        if let Some(remote_frontend_selector) = self.frontend_selector.as_ref() {
+            let frontends = remote_frontend_selector
+                .select(|node| node.peer.addr != self.server_addr)
+                .await
+                .context(error::ListProcessSnafu)?;
+            for mut f in frontends {
+                processes.extend(
+                    f.list_process(ListProcessRequest {
+                        catalog: catalog.unwrap_or_default().to_string(),
+                    })
+                    .await
+                    .context(error::ListProcessSnafu)?
+                    .processes,
+                );
+            }
+        }
+        processes.extend(self.local_processes(catalog)?);
+        Ok(processes)
+    }
+}
+
+pub struct Ticket {
+    pub(crate) catalog: String,
+    pub(crate) manager: ProcessManagerRef,
+    pub(crate) id: u64,
+}
+
+impl Drop for Ticket {
+    fn drop(&mut self) {
+        self.manager
+            .deregister_query(std::mem::take(&mut self.catalog), self.id);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use crate::process_manager::ProcessManager;
+
+    #[tokio::test]
+    async fn test_register_query() {
+        let process_manager = Arc::new(ProcessManager::new("127.0.0.1:8000".to_string(), None));
+        let ticket = process_manager.clone().register_query(
+            "public".to_string(),
+            vec!["test".to_string()],
+            "SELECT * FROM table".to_string(),
+            "".to_string(),
+            None,
+        );
+
+        let running_processes = process_manager.local_processes(None).unwrap();
+        assert_eq!(running_processes.len(), 1);
+        assert_eq!(&running_processes[0].frontend, "127.0.0.1:8000");
+        assert_eq!(running_processes[0].id, ticket.id);
+        assert_eq!(&running_processes[0].query, "SELECT * FROM table");
+
+        drop(ticket);
+        assert_eq!(process_manager.local_processes(None).unwrap().len(), 0);
+    }
+}
--- a/src/catalog/src/system_schema/information_schema.rs
+++ b/src/catalog/src/system_schema/information_schema.rs
@@ -19,6 +19,7 @@ mod information_memory_table;
 pub mod key_column_usage;
 mod partitions;
 mod procedure_info;
+mod process_list;
 pub mod region_peers;
 mod region_statistics;
 mod runtime_metrics;
@@ -42,6 +43,7 @@ use common_recordbatch::SendableRecordBatchStream;
 use datatypes::schema::SchemaRef;
 use lazy_static::lazy_static;
 use paste::paste;
+use process_list::InformationSchemaProcessList;
 use store_api::storage::{ScanRequest, TableId};
 use table::metadata::TableType;
 use table::TableRef;
@@ -50,6 +52,7 @@ use views::InformationSchemaViews;

 use self::columns::InformationSchemaColumns;
 use crate::error::{Error, Result};
+use crate::process_manager::ProcessManagerRef;
 use crate::system_schema::information_schema::cluster_info::InformationSchemaClusterInfo;
 use crate::system_schema::information_schema::flows::InformationSchemaFlows;
 use crate::system_schema::information_schema::information_memory_table::get_schema_columns;
@@ -113,6 +116,7 @@ macro_rules! setup_memory_table {
 pub struct InformationSchemaProvider {
    catalog_name: String,
    catalog_manager: Weak<dyn CatalogManager>,
+    process_manager: Option<ProcessManagerRef>,
    flow_metadata_manager: Arc<FlowMetadataManager>,
    tables: HashMap<String, TableRef>,
 }
@@ -207,6 +211,10 @@ impl SystemSchemaProviderInner for InformationSchemaProvider {
                    self.catalog_manager.clone(),
                ),
            ) as _),
+            PROCESS_LIST => self
+                .process_manager
+                .as_ref()
+                .map(|p| Arc::new(InformationSchemaProcessList::new(p.clone())) as _),
            _ => None,
        }
    }
@@ -217,11 +225,13 @@ impl InformationSchemaProvider {
        catalog_name: String,
        catalog_manager: Weak<dyn CatalogManager>,
        flow_metadata_manager: Arc<FlowMetadataManager>,
+        process_manager: Option<ProcessManagerRef>,
    ) -> Self {
        let mut provider = Self {
            catalog_name,
            catalog_manager,
            flow_metadata_manager,
+            process_manager,
            tables: HashMap::new(),
        };

@@ -277,6 +287,9 @@ impl InformationSchemaProvider {
            self.build_table(TABLE_CONSTRAINTS).unwrap(),
        );
        tables.insert(FLOWS.to_string(), self.build_table(FLOWS).unwrap());
+        if let Some(process_list) = self.build_table(PROCESS_LIST) {
+            tables.insert(PROCESS_LIST.to_string(), process_list);
+        }
        // Add memory tables
        for name in MEMORY_TABLES.iter() {
            tables.insert((*name).to_string(), self.build_table(name).expect(name));
--- a/src/catalog/src/system_schema/information_schema/process_list.rs
+++ b/src/catalog/src/system_schema/information_schema/process_list.rs
@@ -0,0 +1,189 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use common_catalog::consts::INFORMATION_SCHEMA_PROCESS_LIST_TABLE_ID;
+use common_error::ext::BoxedError;
+use common_frontend::DisplayProcessId;
+use common_recordbatch::adapter::RecordBatchStreamAdapter;
+use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
+use common_time::util::current_time_millis;
+use common_time::{Duration, Timestamp};
+use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
+use datatypes::prelude::ConcreteDataType as CDT;
+use datatypes::scalars::ScalarVectorBuilder;
+use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
+use datatypes::value::Value;
+use datatypes::vectors::{
+    DurationMillisecondVectorBuilder, StringVectorBuilder, TimestampMillisecondVectorBuilder,
+    VectorRef,
+};
+use snafu::ResultExt;
+use store_api::storage::{ScanRequest, TableId};
+
+use crate::error::{self, InternalSnafu};
+use crate::information_schema::Predicates;
+use crate::process_manager::ProcessManagerRef;
+use crate::system_schema::information_schema::InformationTable;
+
+/// Column names of `information_schema.process_list`
+const ID: &str = "id";
+const CATALOG: &str = "catalog";
+const SCHEMAS: &str = "schemas";
+const QUERY: &str = "query";
+const CLIENT: &str = "client";
+const FRONTEND: &str = "frontend";
+const START_TIMESTAMP: &str = "start_timestamp";
+const ELAPSED_TIME: &str = "elapsed_time";
+
+/// `information_schema.process_list` table implementation that tracks running
+/// queries in current cluster.
+pub struct InformationSchemaProcessList {
+    schema: SchemaRef,
+    process_manager: ProcessManagerRef,
+}
+
+impl InformationSchemaProcessList {
+    pub fn new(process_manager: ProcessManagerRef) -> Self {
+        Self {
+            schema: Self::schema(),
+            process_manager,
+        }
+    }
+
+    fn schema() -> SchemaRef {
+        Arc::new(Schema::new(vec![
+            ColumnSchema::new(ID, CDT::string_datatype(), false),
+            ColumnSchema::new(CATALOG, CDT::string_datatype(), false),
+            ColumnSchema::new(SCHEMAS, CDT::string_datatype(), false),
+            ColumnSchema::new(QUERY, CDT::string_datatype(), false),
+            ColumnSchema::new(CLIENT, CDT::string_datatype(), false),
+            ColumnSchema::new(FRONTEND, CDT::string_datatype(), false),
+            ColumnSchema::new(
+                START_TIMESTAMP,
+                CDT::timestamp_millisecond_datatype(),
+                false,
+            ),
+            ColumnSchema::new(ELAPSED_TIME, CDT::duration_millisecond_datatype(), false),
+        ]))
+    }
+}
+
+impl InformationTable for InformationSchemaProcessList {
+    fn table_id(&self) -> TableId {
+        INFORMATION_SCHEMA_PROCESS_LIST_TABLE_ID
+    }
+
+    fn table_name(&self) -> &'static str {
+        "process_list"
+    }
+
+    fn schema(&self) -> SchemaRef {
+        self.schema.clone()
+    }
+
+    fn to_stream(&self, request: ScanRequest) -> error::Result<SendableRecordBatchStream> {
+        let process_manager = self.process_manager.clone();
+        let stream = Box::pin(DfRecordBatchStreamAdapter::new(
+            self.schema.arrow_schema().clone(),
+            futures::stream::once(async move {
+                make_process_list(process_manager, request)
+                    .await
+                    .map(RecordBatch::into_df_record_batch)
+                    .map_err(|e| datafusion::error::DataFusionError::External(Box::new(e)))
+            }),
+        ));
+
+        Ok(Box::pin(
+            RecordBatchStreamAdapter::try_new(stream)
+                .map_err(BoxedError::new)
+                .context(InternalSnafu)?,
+        ))
+    }
+}
+
+/// Build running process list.
+async fn make_process_list(
+    process_manager: ProcessManagerRef,
+    request: ScanRequest,
+) -> error::Result<RecordBatch> {
+    let predicates = Predicates::from_scan_request(&Some(request));
+    let current_time = current_time_millis();
+    // todo(hl): find a way to extract user catalog to filter queries from other users.
+    let queries = process_manager.list_all_processes(None).await?;
+
+    let mut id_builder = StringVectorBuilder::with_capacity(queries.len());
+    let mut catalog_builder = StringVectorBuilder::with_capacity(queries.len());
+    let mut schemas_builder = StringVectorBuilder::with_capacity(queries.len());
+    let mut query_builder = StringVectorBuilder::with_capacity(queries.len());
+    let mut client_builder = StringVectorBuilder::with_capacity(queries.len());
+    let mut frontend_builder = StringVectorBuilder::with_capacity(queries.len());
+    let mut start_time_builder = TimestampMillisecondVectorBuilder::with_capacity(queries.len());
+    let mut elapsed_time_builder = DurationMillisecondVectorBuilder::with_capacity(queries.len());
+
+    for process in queries {
+        let display_id = DisplayProcessId {
+            server_addr: process.frontend.to_string(),
+            id: process.id,
+        }
+        .to_string();
+        let schemas = process.schemas.join(",");
+        let id = Value::from(display_id);
+        let catalog = Value::from(process.catalog);
+        let schemas = Value::from(schemas);
+        let query = Value::from(process.query);
+        let client = Value::from(process.client);
+        let frontend = Value::from(process.frontend);
+        let start_timestamp = Value::from(Timestamp::new_millisecond(process.start_timestamp));
+        let elapsed_time = Value::from(Duration::new_millisecond(
+            current_time - process.start_timestamp,
+        ));
+        let row = [
+            (ID, &id),
+            (CATALOG, &catalog),
+            (SCHEMAS, &schemas),
+            (QUERY, &query),
+            (CLIENT, &client),
+            (FRONTEND, &frontend),
+            (START_TIMESTAMP, &start_timestamp),
+            (ELAPSED_TIME, &elapsed_time),
+        ];
+        if predicates.eval(&row) {
+            id_builder.push(id.as_string().as_deref());
+            catalog_builder.push(catalog.as_string().as_deref());
+            schemas_builder.push(schemas.as_string().as_deref());
+            query_builder.push(query.as_string().as_deref());
+            client_builder.push(client.as_string().as_deref());
+            frontend_builder.push(frontend.as_string().as_deref());
+            start_time_builder.push(start_timestamp.as_timestamp().map(|t| t.value().into()));
+            elapsed_time_builder.push(elapsed_time.as_duration().map(|d| d.value().into()));
+        }
+    }
+
+    RecordBatch::new(
+        InformationSchemaProcessList::schema(),
+        vec![
+            Arc::new(id_builder.finish()) as VectorRef,
+            Arc::new(catalog_builder.finish()) as VectorRef,
+            Arc::new(schemas_builder.finish()) as VectorRef,
+            Arc::new(query_builder.finish()) as VectorRef,
+            Arc::new(client_builder.finish()) as VectorRef,
+            Arc::new(frontend_builder.finish()) as VectorRef,
+            Arc::new(start_time_builder.finish()) as VectorRef,
+            Arc::new(elapsed_time_builder.finish()) as VectorRef,
+        ],
+    )
+    .context(error::CreateRecordBatchSnafu)
+}
--- a/src/catalog/src/system_schema/information_schema/table_names.rs
+++ b/src/catalog/src/system_schema/information_schema/table_names.rs
@@ -47,3 +47,4 @@ pub const VIEWS: &str = "views";
 pub const FLOWS: &str = "flows";
 pub const PROCEDURE_INFO: &str = "procedure_info";
 pub const REGION_STATISTICS: &str = "region_statistics";
+pub const PROCESS_LIST: &str = "process_list";
--- a/src/catalog/src/table_source.rs
+++ b/src/catalog/src/table_source.rs
@@ -328,6 +328,7 @@ mod tests {
            backend.clone(),
            layered_cache_registry,
            None,
+            None,
        );
        let table_metadata_manager = TableMetadataManager::new(backend);
        let mut view_info = common_meta::key::test_utils::new_test_table_info(1024, vec![]);
--- a/src/cli/src/bench.rs
+++ b/src/cli/src/bench.rs
@@ -58,6 +58,7 @@ where
    info!("{desc}, average operation cost: {cost:.2} ms");
 }

+/// Command to benchmark table metadata operations.
 #[derive(Debug, Default, Parser)]
 pub struct BenchTableMetadataCommand {
    #[clap(long)]
--- a/src/cli/src/data.rs
+++ b/src/cli/src/data.rs
@@ -0,0 +1,39 @@
+// 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.
+
+mod export;
+mod import;
+
+use clap::Subcommand;
+use common_error::ext::BoxedError;
+
+use crate::data::export::ExportCommand;
+use crate::data::import::ImportCommand;
+use crate::Tool;
+
+/// Command for data operations including exporting data from and importing data into GreptimeDB.
+#[derive(Subcommand)]
+pub enum DataCommand {
+    Export(ExportCommand),
+    Import(ImportCommand),
+}
+
+impl DataCommand {
+    pub async fn build(&self) -> std::result::Result<Box<dyn Tool>, BoxedError> {
+        match self {
+            DataCommand::Export(cmd) => cmd.build().await,
+            DataCommand::Import(cmd) => cmd.build().await,
+        }
+    }
+}
--- a/src/cli/src/data/export.rs
+++ b/src/cli/src/data/export.rs
@@ -50,6 +50,7 @@ enum ExportTarget {
    All,
 }

+/// Command for exporting data from the GreptimeDB.
 #[derive(Debug, Default, Parser)]
 pub struct ExportCommand {
    /// Server address to connect
--- a/src/cli/src/data/import.rs
+++ b/src/cli/src/data/import.rs
@@ -40,6 +40,7 @@ enum ImportTarget {
    All,
 }

+/// Command to import data from a directory into a GreptimeDB instance.
 #[derive(Debug, Default, Parser)]
 pub struct ImportCommand {
    /// Server address to connect
--- a/src/cli/src/error.rs
+++ b/src/cli/src/error.rs
@@ -30,6 +30,7 @@ pub enum Error {
        location: Location,
        msg: String,
    },
+
    #[snafu(display("Failed to create default catalog and schema"))]
    InitMetadata {
        #[snafu(implicit)]
@@ -228,19 +229,41 @@ pub enum Error {
        #[snafu(source)]
        error: ObjectStoreError,
    },
+
    #[snafu(display("S3 config need be set"))]
    S3ConfigNotSet {
        #[snafu(implicit)]
        location: Location,
    },
+
    #[snafu(display("Output directory not set"))]
    OutputDirNotSet {
        #[snafu(implicit)]
        location: Location,
    },
-    #[snafu(display("KV backend not set: {}", backend))]
-    KvBackendNotSet {
-        backend: String,
+
+    #[snafu(display("Empty store addresses"))]
+    EmptyStoreAddrs {
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Unsupported memory backend"))]
+    UnsupportedMemoryBackend {
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("File path invalid: {}", msg))]
+    InvalidFilePath {
+        msg: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Invalid arguments: {}", msg))]
+    InvalidArguments {
+        msg: String,
        #[snafu(implicit)]
        location: Location,
    },
@@ -262,6 +285,9 @@ impl ErrorExt for Error {
            | Error::ConnectEtcd { .. }
            | Error::CreateDir { .. }
            | Error::EmptyResult { .. }
+            | Error::InvalidFilePath { .. }
+            | Error::UnsupportedMemoryBackend { .. }
+            | Error::InvalidArguments { .. }
            | Error::ParseProxyOpts { .. } => StatusCode::InvalidArguments,

            Error::StartProcedureManager { source, .. }
@@ -282,7 +308,7 @@ impl ErrorExt for Error {
            Error::OpenDal { .. } => StatusCode::Internal,
            Error::S3ConfigNotSet { .. }
            | Error::OutputDirNotSet { .. }
-            | Error::KvBackendNotSet { .. } => StatusCode::InvalidArguments,
+            | Error::EmptyStoreAddrs { .. } => StatusCode::InvalidArguments,

            Error::BuildRuntime { source, .. } => source.status_code(),

--- a/src/cli/src/lib.rs
+++ b/src/cli/src/lib.rs
@@ -13,11 +13,10 @@
 // limitations under the License.

 mod bench;
+mod data;
 mod database;
 pub mod error;
-mod export;
-mod import;
-mod meta_snapshot;
+mod metadata;

 use async_trait::async_trait;
 use clap::Parser;
@@ -26,9 +25,8 @@ pub use database::DatabaseClient;
 use error::Result;

 pub use crate::bench::BenchTableMetadataCommand;
-pub use crate::export::ExportCommand;
-pub use crate::import::ImportCommand;
-pub use crate::meta_snapshot::{MetaRestoreCommand, MetaSnapshotCommand};
+pub use crate::data::DataCommand;
+pub use crate::metadata::MetadataCommand;

 #[async_trait]
 pub trait Tool: Send + Sync {
--- a/src/cli/src/metadata.rs
+++ b/src/cli/src/metadata.rs
@@ -0,0 +1,42 @@
+// 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.
+
+mod common;
+mod control;
+mod snapshot;
+
+use clap::Subcommand;
+use common_error::ext::BoxedError;
+
+use crate::metadata::control::ControlCommand;
+use crate::metadata::snapshot::SnapshotCommand;
+use crate::Tool;
+
+/// Command for managing metadata operations, including saving metadata snapshots and restoring metadata from snapshots.
+#[derive(Subcommand)]
+pub enum MetadataCommand {
+    #[clap(subcommand)]
+    Snapshot(SnapshotCommand),
+    #[clap(subcommand)]
+    Control(ControlCommand),
+}
+
+impl MetadataCommand {
+    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
+        match self {
+            MetadataCommand::Snapshot(cmd) => cmd.build().await,
+            MetadataCommand::Control(cmd) => cmd.build().await,
+        }
+    }
+}
--- a/src/cli/src/metadata/common.rs
+++ b/src/cli/src/metadata/common.rs
@@ -0,0 +1,116 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use clap::Parser;
+use common_error::ext::BoxedError;
+use common_meta::kv_backend::chroot::ChrootKvBackend;
+use common_meta::kv_backend::etcd::EtcdStore;
+use common_meta::kv_backend::KvBackendRef;
+use meta_srv::bootstrap::create_etcd_client;
+use meta_srv::metasrv::BackendImpl;
+
+use crate::error::{EmptyStoreAddrsSnafu, UnsupportedMemoryBackendSnafu};
+
+#[derive(Debug, Default, Parser)]
+pub(crate) struct StoreConfig {
+    /// The endpoint of store. one of etcd, postgres or mysql.
+    ///
+    /// For postgres store, the format is:
+    /// "password=password dbname=postgres user=postgres host=localhost port=5432"
+    ///
+    /// For etcd store, the format is:
+    /// "127.0.0.1:2379"
+    ///
+    /// For mysql store, the format is:
+    /// "mysql://user:password@ip:port/dbname"
+    #[clap(long, alias = "store-addr", value_delimiter = ',', num_args = 1..)]
+    store_addrs: Vec<String>,
+
+    /// The maximum number of operations in a transaction. Only used when using [etcd-store].
+    #[clap(long, default_value = "128")]
+    max_txn_ops: usize,
+
+    /// The metadata store backend.
+    #[clap(long, value_enum, default_value = "etcd-store")]
+    backend: BackendImpl,
+
+    /// The key prefix of the metadata store.
+    #[clap(long, default_value = "")]
+    store_key_prefix: String,
+
+    /// The table name in RDS to store metadata. Only used when using [postgres-store] or [mysql-store].
+    #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
+    #[clap(long, default_value = common_meta::kv_backend::DEFAULT_META_TABLE_NAME)]
+    meta_table_name: String,
+}
+
+impl StoreConfig {
+    /// Builds a [`KvBackendRef`] from the store configuration.
+    pub async fn build(&self) -> Result<KvBackendRef, BoxedError> {
+        let max_txn_ops = self.max_txn_ops;
+        let store_addrs = &self.store_addrs;
+        if store_addrs.is_empty() {
+            EmptyStoreAddrsSnafu.fail().map_err(BoxedError::new)
+        } else {
+            let kvbackend = match self.backend {
+                BackendImpl::EtcdStore => {
+                    let etcd_client = create_etcd_client(store_addrs)
+                        .await
+                        .map_err(BoxedError::new)?;
+                    Ok(EtcdStore::with_etcd_client(etcd_client, max_txn_ops))
+                }
+                #[cfg(feature = "pg_kvbackend")]
+                BackendImpl::PostgresStore => {
+                    let table_name = &self.meta_table_name;
+                    let pool = meta_srv::bootstrap::create_postgres_pool(store_addrs)
+                        .await
+                        .map_err(BoxedError::new)?;
+                    Ok(common_meta::kv_backend::rds::PgStore::with_pg_pool(
+                        pool,
+                        table_name,
+                        max_txn_ops,
+                    )
+                    .await
+                    .map_err(BoxedError::new)?)
+                }
+                #[cfg(feature = "mysql_kvbackend")]
+                BackendImpl::MysqlStore => {
+                    let table_name = &self.meta_table_name;
+                    let pool = meta_srv::bootstrap::create_mysql_pool(store_addrs)
+                        .await
+                        .map_err(BoxedError::new)?;
+                    Ok(common_meta::kv_backend::rds::MySqlStore::with_mysql_pool(
+                        pool,
+                        table_name,
+                        max_txn_ops,
+                    )
+                    .await
+                    .map_err(BoxedError::new)?)
+                }
+                BackendImpl::MemoryStore => UnsupportedMemoryBackendSnafu
+                    .fail()
+                    .map_err(BoxedError::new),
+            };
+            if self.store_key_prefix.is_empty() {
+                kvbackend
+            } else {
+                let chroot_kvbackend =
+                    ChrootKvBackend::new(self.store_key_prefix.as_bytes().to_vec(), kvbackend?);
+                Ok(Arc::new(chroot_kvbackend))
+            }
+        }
+    }
+}
--- a/src/cli/src/metadata/control.rs
+++ b/src/cli/src/metadata/control.rs
@@ -0,0 +1,38 @@
+// 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.
+
+mod get;
+mod utils;
+
+use clap::Subcommand;
+use common_error::ext::BoxedError;
+use get::GetCommand;
+
+use crate::Tool;
+
+/// Subcommand for metadata control.
+#[derive(Subcommand)]
+pub enum ControlCommand {
+    /// Get the metadata from the metasrv.
+    #[clap(subcommand)]
+    Get(GetCommand),
+}
+
+impl ControlCommand {
+    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
+        match self {
+            ControlCommand::Get(cmd) => cmd.build().await,
+        }
+    }
+}
--- a/src/cli/src/metadata/control/get.rs
+++ b/src/cli/src/metadata/control/get.rs
@@ -0,0 +1,242 @@
+// 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::cmp::min;
+
+use async_trait::async_trait;
+use clap::{Parser, Subcommand};
+use client::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
+use common_catalog::format_full_table_name;
+use common_error::ext::BoxedError;
+use common_meta::key::table_info::TableInfoKey;
+use common_meta::key::table_name::TableNameKey;
+use common_meta::key::table_route::TableRouteKey;
+use common_meta::key::TableMetadataManager;
+use common_meta::kv_backend::KvBackendRef;
+use common_meta::range_stream::{PaginationStream, DEFAULT_PAGE_SIZE};
+use common_meta::rpc::store::RangeRequest;
+use futures::TryStreamExt;
+
+use crate::error::InvalidArgumentsSnafu;
+use crate::metadata::common::StoreConfig;
+use crate::metadata::control::utils::{decode_key_value, json_fromatter};
+use crate::Tool;
+
+/// Subcommand for get command.
+#[derive(Subcommand)]
+pub enum GetCommand {
+    Key(GetKeyCommand),
+    Table(GetTableCommand),
+}
+
+impl GetCommand {
+    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
+        match self {
+            GetCommand::Key(cmd) => cmd.build().await,
+            GetCommand::Table(cmd) => cmd.build().await,
+        }
+    }
+}
+
+/// Get key-value pairs from the metadata store.
+#[derive(Debug, Default, Parser)]
+pub struct GetKeyCommand {
+    /// The key to get from the metadata store. If empty, returns all key-value pairs.
+    #[clap(default_value = "")]
+    key: String,
+
+    /// Whether to perform a prefix query. If true, returns all key-value pairs where the key starts with the given prefix.
+    #[clap(long, default_value = "false")]
+    prefix: bool,
+
+    /// The maximum number of key-value pairs to return. If 0, returns all key-value pairs.
+    #[clap(long, default_value = "0")]
+    limit: u64,
+
+    #[clap(flatten)]
+    store: StoreConfig,
+}
+
+impl GetKeyCommand {
+    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
+        let kvbackend = self.store.build().await?;
+        Ok(Box::new(GetKeyTool {
+            kvbackend,
+            key: self.key.clone(),
+            prefix: self.prefix,
+            limit: self.limit,
+        }))
+    }
+}
+
+struct GetKeyTool {
+    kvbackend: KvBackendRef,
+    key: String,
+    prefix: bool,
+    limit: u64,
+}
+
+#[async_trait]
+impl Tool for GetKeyTool {
+    async fn do_work(&self) -> Result<(), BoxedError> {
+        let mut req = RangeRequest::default();
+        if self.prefix {
+            req = req.with_prefix(self.key.as_bytes());
+        } else {
+            req = req.with_key(self.key.as_bytes());
+        }
+        let page_size = if self.limit > 0 {
+            min(self.limit as usize, DEFAULT_PAGE_SIZE)
+        } else {
+            DEFAULT_PAGE_SIZE
+        };
+        let pagination_stream =
+            PaginationStream::new(self.kvbackend.clone(), req, page_size, decode_key_value);
+        let mut stream = Box::pin(pagination_stream.into_stream());
+        let mut counter = 0;
+
+        while let Some((key, value)) = stream.try_next().await.map_err(BoxedError::new)? {
+            print!("{}\n{}\n", key, value);
+            counter += 1;
+            if self.limit > 0 && counter >= self.limit {
+                break;
+            }
+        }
+
+        Ok(())
+    }
+}
+
+/// Get table metadata from the metadata store via table id.
+#[derive(Debug, Default, Parser)]
+pub struct GetTableCommand {
+    /// Get table metadata by table id.
+    #[clap(long)]
+    table_id: Option<u32>,
+
+    /// Get table metadata by table name.
+    #[clap(long)]
+    table_name: Option<String>,
+
+    /// The schema name of the table.
+    #[clap(long)]
+    schema_name: Option<String>,
+
+    /// Pretty print the output.
+    #[clap(long, default_value = "false")]
+    pretty: bool,
+
+    #[clap(flatten)]
+    store: StoreConfig,
+}
+
+impl GetTableCommand {
+    pub fn validate(&self) -> Result<(), BoxedError> {
+        if self.table_id.is_none() && self.table_name.is_none() {
+            return Err(BoxedError::new(
+                InvalidArgumentsSnafu {
+                    msg: "You must specify either --table-id or --table-name.",
+                }
+                .build(),
+            ));
+        }
+        Ok(())
+    }
+}
+
+struct GetTableTool {
+    kvbackend: KvBackendRef,
+    table_id: Option<u32>,
+    table_name: Option<String>,
+    schema_name: Option<String>,
+    pretty: bool,
+}
+
+#[async_trait]
+impl Tool for GetTableTool {
+    async fn do_work(&self) -> Result<(), BoxedError> {
+        let table_metadata_manager = TableMetadataManager::new(self.kvbackend.clone());
+        let table_name_manager = table_metadata_manager.table_name_manager();
+        let table_info_manager = table_metadata_manager.table_info_manager();
+        let table_route_manager = table_metadata_manager.table_route_manager();
+
+        let table_id = if let Some(table_name) = &self.table_name {
+            let catalog = DEFAULT_CATALOG_NAME.to_string();
+            let schema_name = self
+                .schema_name
+                .clone()
+                .unwrap_or_else(|| DEFAULT_SCHEMA_NAME.to_string());
+            let key = TableNameKey::new(&catalog, &schema_name, table_name);
+
+            let Some(table_name) = table_name_manager.get(key).await.map_err(BoxedError::new)?
+            else {
+                println!(
+                    "Table({}) not found",
+                    format_full_table_name(&catalog, &schema_name, table_name)
+                );
+                return Ok(());
+            };
+
+            table_name.table_id()
+        } else {
+            // Safety: we have validated that table_id or table_name is not None
+            self.table_id.unwrap()
+        };
+
+        let table_info = table_info_manager
+            .get(table_id)
+            .await
+            .map_err(BoxedError::new)?;
+        if let Some(table_info) = table_info {
+            println!(
+                "{}\n{}",
+                TableInfoKey::new(table_id),
+                json_fromatter(self.pretty, &*table_info)
+            );
+        } else {
+            println!("Table info not found");
+        }
+
+        let table_route = table_route_manager
+            .table_route_storage()
+            .get(table_id)
+            .await
+            .map_err(BoxedError::new)?;
+        if let Some(table_route) = table_route {
+            println!(
+                "{}\n{}",
+                TableRouteKey::new(table_id),
+                json_fromatter(self.pretty, &table_route)
+            );
+        } else {
+            println!("Table route not found");
+        }
+
+        Ok(())
+    }
+}
+
+impl GetTableCommand {
+    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
+        self.validate()?;
+        let kvbackend = self.store.build().await?;
+        Ok(Box::new(GetTableTool {
+            kvbackend,
+            table_id: self.table_id,
+            table_name: self.table_name.clone(),
+            schema_name: self.schema_name.clone(),
+            pretty: self.pretty,
+        }))
+    }
+}
--- a/src/cli/src/metadata/control/utils.rs
+++ b/src/cli/src/metadata/control/utils.rs
@@ -0,0 +1,36 @@
+// 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 common_meta::error::Result as CommonMetaResult;
+use common_meta::rpc::KeyValue;
+use serde::Serialize;
+
+/// Decodes a key-value pair into a string.
+pub fn decode_key_value(kv: KeyValue) -> CommonMetaResult<(String, String)> {
+    let key = String::from_utf8_lossy(&kv.key).to_string();
+    let value = String::from_utf8_lossy(&kv.value).to_string();
+    Ok((key, value))
+}
+
+/// Formats a value as a JSON string.
+pub fn json_fromatter<T>(pretty: bool, value: &T) -> String
+where
+    T: Serialize,
+{
+    if pretty {
+        serde_json::to_string_pretty(value).unwrap()
+    } else {
+        serde_json::to_string(value).unwrap()
+    }
+}
--- a/src/cli/src/metadata/snapshot.rs
+++ b/src/cli/src/metadata/snapshot.rs
@@ -12,96 +12,38 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use std::sync::Arc;
+use std::path::Path;

 use async_trait::async_trait;
-use clap::Parser;
+use clap::{Parser, Subcommand};
 use common_base::secrets::{ExposeSecret, SecretString};
 use common_error::ext::BoxedError;
-use common_meta::kv_backend::chroot::ChrootKvBackend;
-use common_meta::kv_backend::etcd::EtcdStore;
-use common_meta::kv_backend::KvBackendRef;
 use common_meta::snapshot::MetadataSnapshotManager;
-use meta_srv::bootstrap::create_etcd_client;
-use meta_srv::metasrv::BackendImpl;
 use object_store::services::{Fs, S3};
 use object_store::ObjectStore;
-use snafu::ResultExt;
+use snafu::{OptionExt, ResultExt};

-use crate::error::{KvBackendNotSetSnafu, OpenDalSnafu, S3ConfigNotSetSnafu};
+use crate::error::{InvalidFilePathSnafu, OpenDalSnafu, S3ConfigNotSetSnafu};
+use crate::metadata::common::StoreConfig;
 use crate::Tool;
-#[derive(Debug, Default, Parser)]
-struct MetaConnection {
-    /// The endpoint of store. one of etcd, pg or mysql.
-    #[clap(long, alias = "store-addr", value_delimiter = ',', num_args = 1..)]
-    store_addrs: Vec<String>,
-    /// The database backend.
-    #[clap(long, value_enum)]
-    backend: Option<BackendImpl>,
-    #[clap(long, default_value = "")]
-    store_key_prefix: String,
-    #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
-    #[clap(long,default_value = common_meta::kv_backend::DEFAULT_META_TABLE_NAME)]
-    meta_table_name: String,
-    #[clap(long, default_value = "128")]
-    max_txn_ops: usize,
+
+/// Subcommand for metadata snapshot operations, including saving snapshots, restoring from snapshots, and viewing snapshot information.
+#[derive(Subcommand)]
+pub enum SnapshotCommand {
+    /// Save a snapshot of the current metadata state to a specified location.
+    Save(SaveCommand),
+    /// Restore metadata from a snapshot.
+    Restore(RestoreCommand),
+    /// Explore metadata from a snapshot.
+    Info(InfoCommand),
 }

-impl MetaConnection {
-    pub async fn build(&self) -> Result<KvBackendRef, BoxedError> {
-        let max_txn_ops = self.max_txn_ops;
-        let store_addrs = &self.store_addrs;
-        if store_addrs.is_empty() {
-            KvBackendNotSetSnafu { backend: "all" }
-                .fail()
-                .map_err(BoxedError::new)
-        } else {
-            let kvbackend = match self.backend {
-                Some(BackendImpl::EtcdStore) => {
-                    let etcd_client = create_etcd_client(store_addrs)
-                        .await
-                        .map_err(BoxedError::new)?;
-                    Ok(EtcdStore::with_etcd_client(etcd_client, max_txn_ops))
-                }
-                #[cfg(feature = "pg_kvbackend")]
-                Some(BackendImpl::PostgresStore) => {
-                    let table_name = &self.meta_table_name;
-                    let pool = meta_srv::bootstrap::create_postgres_pool(store_addrs)
-                        .await
-                        .map_err(BoxedError::new)?;
-                    Ok(common_meta::kv_backend::rds::PgStore::with_pg_pool(
-                        pool,
-                        table_name,
-                        max_txn_ops,
-                    )
-                    .await
-                    .map_err(BoxedError::new)?)
-                }
-                #[cfg(feature = "mysql_kvbackend")]
-                Some(BackendImpl::MysqlStore) => {
-                    let table_name = &self.meta_table_name;
-                    let pool = meta_srv::bootstrap::create_mysql_pool(store_addrs)
-                        .await
-                        .map_err(BoxedError::new)?;
-                    Ok(common_meta::kv_backend::rds::MySqlStore::with_mysql_pool(
-                        pool,
-                        table_name,
-                        max_txn_ops,
-                    )
-                    .await
-                    .map_err(BoxedError::new)?)
-                }
-                _ => KvBackendNotSetSnafu { backend: "all" }
-                    .fail()
-                    .map_err(BoxedError::new),
-            };
-            if self.store_key_prefix.is_empty() {
-                kvbackend
-            } else {
-                let chroot_kvbackend =
-                    ChrootKvBackend::new(self.store_key_prefix.as_bytes().to_vec(), kvbackend?);
-                Ok(Arc::new(chroot_kvbackend))
-            }
+impl SnapshotCommand {
+    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
+        match self {
+            SnapshotCommand::Save(cmd) => cmd.build().await,
+            SnapshotCommand::Restore(cmd) => cmd.build().await,
+            SnapshotCommand::Info(cmd) => cmd.build().await,
        }
    }
 }
@@ -170,10 +112,10 @@ impl S3Config {
 /// It will dump the metadata snapshot to local file or s3 bucket.
 /// The snapshot file will be in binary format.
 #[derive(Debug, Default, Parser)]
-pub struct MetaSnapshotCommand {
-    /// The connection to the metadata store.
+pub struct SaveCommand {
+    /// The store configuration.
    #[clap(flatten)]
-    connection: MetaConnection,
+    store: StoreConfig,
    /// The s3 config.
    #[clap(flatten)]
    s3_config: S3Config,
@@ -196,9 +138,9 @@ fn create_local_file_object_store(root: &str) -> Result<ObjectStore, BoxedError>
    Ok(object_store)
 }

-impl MetaSnapshotCommand {
+impl SaveCommand {
    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
-        let kvbackend = self.connection.build().await?;
+        let kvbackend = self.store.build().await?;
        let output_dir = &self.output_dir;
        let object_store = self.s3_config.build(output_dir).map_err(BoxedError::new)?;
        if let Some(store) = object_store {
@@ -218,7 +160,7 @@ impl MetaSnapshotCommand {
    }
 }

-pub struct MetaSnapshotTool {
+struct MetaSnapshotTool {
    inner: MetadataSnapshotManager,
    target_file: String,
 }
@@ -234,14 +176,16 @@ impl Tool for MetaSnapshotTool {
    }
 }

-/// Restore metadata snapshot tool.
-/// This tool is used to restore metadata snapshot from etcd, pg or mysql.
-/// It will restore the metadata snapshot from local file or s3 bucket.
+/// Restore metadata from a snapshot file.
+///
+/// This command restores the metadata state from a previously saved snapshot.
+/// The snapshot can be loaded from either a local file system or an S3 bucket,
+/// depending on the provided configuration.
 #[derive(Debug, Default, Parser)]
-pub struct MetaRestoreCommand {
-    /// The connection to the metadata store.
+pub struct RestoreCommand {
+    /// The store configuration.
    #[clap(flatten)]
-    connection: MetaConnection,
+    store: StoreConfig,
    /// The s3 config.
    #[clap(flatten)]
    s3_config: S3Config,
@@ -255,9 +199,9 @@ pub struct MetaRestoreCommand {
    force: bool,
 }

-impl MetaRestoreCommand {
+impl RestoreCommand {
    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
-        let kvbackend = self.connection.build().await?;
+        let kvbackend = self.store.build().await?;
        let input_dir = &self.input_dir;
        let object_store = self.s3_config.build(input_dir).map_err(BoxedError::new)?;
        if let Some(store) = object_store {
@@ -279,7 +223,7 @@ impl MetaRestoreCommand {
    }
 }

-pub struct MetaRestoreTool {
+struct MetaRestoreTool {
    inner: MetadataSnapshotManager,
    source_file: String,
    force: bool,
@@ -327,3 +271,93 @@ impl Tool for MetaRestoreTool {
        }
    }
 }
+
+/// Explore metadata from a snapshot file.
+///
+/// This command allows filtering the metadata by a specific key and limiting the number of results.
+/// It prints the filtered metadata to the console.
+#[derive(Debug, Default, Parser)]
+pub struct InfoCommand {
+    /// The s3 config.
+    #[clap(flatten)]
+    s3_config: S3Config,
+    /// The name of the target snapshot file. we will add the file extension automatically.
+    #[clap(long, default_value = "metadata_snapshot")]
+    file_name: String,
+    /// The query string to filter the metadata.
+    #[clap(long, default_value = "*")]
+    inspect_key: String,
+    /// The limit of the metadata to query.
+    #[clap(long)]
+    limit: Option<usize>,
+}
+
+struct MetaInfoTool {
+    inner: ObjectStore,
+    source_file: String,
+    inspect_key: String,
+    limit: Option<usize>,
+}
+
+#[async_trait]
+impl Tool for MetaInfoTool {
+    #[allow(clippy::print_stdout)]
+    async fn do_work(&self) -> std::result::Result<(), BoxedError> {
+        let result = MetadataSnapshotManager::info(
+            &self.inner,
+            &self.source_file,
+            &self.inspect_key,
+            self.limit,
+        )
+        .await
+        .map_err(BoxedError::new)?;
+        for item in result {
+            println!("{}", item);
+        }
+        Ok(())
+    }
+}
+
+impl InfoCommand {
+    fn decide_object_store_root_for_local_store(
+        file_path: &str,
+    ) -> Result<(&str, &str), BoxedError> {
+        let path = Path::new(file_path);
+        let parent = path
+            .parent()
+            .and_then(|p| p.to_str())
+            .context(InvalidFilePathSnafu { msg: file_path })
+            .map_err(BoxedError::new)?;
+        let file_name = path
+            .file_name()
+            .and_then(|f| f.to_str())
+            .context(InvalidFilePathSnafu { msg: file_path })
+            .map_err(BoxedError::new)?;
+        let root = if parent.is_empty() { "." } else { parent };
+        Ok((root, file_name))
+    }
+
+    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
+        let object_store = self.s3_config.build("").map_err(BoxedError::new)?;
+        if let Some(store) = object_store {
+            let tool = MetaInfoTool {
+                inner: store,
+                source_file: self.file_name.clone(),
+                inspect_key: self.inspect_key.clone(),
+                limit: self.limit,
+            };
+            Ok(Box::new(tool))
+        } else {
+            let (root, file_name) =
+                Self::decide_object_store_root_for_local_store(&self.file_name)?;
+            let object_store = create_local_file_object_store(root)?;
+            let tool = MetaInfoTool {
+                inner: object_store,
+                source_file: file_name.to_string(),
+                inspect_key: self.inspect_key.clone(),
+                limit: self.limit,
+            };
+            Ok(Box::new(tool))
+        }
+    }
+}
--- a/src/client/src/client.rs
+++ b/src/client/src/client.rs
@@ -162,12 +162,23 @@ impl Client {
            .as_bytes() as usize
    }

-    pub fn make_flight_client(&self) -> Result<FlightClient> {
+    pub fn make_flight_client(
+        &self,
+        send_compression: bool,
+        accept_compression: bool,
+    ) -> Result<FlightClient> {
        let (addr, channel) = self.find_channel()?;

-        let client = FlightServiceClient::new(channel)
+        let mut client = FlightServiceClient::new(channel)
            .max_decoding_message_size(self.max_grpc_recv_message_size())
            .max_encoding_message_size(self.max_grpc_send_message_size());
+        // todo(hl): support compression methods.
+        if send_compression {
+            client = client.send_compressed(CompressionEncoding::Zstd);
+        }
+        if accept_compression {
+            client = client.accept_compressed(CompressionEncoding::Zstd);
+        }

        Ok(FlightClient { addr, client })
    }
--- a/src/client/src/client_manager.rs
+++ b/src/client/src/client_manager.rs
@@ -49,7 +49,16 @@ impl NodeManager for NodeClients {
    async fn datanode(&self, datanode: &Peer) -> DatanodeRef {
        let client = self.get_client(datanode).await;

-        Arc::new(RegionRequester::new(client))
+        let ChannelConfig {
+            send_compression,
+            accept_compression,
+            ..
+        } = self.channel_manager.config();
+        Arc::new(RegionRequester::new(
+            client,
+            *send_compression,
+            *accept_compression,
+        ))
    }

    async fn flownode(&self, flownode: &Peer) -> FlownodeRef {
--- a/src/client/src/database.rs
+++ b/src/client/src/database.rs
@@ -287,7 +287,7 @@ impl Database {
        let mut request = tonic::Request::new(request);
        Self::put_hints(request.metadata_mut(), hints)?;

-        let mut client = self.client.make_flight_client()?;
+        let mut client = self.client.make_flight_client(false, false)?;

        let response = client.mut_inner().do_get(request).await.or_else(|e| {
            let tonic_code = e.code();
@@ -409,7 +409,7 @@ impl Database {
            MetadataValue::from_str(db_to_put).context(InvalidTonicMetadataValueSnafu)?,
        );

-        let mut client = self.client.make_flight_client()?;
+        let mut client = self.client.make_flight_client(false, false)?;
        let response = client.mut_inner().do_put(request).await?;
        let response = response
            .into_inner()
--- a/src/client/src/region.rs
+++ b/src/client/src/region.rs
@@ -46,6 +46,8 @@ use crate::{metrics, Client, Error};
 #[derive(Debug)]
 pub struct RegionRequester {
    client: Client,
+    send_compression: bool,
+    accept_compression: bool,
 }

 #[async_trait]
@@ -89,12 +91,18 @@ impl Datanode for RegionRequester {
 }

 impl RegionRequester {
-    pub fn new(client: Client) -> Self {
-        Self { client }
+    pub fn new(client: Client, send_compression: bool, accept_compression: bool) -> Self {
+        Self {
+            client,
+            send_compression,
+            accept_compression,
+        }
    }

    pub async fn do_get_inner(&self, ticket: Ticket) -> Result<SendableRecordBatchStream> {
-        let mut flight_client = self.client.make_flight_client()?;
+        let mut flight_client = self
+            .client
+            .make_flight_client(self.send_compression, self.accept_compression)?;
        let response = flight_client
            .mut_inner()
            .do_get(ticket)
--- a/src/cmd/src/cli.rs
+++ b/src/cmd/src/cli.rs
@@ -146,6 +146,7 @@ mod tests {
        let output_dir = tempfile::tempdir().unwrap();
        let cli = cli::Command::parse_from([
            "cli",
+            "data",
            "export",
            "--addr",
            "127.0.0.1:4000",
--- a/src/cmd/src/flownode.rs
+++ b/src/cmd/src/flownode.rs
@@ -323,6 +323,7 @@ impl StartCommand {
            cached_meta_backend.clone(),
            layered_cache_registry.clone(),
            None,
+            None,
        );

        let table_metadata_manager =
--- a/src/cmd/src/frontend.rs
+++ b/src/cmd/src/frontend.rs
@@ -20,6 +20,7 @@ use async_trait::async_trait;
 use cache::{build_fundamental_cache_registry, with_default_composite_cache_registry};
 use catalog::information_extension::DistributedInformationExtension;
 use catalog::kvbackend::{CachedKvBackendBuilder, KvBackendCatalogManager, MetaKvBackend};
+use catalog::process_manager::ProcessManager;
 use clap::Parser;
 use client::client_manager::NodeClients;
 use common_base::Plugins;
@@ -38,6 +39,7 @@ use frontend::heartbeat::HeartbeatTask;
 use frontend::instance::builder::FrontendBuilder;
 use frontend::server::Services;
 use meta_client::{MetaClientOptions, MetaClientType};
+use servers::addrs;
 use servers::export_metrics::ExportMetricsTask;
 use servers::tls::{TlsMode, TlsOption};
 use snafu::{OptionExt, ResultExt};
@@ -342,11 +344,17 @@ impl StartCommand {

        let information_extension =
            Arc::new(DistributedInformationExtension::new(meta_client.clone()));
+
+        let process_manager = Arc::new(ProcessManager::new(
+            addrs::resolve_addr(&opts.grpc.bind_addr, Some(&opts.grpc.server_addr)),
+            Some(meta_client.clone()),
+        ));
        let catalog_manager = KvBackendCatalogManager::new(
            information_extension,
            cached_meta_backend.clone(),
            layered_cache_registry.clone(),
            None,
+            Some(process_manager.clone()),
        );

        let executor = HandlerGroupExecutor::new(vec![
@@ -364,12 +372,16 @@ impl StartCommand {

        // frontend to datanode need not timeout.
        // Some queries are expected to take long time.
-        let channel_config = ChannelConfig {
+        let mut channel_config = ChannelConfig {
            timeout: None,
            tcp_nodelay: opts.datanode.client.tcp_nodelay,
            connect_timeout: Some(opts.datanode.client.connect_timeout),
            ..Default::default()
        };
+        if opts.grpc.flight_compression.transport_compression() {
+            channel_config.accept_compression = true;
+            channel_config.send_compression = true;
+        }
        let client = NodeClients::new(channel_config);

        let instance = FrontendBuilder::new(
@@ -379,6 +391,7 @@ impl StartCommand {
            catalog_manager,
            Arc::new(client),
            meta_client,
+            process_manager,
        )
        .with_plugin(plugins.clone())
        .with_local_cache_invalidator(layered_cache_registry)
--- a/src/cmd/src/metasrv.rs
+++ b/src/cmd/src/metasrv.rs
@@ -237,12 +237,20 @@ impl StartCommand {
            tokio_console_addr: global_options.tokio_console_addr.clone(),
        };

+        #[allow(deprecated)]
        if let Some(addr) = &self.rpc_bind_addr {
            opts.bind_addr.clone_from(addr);
+            opts.grpc.bind_addr.clone_from(addr);
+        } else if !opts.bind_addr.is_empty() {
+            opts.grpc.bind_addr.clone_from(&opts.bind_addr);
        }

+        #[allow(deprecated)]
        if let Some(addr) = &self.rpc_server_addr {
            opts.server_addr.clone_from(addr);
+            opts.grpc.server_addr.clone_from(addr);
+        } else if !opts.server_addr.is_empty() {
+            opts.grpc.server_addr.clone_from(&opts.server_addr);
        }

        if let Some(addrs) = &self.store_addrs {
@@ -319,7 +327,7 @@ impl StartCommand {

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

        info!("Metasrv options: {:#?}", opts);

@@ -363,7 +371,7 @@ mod tests {
        };

        let options = cmd.load_options(&Default::default()).unwrap().component;
-        assert_eq!("127.0.0.1:3002".to_string(), options.bind_addr);
+        assert_eq!("127.0.0.1:3002".to_string(), options.grpc.bind_addr);
        assert_eq!(vec!["127.0.0.1:2380".to_string()], options.store_addrs);
        assert_eq!(SelectorType::LoadBased, options.selector);
    }
@@ -396,8 +404,8 @@ mod tests {
        };

        let options = cmd.load_options(&Default::default()).unwrap().component;
-        assert_eq!("127.0.0.1:3002".to_string(), options.bind_addr);
-        assert_eq!("127.0.0.1:3002".to_string(), options.server_addr);
+        assert_eq!("127.0.0.1:3002".to_string(), options.grpc.bind_addr);
+        assert_eq!("127.0.0.1:3002".to_string(), options.grpc.server_addr);
        assert_eq!(vec!["127.0.0.1:2379".to_string()], options.store_addrs);
        assert_eq!(SelectorType::LeaseBased, options.selector);
        assert_eq!("debug", options.logging.level.as_ref().unwrap());
@@ -509,10 +517,10 @@ mod tests {
                let opts = command.load_options(&Default::default()).unwrap().component;

                // Should be read from env, env > default values.
-                assert_eq!(opts.bind_addr, "127.0.0.1:14002");
+                assert_eq!(opts.grpc.bind_addr, "127.0.0.1:14002");

                // Should be read from config file, config file > env > default values.
-                assert_eq!(opts.server_addr, "127.0.0.1:3002");
+                assert_eq!(opts.grpc.server_addr, "127.0.0.1:3002");

                // Should be read from cli, cli > config file > env > default values.
                assert_eq!(opts.http.addr, "127.0.0.1:14000");
--- a/src/cmd/src/standalone.rs
+++ b/src/cmd/src/standalone.rs
@@ -21,6 +21,7 @@ use async_trait::async_trait;
 use cache::{build_fundamental_cache_registry, with_default_composite_cache_registry};
 use catalog::information_schema::InformationExtension;
 use catalog::kvbackend::KvBackendCatalogManager;
+use catalog::process_manager::ProcessManager;
 use clap::Parser;
 use client::api::v1::meta::RegionRole;
 use common_base::readable_size::ReadableSize;
@@ -526,11 +527,14 @@ impl StartCommand {
            datanode.region_server(),
            procedure_manager.clone(),
        ));
+
+        let process_manager = Arc::new(ProcessManager::new(opts.grpc.server_addr.clone(), None));
        let catalog_manager = KvBackendCatalogManager::new(
            information_extension.clone(),
            kv_backend.clone(),
            layered_cache_registry.clone(),
            Some(procedure_manager.clone()),
+            Some(process_manager.clone()),
        );

        let table_metadata_manager =
@@ -620,6 +624,7 @@ impl StartCommand {
            catalog_manager.clone(),
            node_manager.clone(),
            ddl_task_executor.clone(),
+            process_manager,
        )
        .with_plugin(plugins.clone())
        .try_build()
@@ -647,7 +652,7 @@ impl StartCommand {
            node_manager,
        )
        .await
-        .context(error::StartFlownodeSnafu)?;
+        .context(StartFlownodeSnafu)?;
        flow_streaming_engine.set_frontend_invoker(invoker).await;

        let export_metrics_task = ExportMetricsTask::try_new(&opts.export_metrics, Some(&plugins))
--- a/src/cmd/tests/load_config_test.rs
+++ b/src/cmd/tests/load_config_test.rs
@@ -95,7 +95,7 @@ fn test_load_datanode_example_config() {
                ..Default::default()
            },
            export_metrics: ExportMetricsOption {
-                self_import: Some(Default::default()),
+                self_import: None,
                remote_write: Some(Default::default()),
                ..Default::default()
            },
@@ -148,7 +148,7 @@ fn test_load_frontend_example_config() {
                },
            },
            export_metrics: ExportMetricsOption {
-                self_import: Some(Default::default()),
+                self_import: None,
                remote_write: Some(Default::default()),
                ..Default::default()
            },
@@ -176,7 +176,11 @@ fn test_load_metasrv_example_config() {
        component: MetasrvOptions {
            selector: SelectorType::default(),
            data_home: DEFAULT_DATA_HOME.to_string(),
-            server_addr: "127.0.0.1:3002".to_string(),
+            grpc: GrpcOptions {
+                bind_addr: "127.0.0.1:3002".to_string(),
+                server_addr: "127.0.0.1:3002".to_string(),
+                ..Default::default()
+            },
            logging: LoggingOptions {
                dir: Path::new(DEFAULT_DATA_HOME)
                    .join(DEFAULT_LOGGING_DIR)
@@ -195,7 +199,7 @@ fn test_load_metasrv_example_config() {
                },
            },
            export_metrics: ExportMetricsOption {
-                self_import: Some(Default::default()),
+                self_import: None,
                remote_write: Some(Default::default()),
                ..Default::default()
            },
--- a/src/common/catalog/src/consts.rs
+++ b/src/common/catalog/src/consts.rs
@@ -102,6 +102,8 @@ pub const INFORMATION_SCHEMA_FLOW_TABLE_ID: u32 = 33;
 pub const INFORMATION_SCHEMA_PROCEDURE_INFO_TABLE_ID: u32 = 34;
 /// id for information_schema.region_statistics
 pub const INFORMATION_SCHEMA_REGION_STATISTICS_TABLE_ID: u32 = 35;
+/// id for information_schema.process_list
+pub const INFORMATION_SCHEMA_PROCESS_LIST_TABLE_ID: u32 = 36;

 // ----- End of information_schema tables -----

--- a/src/common/frontend/Cargo.toml
+++ b/src/common/frontend/Cargo.toml
@@ -7,5 +7,13 @@ license.workspace = true
 [dependencies]
 async-trait.workspace = true
 common-error.workspace = true
+common-grpc.workspace = true
 common-macro.workspace = true
+common-meta.workspace = true
+greptime-proto.workspace = true
+meta-client.workspace = true
 snafu.workspace = true
+tonic.workspace = true
+
+[dev-dependencies]
+tokio.workspace = true
--- a/src/common/frontend/src/error.rs
+++ b/src/common/frontend/src/error.rs
@@ -27,6 +27,35 @@ pub enum Error {
        location: Location,
        source: BoxedError,
    },
+
+    #[snafu(display("Failed to list nodes from metasrv"))]
+    Meta {
+        source: Box<meta_client::error::Error>,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Failed to parse process id: {}", s))]
+    ParseProcessId {
+        s: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Failed to invoke list process service"))]
+    ListProcess {
+        #[snafu(source)]
+        error: tonic::Status,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Failed to invoke list process service"))]
+    CreateChannel {
+        source: common_grpc::error::Error,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }

 pub type Result<T> = std::result::Result<T, Error>;
@@ -36,6 +65,10 @@ impl ErrorExt for Error {
        use Error::*;
        match self {
            External { source, .. } => source.status_code(),
+            Meta { source, .. } => source.status_code(),
+            ParseProcessId { .. } => StatusCode::InvalidArguments,
+            ListProcess { .. } => StatusCode::External,
+            CreateChannel { source, .. } => source.status_code(),
        }
    }

--- a/src/common/frontend/src/lib.rs
+++ b/src/common/frontend/src/lib.rs
@@ -12,4 +12,41 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use std::fmt::{Display, Formatter};
+use std::str::FromStr;
+
+use snafu::OptionExt;
+
 pub mod error;
+pub mod selector;
+
+#[derive(Debug, Clone, Eq, PartialEq)]
+pub struct DisplayProcessId {
+    pub server_addr: String,
+    pub id: u64,
+}
+
+impl Display for DisplayProcessId {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}/{}", self.server_addr, self.id)
+    }
+}
+
+impl TryFrom<&str> for DisplayProcessId {
+    type Error = error::Error;
+
+    fn try_from(value: &str) -> Result<Self, Self::Error> {
+        let mut split = value.split('/');
+        let server_addr = split
+            .next()
+            .context(error::ParseProcessIdSnafu { s: value })?
+            .to_string();
+        let id = split
+            .next()
+            .context(error::ParseProcessIdSnafu { s: value })?;
+        let id = u64::from_str(id)
+            .ok()
+            .context(error::ParseProcessIdSnafu { s: value })?;
+        Ok(DisplayProcessId { server_addr, id })
+    }
+}
--- a/src/common/frontend/src/selector.rs
+++ b/src/common/frontend/src/selector.rs
@@ -0,0 +1,98 @@
+// 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::time::Duration;
+
+use common_grpc::channel_manager::{ChannelConfig, ChannelManager};
+use common_meta::cluster::{ClusterInfo, NodeInfo, Role};
+use greptime_proto::v1::frontend::{frontend_client, ListProcessRequest, ListProcessResponse};
+use meta_client::MetaClientRef;
+use snafu::ResultExt;
+
+use crate::error;
+use crate::error::{MetaSnafu, Result};
+
+pub type FrontendClientPtr = Box<dyn FrontendClient>;
+
+#[async_trait::async_trait]
+pub trait FrontendClient: Send {
+    async fn list_process(&mut self, req: ListProcessRequest) -> Result<ListProcessResponse>;
+}
+
+#[async_trait::async_trait]
+impl FrontendClient for frontend_client::FrontendClient<tonic::transport::channel::Channel> {
+    async fn list_process(&mut self, req: ListProcessRequest) -> Result<ListProcessResponse> {
+        let response: ListProcessResponse = frontend_client::FrontendClient::<
+            tonic::transport::channel::Channel,
+        >::list_process(self, req)
+        .await
+        .context(error::ListProcessSnafu)?
+        .into_inner();
+        Ok(response)
+    }
+}
+
+#[async_trait::async_trait]
+pub trait FrontendSelector {
+    async fn select<F>(&self, predicate: F) -> Result<Vec<FrontendClientPtr>>
+    where
+        F: Fn(&NodeInfo) -> bool + Send;
+}
+
+#[derive(Debug, Clone)]
+pub struct MetaClientSelector {
+    meta_client: MetaClientRef,
+    channel_manager: ChannelManager,
+}
+
+#[async_trait::async_trait]
+impl FrontendSelector for MetaClientSelector {
+    async fn select<F>(&self, predicate: F) -> Result<Vec<FrontendClientPtr>>
+    where
+        F: Fn(&NodeInfo) -> bool + Send,
+    {
+        let nodes = self
+            .meta_client
+            .list_nodes(Some(Role::Frontend))
+            .await
+            .map_err(Box::new)
+            .context(MetaSnafu)?;
+
+        nodes
+            .into_iter()
+            .filter(predicate)
+            .map(|node| {
+                let channel = self
+                    .channel_manager
+                    .get(node.peer.addr)
+                    .context(error::CreateChannelSnafu)?;
+                let client = frontend_client::FrontendClient::new(channel);
+                Ok(Box::new(client) as FrontendClientPtr)
+            })
+            .collect::<Result<Vec<_>>>()
+    }
+}
+
+impl MetaClientSelector {
+    pub fn new(meta_client: MetaClientRef) -> Self {
+        let cfg = ChannelConfig::new()
+            .connect_timeout(Duration::from_secs(30))
+            .timeout(Duration::from_secs(30));
+        let channel_manager = ChannelManager::with_config(cfg);
+        Self {
+            meta_client,
+            channel_manager,
+        }
+    }
+}
--- a/src/common/function/src/aggrs.rs
+++ b/src/common/function/src/aggrs.rs
@@ -12,11 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-mod geo_path;
-mod hll;
-mod uddsketch_state;
-
-pub use geo_path::{GeoPathAccumulator, GEO_PATH_NAME};
-pub(crate) use hll::HllStateType;
-pub use hll::{HllState, HLL_MERGE_NAME, HLL_NAME};
-pub use uddsketch_state::{UddSketchState, UDDSKETCH_MERGE_NAME, UDDSKETCH_STATE_NAME};
+pub mod approximate;
+#[cfg(feature = "geo")]
+pub mod geo;
+pub mod vector;
--- a/src/common/function/src/aggrs/approximate.rs
+++ b/src/common/function/src/aggrs/approximate.rs
@@ -0,0 +1,32 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::function_registry::FunctionRegistry;
+
+pub(crate) mod hll;
+mod uddsketch;
+
+pub(crate) struct ApproximateFunction;
+
+impl ApproximateFunction {
+    pub fn register(registry: &FunctionRegistry) {
+        // uddsketch
+        registry.register_aggr(uddsketch::UddSketchState::state_udf_impl());
+        registry.register_aggr(uddsketch::UddSketchState::merge_udf_impl());
+
+        // hll
+        registry.register_aggr(hll::HllState::state_udf_impl());
+        registry.register_aggr(hll::HllState::merge_udf_impl());
+    }
+}
--- a/src/common/function/src/aggrs/approximate/hll.rs
+++ b/src/common/function/src/aggrs/approximate/hll.rs
--- a/src/common/function/src/aggrs/approximate/uddsketch.rs
+++ b/src/common/function/src/aggrs/approximate/uddsketch.rs
--- a/src/common/function/src/aggrs/geo.rs
+++ b/src/common/function/src/aggrs/geo.rs
@@ -0,0 +1,27 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::function_registry::FunctionRegistry;
+
+mod encoding;
+mod geo_path;
+
+pub(crate) struct GeoFunction;
+
+impl GeoFunction {
+    pub fn register(registry: &FunctionRegistry) {
+        registry.register_aggr(geo_path::GeoPathAccumulator::uadf_impl());
+        registry.register_aggr(encoding::JsonPathAccumulator::uadf_impl());
+    }
+}
--- a/src/common/function/src/scalars/geo/encoding.rs
+++ b/src/common/function/src/scalars/geo/encoding.rs
@@ -19,9 +19,12 @@ use common_error::status_code::StatusCode;
 use common_macro::{as_aggr_func_creator, AggrFuncTypeStore};
 use common_query::error::{self, InvalidInputStateSnafu, Result};
 use common_query::logical_plan::accumulator::AggrFuncTypeStore;
-use common_query::logical_plan::{Accumulator, AggregateFunctionCreator};
+use common_query::logical_plan::{
+    create_aggregate_function, Accumulator, AggregateFunctionCreator,
+};
 use common_query::prelude::AccumulatorCreatorFunction;
 use common_time::Timestamp;
+use datafusion_expr::AggregateUDF;
 use datatypes::prelude::ConcreteDataType;
 use datatypes::value::{ListValue, Value};
 use datatypes::vectors::VectorRef;
@@ -47,6 +50,16 @@ impl JsonPathAccumulator {
            timestamp_type,
        }
    }
+
+    /// Create a new `AggregateUDF` for the `json_encode_path` aggregate function.
+    pub fn uadf_impl() -> AggregateUDF {
+        create_aggregate_function(
+            "json_encode_path".to_string(),
+            3,
+            Arc::new(JsonPathEncodeFunctionCreator::default()),
+        )
+        .into()
+    }
 }

 impl Accumulator for JsonPathAccumulator {
--- a/src/common/function/src/aggrs/geo/geo_path.rs
+++ b/src/common/function/src/aggrs/geo/geo_path.rs
@@ -47,7 +47,7 @@ impl GeoPathAccumulator {
        Self::default()
    }

-    pub fn udf_impl() -> AggregateUDF {
+    pub fn uadf_impl() -> AggregateUDF {
        create_udaf(
            GEO_PATH_NAME,
            // Input types: lat, lng, timestamp
--- a/src/common/function/src/aggrs/vector.rs
+++ b/src/common/function/src/aggrs/vector.rs
@@ -0,0 +1,29 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::aggrs::vector::product::VectorProduct;
+use crate::aggrs::vector::sum::VectorSum;
+use crate::function_registry::FunctionRegistry;
+
+mod product;
+mod sum;
+
+pub(crate) struct VectorFunction;
+
+impl VectorFunction {
+    pub fn register(registry: &FunctionRegistry) {
+        registry.register_aggr(VectorSum::uadf_impl());
+        registry.register_aggr(VectorProduct::uadf_impl());
+    }
+}
--- a/src/common/function/src/scalars/vector/product.rs
+++ b/src/common/function/src/scalars/vector/product.rs
@@ -16,8 +16,11 @@ use std::sync::Arc;

 use common_macro::{as_aggr_func_creator, AggrFuncTypeStore};
 use common_query::error::{CreateAccumulatorSnafu, Error, InvalidFuncArgsSnafu};
-use common_query::logical_plan::{Accumulator, AggregateFunctionCreator};
+use common_query::logical_plan::{
+    create_aggregate_function, Accumulator, AggregateFunctionCreator,
+};
 use common_query::prelude::AccumulatorCreatorFunction;
+use datafusion_expr::AggregateUDF;
 use datatypes::prelude::{ConcreteDataType, Value, *};
 use datatypes::vectors::VectorRef;
 use nalgebra::{Const, DVectorView, Dyn, OVector};
@@ -75,6 +78,16 @@ impl AggregateFunctionCreator for VectorProductCreator {
 }

 impl VectorProduct {
+    /// Create a new `AggregateUDF` for the `vec_product` aggregate function.
+    pub fn uadf_impl() -> AggregateUDF {
+        create_aggregate_function(
+            "vec_product".to_string(),
+            1,
+            Arc::new(VectorProductCreator::default()),
+        )
+        .into()
+    }
+
    fn inner(&mut self, len: usize) -> &mut OVector<f32, Dyn> {
        self.product.get_or_insert_with(|| {
            OVector::from_iterator_generic(Dyn(len), Const::<1>, (0..len).map(|_| 1.0))
--- a/src/common/function/src/scalars/vector/sum.rs
+++ b/src/common/function/src/scalars/vector/sum.rs
@@ -16,8 +16,11 @@ use std::sync::Arc;

 use common_macro::{as_aggr_func_creator, AggrFuncTypeStore};
 use common_query::error::{CreateAccumulatorSnafu, Error, InvalidFuncArgsSnafu};
-use common_query::logical_plan::{Accumulator, AggregateFunctionCreator};
+use common_query::logical_plan::{
+    create_aggregate_function, Accumulator, AggregateFunctionCreator,
+};
 use common_query::prelude::AccumulatorCreatorFunction;
+use datafusion_expr::AggregateUDF;
 use datatypes::prelude::{ConcreteDataType, Value, *};
 use datatypes::vectors::VectorRef;
 use nalgebra::{Const, DVectorView, Dyn, OVector};
@@ -25,6 +28,7 @@ use snafu::ensure;

 use crate::scalars::vector::impl_conv::{as_veclit, as_veclit_if_const, veclit_to_binlit};

+/// The accumulator for the `vec_sum` aggregate function.
 #[derive(Debug, Default)]
 pub struct VectorSum {
    sum: Option<OVector<f32, Dyn>>,
@@ -74,6 +78,16 @@ impl AggregateFunctionCreator for VectorSumCreator {
 }

 impl VectorSum {
+    /// Create a new `AggregateUDF` for the `vec_sum` aggregate function.
+    pub fn uadf_impl() -> AggregateUDF {
+        create_aggregate_function(
+            "vec_sum".to_string(),
+            1,
+            Arc::new(VectorSumCreator::default()),
+        )
+        .into()
+    }
+
    fn inner(&mut self, len: usize) -> &mut OVector<f32, Dyn> {
        self.sum
            .get_or_insert_with(|| OVector::zeros_generic(Dyn(len), Const::<1>))
--- a/src/common/function/src/function_factory.rs
+++ b/src/common/function/src/function_factory.rs
@@ -0,0 +1,63 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use datafusion_expr::ScalarUDF;
+
+use crate::function::{FunctionContext, FunctionRef};
+use crate::scalars::udf::create_udf;
+
+/// A factory for creating `ScalarUDF` that require a function context.
+#[derive(Clone)]
+pub struct ScalarFunctionFactory {
+    name: String,
+    factory: Arc<dyn Fn(FunctionContext) -> ScalarUDF + Send + Sync>,
+}
+
+impl ScalarFunctionFactory {
+    /// Returns the name of the function.
+    pub fn name(&self) -> &str {
+        &self.name
+    }
+
+    /// Returns a `ScalarUDF` when given a function context.
+    pub fn provide(&self, ctx: FunctionContext) -> ScalarUDF {
+        (self.factory)(ctx)
+    }
+}
+
+impl From<ScalarUDF> for ScalarFunctionFactory {
+    fn from(df_udf: ScalarUDF) -> Self {
+        let name = df_udf.name().to_string();
+        let func = Arc::new(move |_ctx| df_udf.clone());
+        Self {
+            name,
+            factory: func,
+        }
+    }
+}
+
+impl From<FunctionRef> for ScalarFunctionFactory {
+    fn from(func: FunctionRef) -> Self {
+        let name = func.name().to_string();
+        let func = Arc::new(move |ctx: FunctionContext| {
+            create_udf(func.clone(), ctx.query_ctx, ctx.state)
+        });
+        Self {
+            name,
+            factory: func,
+        }
+    }
+}
--- a/src/common/function/src/function_registry.rs
+++ b/src/common/function/src/function_registry.rs
@@ -16,11 +16,14 @@
 use std::collections::HashMap;
 use std::sync::{Arc, RwLock};

+use datafusion_expr::AggregateUDF;
 use once_cell::sync::Lazy;

 use crate::admin::AdminFunction;
-use crate::function::{AsyncFunctionRef, FunctionRef};
-use crate::scalars::aggregate::{AggregateFunctionMetaRef, AggregateFunctions};
+use crate::aggrs::approximate::ApproximateFunction;
+use crate::aggrs::vector::VectorFunction as VectorAggrFunction;
+use crate::function::{AsyncFunctionRef, Function, FunctionRef};
+use crate::function_factory::ScalarFunctionFactory;
 use crate::scalars::date::DateFunction;
 use crate::scalars::expression::ExpressionFunction;
 use crate::scalars::hll_count::HllCalcFunction;
@@ -31,18 +34,19 @@ use crate::scalars::matches_term::MatchesTermFunction;
 use crate::scalars::math::MathFunction;
 use crate::scalars::timestamp::TimestampFunction;
 use crate::scalars::uddsketch_calc::UddSketchCalcFunction;
-use crate::scalars::vector::VectorFunction;
+use crate::scalars::vector::VectorFunction as VectorScalarFunction;
 use crate::system::SystemFunction;

 #[derive(Default)]
 pub struct FunctionRegistry {
-    functions: RwLock<HashMap<String, FunctionRef>>,
+    functions: RwLock<HashMap<String, ScalarFunctionFactory>>,
    async_functions: RwLock<HashMap<String, AsyncFunctionRef>>,
-    aggregate_functions: RwLock<HashMap<String, AggregateFunctionMetaRef>>,
+    aggregate_functions: RwLock<HashMap<String, AggregateUDF>>,
 }

 impl FunctionRegistry {
-    pub fn register(&self, func: FunctionRef) {
+    pub fn register(&self, func: impl Into<ScalarFunctionFactory>) {
+        let func = func.into();
        let _ = self
            .functions
            .write()
@@ -50,6 +54,10 @@ impl FunctionRegistry {
            .insert(func.name().to_string(), func);
    }

+    pub fn register_scalar(&self, func: impl Function + 'static) {
+        self.register(Arc::new(func) as FunctionRef);
+    }
+
    pub fn register_async(&self, func: AsyncFunctionRef) {
        let _ = self
            .async_functions
@@ -58,6 +66,14 @@ impl FunctionRegistry {
            .insert(func.name().to_string(), func);
    }

+    pub fn register_aggr(&self, func: AggregateUDF) {
+        let _ = self
+            .aggregate_functions
+            .write()
+            .unwrap()
+            .insert(func.name().to_string(), func);
+    }
+
    pub fn get_async_function(&self, name: &str) -> Option<AsyncFunctionRef> {
        self.async_functions.read().unwrap().get(name).cloned()
    }
@@ -71,27 +87,16 @@ impl FunctionRegistry {
            .collect()
    }

-    pub fn register_aggregate_function(&self, func: AggregateFunctionMetaRef) {
-        let _ = self
-            .aggregate_functions
-            .write()
-            .unwrap()
-            .insert(func.name(), func);
-    }
-
-    pub fn get_aggr_function(&self, name: &str) -> Option<AggregateFunctionMetaRef> {
-        self.aggregate_functions.read().unwrap().get(name).cloned()
-    }
-
-    pub fn get_function(&self, name: &str) -> Option<FunctionRef> {
+    #[cfg(test)]
+    pub fn get_function(&self, name: &str) -> Option<ScalarFunctionFactory> {
        self.functions.read().unwrap().get(name).cloned()
    }

-    pub fn functions(&self) -> Vec<FunctionRef> {
+    pub fn scalar_functions(&self) -> Vec<ScalarFunctionFactory> {
        self.functions.read().unwrap().values().cloned().collect()
    }

-    pub fn aggregate_functions(&self) -> Vec<AggregateFunctionMetaRef> {
+    pub fn aggregate_functions(&self) -> Vec<AggregateUDF> {
        self.aggregate_functions
            .read()
            .unwrap()
@@ -112,9 +117,6 @@ pub static FUNCTION_REGISTRY: Lazy<Arc<FunctionRegistry>> = Lazy::new(|| {
    UddSketchCalcFunction::register(&function_registry);
    HllCalcFunction::register(&function_registry);

-    // Aggregate functions
-    AggregateFunctions::register(&function_registry);
-
    // Full text search function
    MatchesFunction::register(&function_registry);
    MatchesTermFunction::register(&function_registry);
@@ -127,15 +129,21 @@ pub static FUNCTION_REGISTRY: Lazy<Arc<FunctionRegistry>> = Lazy::new(|| {
    JsonFunction::register(&function_registry);

    // Vector related functions
-    VectorFunction::register(&function_registry);
+    VectorScalarFunction::register(&function_registry);
+    VectorAggrFunction::register(&function_registry);

    // Geo functions
    #[cfg(feature = "geo")]
    crate::scalars::geo::GeoFunctions::register(&function_registry);
+    #[cfg(feature = "geo")]
+    crate::aggrs::geo::GeoFunction::register(&function_registry);

    // Ip functions
    IpFunctions::register(&function_registry);

+    // Approximate functions
+    ApproximateFunction::register(&function_registry);
+
    Arc::new(function_registry)
 });

@@ -147,12 +155,11 @@ mod tests {
    #[test]
    fn test_function_registry() {
        let registry = FunctionRegistry::default();
-        let func = Arc::new(TestAndFunction);

        assert!(registry.get_function("test_and").is_none());
-        assert!(registry.functions().is_empty());
-        registry.register(func);
+        assert!(registry.scalar_functions().is_empty());
+        registry.register_scalar(TestAndFunction);
        let _ = registry.get_function("test_and").unwrap();
-        assert_eq!(1, registry.functions().len());
+        assert_eq!(1, registry.scalar_functions().len());
    }
 }
--- a/src/common/function/src/lib.rs
+++ b/src/common/function/src/lib.rs
@@ -18,13 +18,14 @@
 mod admin;
 mod flush_flow;
 mod macros;
-pub mod scalars;
 mod system;

-pub mod aggr;
+pub mod aggrs;
 pub mod function;
+pub mod function_factory;
 pub mod function_registry;
 pub mod handlers;
 pub mod helper;
+pub mod scalars;
 pub mod state;
 pub mod utils;
--- a/src/common/function/src/scalars.rs
+++ b/src/common/function/src/scalars.rs
@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-pub mod aggregate;
 pub(crate) mod date;
 pub mod expression;
 #[cfg(feature = "geo")]
--- a/src/common/function/src/scalars/aggregate.rs
+++ b/src/common/function/src/scalars/aggregate.rs
@@ -1,89 +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.
-
-//! # Deprecate Warning:
-//!
-//! This module is deprecated and will be removed in the future.
-//! All UDAF implementation here are not maintained and should
-//! not be used before they are refactored into the `src/aggr`
-//! version.
-
-use std::sync::Arc;
-
-use common_query::logical_plan::AggregateFunctionCreatorRef;
-
-use crate::function_registry::FunctionRegistry;
-use crate::scalars::vector::product::VectorProductCreator;
-use crate::scalars::vector::sum::VectorSumCreator;
-
-/// A function creates `AggregateFunctionCreator`.
-/// "Aggregator" *is* AggregatorFunction. Since the later one is long, we named an short alias for it.
-/// The two names might be used interchangeably.
-type AggregatorCreatorFunction = Arc<dyn Fn() -> AggregateFunctionCreatorRef + Send + Sync>;
-
-/// `AggregateFunctionMeta` dynamically creates AggregateFunctionCreator.
-#[derive(Clone)]
-pub struct AggregateFunctionMeta {
-    name: String,
-    args_count: u8,
-    creator: AggregatorCreatorFunction,
-}
-
-pub type AggregateFunctionMetaRef = Arc<AggregateFunctionMeta>;
-
-impl AggregateFunctionMeta {
-    pub fn new(name: &str, args_count: u8, creator: AggregatorCreatorFunction) -> Self {
-        Self {
-            name: name.to_string(),
-            args_count,
-            creator,
-        }
-    }
-
-    pub fn name(&self) -> String {
-        self.name.to_string()
-    }
-
-    pub fn args_count(&self) -> u8 {
-        self.args_count
-    }
-
-    pub fn create(&self) -> AggregateFunctionCreatorRef {
-        (self.creator)()
-    }
-}
-
-pub(crate) struct AggregateFunctions;
-
-impl AggregateFunctions {
-    pub fn register(registry: &FunctionRegistry) {
-        registry.register_aggregate_function(Arc::new(AggregateFunctionMeta::new(
-            "vec_sum",
-            1,
-            Arc::new(|| Arc::new(VectorSumCreator::default())),
-        )));
-        registry.register_aggregate_function(Arc::new(AggregateFunctionMeta::new(
-            "vec_product",
-            1,
-            Arc::new(|| Arc::new(VectorProductCreator::default())),
-        )));
-
-        #[cfg(feature = "geo")]
-        registry.register_aggregate_function(Arc::new(AggregateFunctionMeta::new(
-            "json_encode_path",
-            3,
-            Arc::new(|| Arc::new(super::geo::encoding::JsonPathEncodeFunctionCreator::default())),
-        )));
-    }
-}
--- a/src/common/function/src/scalars/date.rs
+++ b/src/common/function/src/scalars/date.rs
@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use std::sync::Arc;
 mod date_add;
 mod date_format;
 mod date_sub;
@@ -27,8 +26,8 @@ pub(crate) struct DateFunction;

 impl DateFunction {
    pub fn register(registry: &FunctionRegistry) {
-        registry.register(Arc::new(DateAddFunction));
-        registry.register(Arc::new(DateSubFunction));
-        registry.register(Arc::new(DateFormatFunction));
+        registry.register_scalar(DateAddFunction);
+        registry.register_scalar(DateSubFunction);
+        registry.register_scalar(DateFormatFunction);
    }
 }
--- a/src/common/function/src/scalars/expression.rs
+++ b/src/common/function/src/scalars/expression.rs
@@ -17,8 +17,6 @@ mod ctx;
 mod is_null;
 mod unary;

-use std::sync::Arc;
-
 pub use binary::scalar_binary_op;
 pub use ctx::EvalContext;
 pub use unary::scalar_unary_op;
@@ -30,6 +28,6 @@ pub(crate) struct ExpressionFunction;

 impl ExpressionFunction {
    pub fn register(registry: &FunctionRegistry) {
-        registry.register(Arc::new(IsNullFunction));
+        registry.register_scalar(IsNullFunction);
    }
 }
--- a/src/common/function/src/scalars/geo.rs
+++ b/src/common/function/src/scalars/geo.rs
@@ -12,11 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use std::sync::Arc;
-pub(crate) mod encoding;
 mod geohash;
 mod h3;
-mod helpers;
+pub(crate) mod helpers;
 mod measure;
 mod relation;
 mod s2;
@@ -29,57 +27,57 @@ pub(crate) struct GeoFunctions;
 impl GeoFunctions {
    pub fn register(registry: &FunctionRegistry) {
        // geohash
-        registry.register(Arc::new(geohash::GeohashFunction));
-        registry.register(Arc::new(geohash::GeohashNeighboursFunction));
+        registry.register_scalar(geohash::GeohashFunction);
+        registry.register_scalar(geohash::GeohashNeighboursFunction);

        // h3 index
-        registry.register(Arc::new(h3::H3LatLngToCell));
-        registry.register(Arc::new(h3::H3LatLngToCellString));
+        registry.register_scalar(h3::H3LatLngToCell);
+        registry.register_scalar(h3::H3LatLngToCellString);

        // h3 index inspection
-        registry.register(Arc::new(h3::H3CellBase));
-        registry.register(Arc::new(h3::H3CellIsPentagon));
-        registry.register(Arc::new(h3::H3StringToCell));
-        registry.register(Arc::new(h3::H3CellToString));
-        registry.register(Arc::new(h3::H3CellCenterLatLng));
-        registry.register(Arc::new(h3::H3CellResolution));
+        registry.register_scalar(h3::H3CellBase);
+        registry.register_scalar(h3::H3CellIsPentagon);
+        registry.register_scalar(h3::H3StringToCell);
+        registry.register_scalar(h3::H3CellToString);
+        registry.register_scalar(h3::H3CellCenterLatLng);
+        registry.register_scalar(h3::H3CellResolution);

        // h3 hierarchical grid
-        registry.register(Arc::new(h3::H3CellCenterChild));
-        registry.register(Arc::new(h3::H3CellParent));
-        registry.register(Arc::new(h3::H3CellToChildren));
-        registry.register(Arc::new(h3::H3CellToChildrenSize));
-        registry.register(Arc::new(h3::H3CellToChildPos));
-        registry.register(Arc::new(h3::H3ChildPosToCell));
-        registry.register(Arc::new(h3::H3CellContains));
+        registry.register_scalar(h3::H3CellCenterChild);
+        registry.register_scalar(h3::H3CellParent);
+        registry.register_scalar(h3::H3CellToChildren);
+        registry.register_scalar(h3::H3CellToChildrenSize);
+        registry.register_scalar(h3::H3CellToChildPos);
+        registry.register_scalar(h3::H3ChildPosToCell);
+        registry.register_scalar(h3::H3CellContains);

        // h3 grid traversal
-        registry.register(Arc::new(h3::H3GridDisk));
-        registry.register(Arc::new(h3::H3GridDiskDistances));
-        registry.register(Arc::new(h3::H3GridDistance));
-        registry.register(Arc::new(h3::H3GridPathCells));
+        registry.register_scalar(h3::H3GridDisk);
+        registry.register_scalar(h3::H3GridDiskDistances);
+        registry.register_scalar(h3::H3GridDistance);
+        registry.register_scalar(h3::H3GridPathCells);

        // h3 measurement
-        registry.register(Arc::new(h3::H3CellDistanceSphereKm));
-        registry.register(Arc::new(h3::H3CellDistanceEuclideanDegree));
+        registry.register_scalar(h3::H3CellDistanceSphereKm);
+        registry.register_scalar(h3::H3CellDistanceEuclideanDegree);

        // s2
-        registry.register(Arc::new(s2::S2LatLngToCell));
-        registry.register(Arc::new(s2::S2CellLevel));
-        registry.register(Arc::new(s2::S2CellToToken));
-        registry.register(Arc::new(s2::S2CellParent));
+        registry.register_scalar(s2::S2LatLngToCell);
+        registry.register_scalar(s2::S2CellLevel);
+        registry.register_scalar(s2::S2CellToToken);
+        registry.register_scalar(s2::S2CellParent);

        // spatial data type
-        registry.register(Arc::new(wkt::LatLngToPointWkt));
+        registry.register_scalar(wkt::LatLngToPointWkt);

        // spatial relation
-        registry.register(Arc::new(relation::STContains));
-        registry.register(Arc::new(relation::STWithin));
-        registry.register(Arc::new(relation::STIntersects));
+        registry.register_scalar(relation::STContains);
+        registry.register_scalar(relation::STWithin);
+        registry.register_scalar(relation::STIntersects);

        // spatial measure
-        registry.register(Arc::new(measure::STDistance));
-        registry.register(Arc::new(measure::STDistanceSphere));
-        registry.register(Arc::new(measure::STArea));
+        registry.register_scalar(measure::STDistance);
+        registry.register_scalar(measure::STDistanceSphere);
+        registry.register_scalar(measure::STArea);
    }
 }
--- a/src/common/function/src/scalars/geo/helpers.rs
+++ b/src/common/function/src/scalars/geo/helpers.rs
@@ -37,7 +37,7 @@ macro_rules! ensure_columns_len {
    };
 }

-pub(super) use ensure_columns_len;
+pub(crate) use ensure_columns_len;

 macro_rules! ensure_columns_n {
    ($columns:ident, $n:literal) => {
@@ -58,7 +58,7 @@ macro_rules! ensure_columns_n {
    };
 }

-pub(super) use ensure_columns_n;
+pub(crate) use ensure_columns_n;

 macro_rules! ensure_and_coerce {
    ($compare:expr, $coerce:expr) => {{
@@ -72,4 +72,4 @@ macro_rules! ensure_and_coerce {
    }};
 }

-pub(super) use ensure_and_coerce;
+pub(crate) use ensure_and_coerce;
--- a/src/common/function/src/scalars/hll_count.rs
+++ b/src/common/function/src/scalars/hll_count.rs
@@ -16,7 +16,6 @@

 use std::fmt;
 use std::fmt::Display;
-use std::sync::Arc;

 use common_query::error::{DowncastVectorSnafu, InvalidFuncArgsSnafu, Result};
 use common_query::prelude::{Signature, Volatility};
@@ -27,7 +26,7 @@ use datatypes::vectors::{BinaryVector, MutableVector, UInt64VectorBuilder, Vecto
 use hyperloglogplus::HyperLogLog;
 use snafu::OptionExt;

-use crate::aggr::HllStateType;
+use crate::aggrs::approximate::hll::HllStateType;
 use crate::function::{Function, FunctionContext};
 use crate::function_registry::FunctionRegistry;

@@ -44,7 +43,7 @@ pub struct HllCalcFunction;

 impl HllCalcFunction {
    pub fn register(registry: &FunctionRegistry) {
-        registry.register(Arc::new(HllCalcFunction));
+        registry.register_scalar(HllCalcFunction);
    }
 }

@@ -117,6 +116,8 @@ impl Function for HllCalcFunction {

 #[cfg(test)]
 mod tests {
+    use std::sync::Arc;
+
    use datatypes::vectors::BinaryVector;

    use super::*;
--- a/src/common/function/src/scalars/ip.rs
+++ b/src/common/function/src/scalars/ip.rs
@@ -17,8 +17,6 @@ mod ipv4;
 mod ipv6;
 mod range;

-use std::sync::Arc;
-
 use cidr::{Ipv4ToCidr, Ipv6ToCidr};
 use ipv4::{Ipv4NumToString, Ipv4StringToNum};
 use ipv6::{Ipv6NumToString, Ipv6StringToNum};
@@ -31,15 +29,15 @@ pub(crate) struct IpFunctions;
 impl IpFunctions {
    pub fn register(registry: &FunctionRegistry) {
        // Register IPv4 functions
-        registry.register(Arc::new(Ipv4NumToString));
-        registry.register(Arc::new(Ipv4StringToNum));
-        registry.register(Arc::new(Ipv4ToCidr));
-        registry.register(Arc::new(Ipv4InRange));
+        registry.register_scalar(Ipv4NumToString);
+        registry.register_scalar(Ipv4StringToNum);
+        registry.register_scalar(Ipv4ToCidr);
+        registry.register_scalar(Ipv4InRange);

        // Register IPv6 functions
-        registry.register(Arc::new(Ipv6NumToString));
-        registry.register(Arc::new(Ipv6StringToNum));
-        registry.register(Arc::new(Ipv6ToCidr));
-        registry.register(Arc::new(Ipv6InRange));
+        registry.register_scalar(Ipv6NumToString);
+        registry.register_scalar(Ipv6StringToNum);
+        registry.register_scalar(Ipv6ToCidr);
+        registry.register_scalar(Ipv6InRange);
    }
 }
--- a/src/common/function/src/scalars/json.rs
+++ b/src/common/function/src/scalars/json.rs
@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use std::sync::Arc;
 pub mod json_get;
 mod json_is;
 mod json_path_exists;
@@ -33,23 +32,23 @@ pub(crate) struct JsonFunction;

 impl JsonFunction {
    pub fn register(registry: &FunctionRegistry) {
-        registry.register(Arc::new(JsonToStringFunction));
-        registry.register(Arc::new(ParseJsonFunction));
+        registry.register_scalar(JsonToStringFunction);
+        registry.register_scalar(ParseJsonFunction);

-        registry.register(Arc::new(JsonGetInt));
-        registry.register(Arc::new(JsonGetFloat));
-        registry.register(Arc::new(JsonGetString));
-        registry.register(Arc::new(JsonGetBool));
+        registry.register_scalar(JsonGetInt);
+        registry.register_scalar(JsonGetFloat);
+        registry.register_scalar(JsonGetString);
+        registry.register_scalar(JsonGetBool);

-        registry.register(Arc::new(JsonIsNull));
-        registry.register(Arc::new(JsonIsInt));
-        registry.register(Arc::new(JsonIsFloat));
-        registry.register(Arc::new(JsonIsString));
-        registry.register(Arc::new(JsonIsBool));
-        registry.register(Arc::new(JsonIsArray));
-        registry.register(Arc::new(JsonIsObject));
+        registry.register_scalar(JsonIsNull);
+        registry.register_scalar(JsonIsInt);
+        registry.register_scalar(JsonIsFloat);
+        registry.register_scalar(JsonIsString);
+        registry.register_scalar(JsonIsBool);
+        registry.register_scalar(JsonIsArray);
+        registry.register_scalar(JsonIsObject);

-        registry.register(Arc::new(json_path_exists::JsonPathExistsFunction));
-        registry.register(Arc::new(json_path_match::JsonPathMatchFunction));
+        registry.register_scalar(json_path_exists::JsonPathExistsFunction);
+        registry.register_scalar(json_path_match::JsonPathMatchFunction);
    }
 }
--- a/src/common/function/src/scalars/matches.rs
+++ b/src/common/function/src/scalars/matches.rs
@@ -38,11 +38,11 @@ use crate::function_registry::FunctionRegistry;
 ///
 /// Usage: matches(`<col>`, `<pattern>`) -> boolean
 #[derive(Clone, Debug, Default)]
-pub(crate) struct MatchesFunction;
+pub struct MatchesFunction;

 impl MatchesFunction {
    pub fn register(registry: &FunctionRegistry) {
-        registry.register(Arc::new(MatchesFunction));
+        registry.register_scalar(MatchesFunction);
    }
 }

--- a/src/common/function/src/scalars/matches_term.rs
+++ b/src/common/function/src/scalars/matches_term.rs
@@ -77,7 +77,7 @@ pub struct MatchesTermFunction;

 impl MatchesTermFunction {
    pub fn register(registry: &FunctionRegistry) {
-        registry.register(Arc::new(MatchesTermFunction));
+        registry.register_scalar(MatchesTermFunction);
    }
 }

--- a/src/common/function/src/scalars/math.rs
+++ b/src/common/function/src/scalars/math.rs
@@ -18,7 +18,6 @@ mod pow;
 mod rate;

 use std::fmt;
-use std::sync::Arc;

 pub use clamp::{ClampFunction, ClampMaxFunction, ClampMinFunction};
 use common_query::error::{GeneralDataFusionSnafu, Result};
@@ -39,13 +38,13 @@ pub(crate) struct MathFunction;

 impl MathFunction {
    pub fn register(registry: &FunctionRegistry) {
-        registry.register(Arc::new(ModuloFunction));
-        registry.register(Arc::new(PowFunction));
-        registry.register(Arc::new(RateFunction));
-        registry.register(Arc::new(RangeFunction));
-        registry.register(Arc::new(ClampFunction));
-        registry.register(Arc::new(ClampMinFunction));
-        registry.register(Arc::new(ClampMaxFunction));
+        registry.register_scalar(ModuloFunction);
+        registry.register_scalar(PowFunction);
+        registry.register_scalar(RateFunction);
+        registry.register_scalar(RangeFunction);
+        registry.register_scalar(ClampFunction);
+        registry.register_scalar(ClampMinFunction);
+        registry.register_scalar(ClampMaxFunction);
    }
 }

--- a/src/common/function/src/scalars/timestamp.rs
+++ b/src/common/function/src/scalars/timestamp.rs
@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use std::sync::Arc;
 mod to_unixtime;

 use to_unixtime::ToUnixtimeFunction;
@@ -23,6 +22,6 @@ pub(crate) struct TimestampFunction;

 impl TimestampFunction {
    pub fn register(registry: &FunctionRegistry) {
-        registry.register(Arc::new(ToUnixtimeFunction));
+        registry.register_scalar(ToUnixtimeFunction);
    }
 }
--- a/src/common/function/src/scalars/uddsketch_calc.rs
+++ b/src/common/function/src/scalars/uddsketch_calc.rs
@@ -16,7 +16,6 @@

 use std::fmt;
 use std::fmt::Display;
-use std::sync::Arc;

 use common_query::error::{DowncastVectorSnafu, InvalidFuncArgsSnafu, Result};
 use common_query::prelude::{Signature, Volatility};
@@ -44,7 +43,7 @@ pub struct UddSketchCalcFunction;

 impl UddSketchCalcFunction {
    pub fn register(registry: &FunctionRegistry) {
-        registry.register(Arc::new(UddSketchCalcFunction));
+        registry.register_scalar(UddSketchCalcFunction);
    }
 }

--- a/src/common/function/src/scalars/vector.rs
+++ b/src/common/function/src/scalars/vector.rs
@@ -17,10 +17,8 @@ mod distance;
 mod elem_product;
 mod elem_sum;
 pub mod impl_conv;
-pub(crate) mod product;
 mod scalar_add;
 mod scalar_mul;
-pub(crate) mod sum;
 mod vector_add;
 mod vector_dim;
 mod vector_div;
@@ -30,37 +28,34 @@ mod vector_norm;
 mod vector_sub;
 mod vector_subvector;

-use std::sync::Arc;
-
 use crate::function_registry::FunctionRegistry;
-
 pub(crate) struct VectorFunction;

 impl VectorFunction {
    pub fn register(registry: &FunctionRegistry) {
        // conversion
-        registry.register(Arc::new(convert::ParseVectorFunction));
-        registry.register(Arc::new(convert::VectorToStringFunction));
+        registry.register_scalar(convert::ParseVectorFunction);
+        registry.register_scalar(convert::VectorToStringFunction);

        // distance
-        registry.register(Arc::new(distance::CosDistanceFunction));
-        registry.register(Arc::new(distance::DotProductFunction));
-        registry.register(Arc::new(distance::L2SqDistanceFunction));
+        registry.register_scalar(distance::CosDistanceFunction);
+        registry.register_scalar(distance::DotProductFunction);
+        registry.register_scalar(distance::L2SqDistanceFunction);

        // scalar calculation
-        registry.register(Arc::new(scalar_add::ScalarAddFunction));
-        registry.register(Arc::new(scalar_mul::ScalarMulFunction));
+        registry.register_scalar(scalar_add::ScalarAddFunction);
+        registry.register_scalar(scalar_mul::ScalarMulFunction);

        // vector calculation
-        registry.register(Arc::new(vector_add::VectorAddFunction));
-        registry.register(Arc::new(vector_sub::VectorSubFunction));
-        registry.register(Arc::new(vector_mul::VectorMulFunction));
-        registry.register(Arc::new(vector_div::VectorDivFunction));
-        registry.register(Arc::new(vector_norm::VectorNormFunction));
-        registry.register(Arc::new(vector_dim::VectorDimFunction));
-        registry.register(Arc::new(vector_kth_elem::VectorKthElemFunction));
-        registry.register(Arc::new(vector_subvector::VectorSubvectorFunction));
-        registry.register(Arc::new(elem_sum::ElemSumFunction));
-        registry.register(Arc::new(elem_product::ElemProductFunction));
+        registry.register_scalar(vector_add::VectorAddFunction);
+        registry.register_scalar(vector_sub::VectorSubFunction);
+        registry.register_scalar(vector_mul::VectorMulFunction);
+        registry.register_scalar(vector_div::VectorDivFunction);
+        registry.register_scalar(vector_norm::VectorNormFunction);
+        registry.register_scalar(vector_dim::VectorDimFunction);
+        registry.register_scalar(vector_kth_elem::VectorKthElemFunction);
+        registry.register_scalar(vector_subvector::VectorSubvectorFunction);
+        registry.register_scalar(elem_sum::ElemSumFunction);
+        registry.register_scalar(elem_product::ElemProductFunction);
    }
 }
--- a/src/common/function/src/system.rs
+++ b/src/common/function/src/system.rs
@@ -36,13 +36,13 @@ pub(crate) struct SystemFunction;

 impl SystemFunction {
    pub fn register(registry: &FunctionRegistry) {
-        registry.register(Arc::new(BuildFunction));
-        registry.register(Arc::new(VersionFunction));
-        registry.register(Arc::new(CurrentSchemaFunction));
-        registry.register(Arc::new(DatabaseFunction));
-        registry.register(Arc::new(SessionUserFunction));
-        registry.register(Arc::new(ReadPreferenceFunction));
-        registry.register(Arc::new(TimezoneFunction));
+        registry.register_scalar(BuildFunction);
+        registry.register_scalar(VersionFunction);
+        registry.register_scalar(CurrentSchemaFunction);
+        registry.register_scalar(DatabaseFunction);
+        registry.register_scalar(SessionUserFunction);
+        registry.register_scalar(ReadPreferenceFunction);
+        registry.register_scalar(TimezoneFunction);
        registry.register_async(Arc::new(ProcedureStateFunction));
        PGCatalogFunction::register(registry);
    }
--- a/src/common/function/src/system/pg_catalog.rs
+++ b/src/common/function/src/system/pg_catalog.rs
@@ -16,8 +16,6 @@ mod pg_get_userbyid;
 mod table_is_visible;
 mod version;

-use std::sync::Arc;
-
 use pg_get_userbyid::PGGetUserByIdFunction;
 use table_is_visible::PGTableIsVisibleFunction;
 use version::PGVersionFunction;
@@ -35,8 +33,8 @@ pub(super) struct PGCatalogFunction;

 impl PGCatalogFunction {
    pub fn register(registry: &FunctionRegistry) {
-        registry.register(Arc::new(PGTableIsVisibleFunction));
-        registry.register(Arc::new(PGGetUserByIdFunction));
-        registry.register(Arc::new(PGVersionFunction));
+        registry.register_scalar(PGTableIsVisibleFunction);
+        registry.register_scalar(PGGetUserByIdFunction);
+        registry.register_scalar(PGVersionFunction);
    }
 }
--- a/src/common/grpc/src/channel_manager.rs
+++ b/src/common/grpc/src/channel_manager.rs
@@ -296,6 +296,8 @@ pub struct ChannelConfig {
    pub max_recv_message_size: ReadableSize,
    // Max gRPC sending(encoding) message size
    pub max_send_message_size: ReadableSize,
+    pub send_compression: bool,
+    pub accept_compression: bool,
 }

 impl Default for ChannelConfig {
@@ -316,6 +318,8 @@ impl Default for ChannelConfig {
            client_tls: None,
            max_recv_message_size: DEFAULT_MAX_GRPC_RECV_MESSAGE_SIZE,
            max_send_message_size: DEFAULT_MAX_GRPC_SEND_MESSAGE_SIZE,
+            send_compression: false,
+            accept_compression: false,
        }
    }
 }
@@ -566,6 +570,8 @@ mod tests {
                client_tls: None,
                max_recv_message_size: DEFAULT_MAX_GRPC_RECV_MESSAGE_SIZE,
                max_send_message_size: DEFAULT_MAX_GRPC_SEND_MESSAGE_SIZE,
+                send_compression: false,
+                accept_compression: false,
            },
            default_cfg
        );
@@ -610,6 +616,8 @@ mod tests {
                }),
                max_recv_message_size: DEFAULT_MAX_GRPC_RECV_MESSAGE_SIZE,
                max_send_message_size: DEFAULT_MAX_GRPC_SEND_MESSAGE_SIZE,
+                send_compression: false,
+                accept_compression: false,
            },
            cfg
        );
--- a/src/common/grpc/src/flight.rs
+++ b/src/common/grpc/src/flight.rs
@@ -64,6 +64,7 @@ impl Default for FlightEncoder {
 }

 impl FlightEncoder {
+    /// Creates new [FlightEncoder] with compression disabled.
    pub fn with_compression_disabled() -> Self {
        let write_options = writer::IpcWriteOptions::default()
            .try_with_compression(None)
--- a/src/common/meta/src/distributed_time_constants.rs
+++ b/src/common/meta/src/distributed_time_constants.rs
@@ -35,6 +35,9 @@ pub const FLOWNODE_LEASE_SECS: u64 = DATANODE_LEASE_SECS;
 /// The lease seconds of metasrv leader.
 pub const META_LEASE_SECS: u64 = 5;

+/// The keep-alive interval of the Postgres connection.
+pub const POSTGRES_KEEP_ALIVE_SECS: u64 = 30;
+
 /// In a lease, there are two opportunities for renewal.
 pub const META_KEEP_ALIVE_INTERVAL_SECS: u64 = META_LEASE_SECS / 2;

--- a/src/common/meta/src/error.rs
+++ b/src/common/meta/src/error.rs
@@ -1001,7 +1001,7 @@ impl ErrorExt for Error {
            }
            #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
            RdsTransactionRetryFailed { .. } => StatusCode::Internal,
-            Error::DatanodeTableInfoNotFound { .. } => StatusCode::Internal,
+            DatanodeTableInfoNotFound { .. } => StatusCode::Internal,
        }
    }

--- a/src/common/meta/src/key/view_info.rs
+++ b/src/common/meta/src/key/view_info.rs
@@ -70,11 +70,12 @@ impl MetadataKey<'_, ViewInfoKey> for ViewInfoKey {
            }
            .build()
        })?;
-        let captures = VIEW_INFO_KEY_PATTERN
-            .captures(key)
-            .context(InvalidViewInfoSnafu {
-                err_msg: format!("Invalid ViewInfoKey '{key}'"),
-            })?;
+        let captures =
+            VIEW_INFO_KEY_PATTERN
+                .captures(key)
+                .with_context(|| InvalidViewInfoSnafu {
+                    err_msg: format!("Invalid ViewInfoKey '{key}'"),
+                })?;
        // Safety: pass the regex check above
        let view_id = captures[1].parse::<TableId>().unwrap();
        Ok(ViewInfoKey { view_id })
--- a/src/common/meta/src/snapshot.rs
+++ b/src/common/meta/src/snapshot.rs
@@ -14,6 +14,7 @@

 pub mod file;

+use std::borrow::Cow;
 use std::fmt::{Display, Formatter};
 use std::path::{Path, PathBuf};
 use std::time::Instant;
@@ -271,6 +272,49 @@ impl MetadataSnapshotManager {

        Ok((filename.to_string(), num_keyvalues as u64))
    }
+
+    fn format_output(key: Cow<'_, str>, value: Cow<'_, str>) -> String {
+        format!("{} => {}", key, value)
+    }
+
+    pub async fn info(
+        object_store: &ObjectStore,
+        file_path: &str,
+        query_str: &str,
+        limit: Option<usize>,
+    ) -> Result<Vec<String>> {
+        let path = Path::new(file_path);
+
+        let file_name = path
+            .file_name()
+            .and_then(|s| s.to_str())
+            .context(InvalidFilePathSnafu { file_path })?;
+
+        let filename = FileName::try_from(file_name)?;
+        let data = object_store
+            .read(file_path)
+            .await
+            .context(ReadObjectSnafu { file_path })?;
+        let document = Document::from_slice(&filename.extension.format, &data.to_bytes())?;
+        let metadata_content = document.into_metadata_content()?.values();
+        let mut results = Vec::with_capacity(limit.unwrap_or(256));
+        for kv in metadata_content {
+            let key_str = String::from_utf8_lossy(&kv.key);
+            if let Some(prefix) = query_str.strip_suffix('*') {
+                if key_str.starts_with(prefix) {
+                    let value_str = String::from_utf8_lossy(&kv.value);
+                    results.push(Self::format_output(key_str, value_str));
+                }
+            } else if key_str == query_str {
+                let value_str = String::from_utf8_lossy(&kv.value);
+                results.push(Self::format_output(key_str, value_str));
+            }
+            if results.len() == limit.unwrap_or(usize::MAX) {
+                break;
+            }
+        }
+        Ok(results)
+    }
 }

 #[cfg(test)]
--- a/src/common/meta/src/snapshot/file.rs
+++ b/src/common/meta/src/snapshot/file.rs
@@ -111,6 +111,11 @@ impl MetadataContent {
    pub fn into_iter(self) -> impl Iterator<Item = KeyValue> {
        self.values.into_iter()
    }
+
+    /// Returns the key-value pairs as a vector.
+    pub fn values(self) -> Vec<KeyValue> {
+        self.values
+    }
 }

 /// The key-value pair of the backup file.
--- a/src/datanode/src/datanode.rs
+++ b/src/datanode/src/datanode.rs
@@ -372,6 +372,7 @@ impl DatanodeBuilder {
            opts.max_concurrent_queries,
            //TODO: revaluate the hardcoded timeout on the next version of datanode concurrency limiter.
            Duration::from_millis(100),
+            opts.grpc.flight_compression,
        );

        let object_store_manager = Self::build_object_store_manager(&opts.storage).await?;
--- a/src/datanode/src/region_server.rs
+++ b/src/datanode/src/region_server.rs
@@ -50,6 +50,7 @@ use query::QueryEngineRef;
 use servers::error::{self as servers_error, ExecuteGrpcRequestSnafu, Result as ServerResult};
 use servers::grpc::flight::{FlightCraft, FlightRecordBatchStream, TonicStream};
 use servers::grpc::region_server::RegionServerHandler;
+use servers::grpc::FlightCompression;
 use session::context::{QueryContextBuilder, QueryContextRef};
 use snafu::{ensure, OptionExt, ResultExt};
 use store_api::metric_engine_consts::{
@@ -80,6 +81,7 @@ use crate::event_listener::RegionServerEventListenerRef;
 #[derive(Clone)]
 pub struct RegionServer {
    inner: Arc<RegionServerInner>,
+    flight_compression: FlightCompression,
 }

 pub struct RegionStat {
@@ -93,6 +95,7 @@ impl RegionServer {
        query_engine: QueryEngineRef,
        runtime: Runtime,
        event_listener: RegionServerEventListenerRef,
+        flight_compression: FlightCompression,
    ) -> Self {
        Self::with_table_provider(
            query_engine,
@@ -101,6 +104,7 @@ impl RegionServer {
            Arc::new(DummyTableProviderFactory),
            0,
            Duration::from_millis(0),
+            flight_compression,
        )
    }

@@ -111,6 +115,7 @@ impl RegionServer {
        table_provider_factory: TableProviderFactoryRef,
        max_concurrent_queries: usize,
        concurrent_query_limiter_timeout: Duration,
+        flight_compression: FlightCompression,
    ) -> Self {
        Self {
            inner: Arc::new(RegionServerInner::new(
@@ -123,6 +128,7 @@ impl RegionServer {
                    concurrent_query_limiter_timeout,
                ),
            )),
+            flight_compression,
        }
    }

@@ -536,7 +542,11 @@ impl FlightCraft for RegionServer {
            .trace(tracing_context.attach(info_span!("RegionServer::handle_read")))
            .await?;

-        let stream = Box::pin(FlightRecordBatchStream::new(result, tracing_context));
+        let stream = Box::pin(FlightRecordBatchStream::new(
+            result,
+            tracing_context,
+            self.flight_compression,
+        ));
        Ok(Response::new(stream))
    }
 }
--- a/src/datanode/src/tests.rs
+++ b/src/datanode/src/tests.rs
@@ -19,16 +19,16 @@ use std::time::Duration;
 use api::region::RegionResponse;
 use async_trait::async_trait;
 use common_error::ext::BoxedError;
-use common_function::function::FunctionRef;
-use common_function::scalars::aggregate::AggregateFunctionMetaRef;
+use common_function::function_factory::ScalarFunctionFactory;
 use common_query::Output;
 use common_runtime::runtime::{BuilderBuild, RuntimeTrait};
 use common_runtime::Runtime;
-use datafusion_expr::LogicalPlan;
+use datafusion_expr::{AggregateUDF, LogicalPlan};
 use query::dataframe::DataFrame;
 use query::planner::LogicalPlanner;
 use query::query_engine::{DescribeResult, QueryEngineState};
 use query::{QueryEngine, QueryEngineContext};
+use servers::grpc::FlightCompression;
 use session::context::QueryContextRef;
 use store_api::metadata::RegionMetadataRef;
 use store_api::region_engine::{
@@ -76,9 +76,9 @@ impl QueryEngine for MockQueryEngine {
        unimplemented!()
    }

-    fn register_aggregate_function(&self, _func: AggregateFunctionMetaRef) {}
+    fn register_aggregate_function(&self, _func: AggregateUDF) {}

-    fn register_function(&self, _func: FunctionRef) {}
+    fn register_scalar_function(&self, _func: ScalarFunctionFactory) {}

    fn read_table(&self, _table: TableRef) -> query::error::Result<DataFrame> {
        unimplemented!()
@@ -98,6 +98,7 @@ pub fn mock_region_server() -> RegionServer {
        Arc::new(MockQueryEngine),
        Runtime::builder().build().unwrap(),
        Box::new(NoopRegionServerEventListener),
+        FlightCompression::default(),
    )
 }

--- a/src/datatypes/src/value.rs
+++ b/src/datatypes/src/value.rs
@@ -326,6 +326,14 @@ impl Value {
        }
    }

+    /// Cast Value to [Duration]. Return None if value is not a valid duration data type.
+    pub fn as_duration(&self) -> Option<Duration> {
+        match self {
+            Value::Duration(d) => Some(*d),
+            _ => None,
+        }
+    }
+
    /// Returns the logical type of the value.
    pub fn logical_type_id(&self) -> LogicalTypeId {
        match self {
--- a/src/flow/src/transform.rs
+++ b/src/flow/src/transform.rs
@@ -17,7 +17,7 @@ use std::collections::BTreeMap;
 use std::sync::Arc;

 use common_error::ext::BoxedError;
-use common_function::function::FunctionContext;
+use common_function::function::{FunctionContext, FunctionRef};
 use datafusion_substrait::extensions::Extensions;
 use datatypes::data_type::ConcreteDataType as CDT;
 use query::QueryEngine;
@@ -108,9 +108,13 @@ impl FunctionExtensions {

 /// register flow-specific functions to the query engine
 pub fn register_function_to_query_engine(engine: &Arc<dyn QueryEngine>) {
-    engine.register_function(Arc::new(TumbleFunction::new("tumble")));
-    engine.register_function(Arc::new(TumbleFunction::new(TUMBLE_START)));
-    engine.register_function(Arc::new(TumbleFunction::new(TUMBLE_END)));
+    let tumble_fn = Arc::new(TumbleFunction::new("tumble")) as FunctionRef;
+    let tumble_start_fn = Arc::new(TumbleFunction::new(TUMBLE_START)) as FunctionRef;
+    let tumble_end_fn = Arc::new(TumbleFunction::new(TUMBLE_END)) as FunctionRef;
+
+    engine.register_scalar_function(tumble_fn.into());
+    engine.register_scalar_function(tumble_start_fn.into());
+    engine.register_scalar_function(tumble_end_fn.into());
 }

 #[derive(Debug)]
--- a/src/frontend/Cargo.toml
+++ b/src/frontend/Cargo.toml
@@ -25,6 +25,7 @@ common-catalog.workspace = true
 common-config.workspace = true
 common-datasource.workspace = true
 common-error.workspace = true
+common-frontend.workspace = true
 common-function.workspace = true
 common-grpc.workspace = true
 common-macro.workspace = true
--- a/src/frontend/src/instance.rs
+++ b/src/frontend/src/instance.rs
@@ -30,6 +30,7 @@ use std::time::SystemTime;

 use async_trait::async_trait;
 use auth::{PermissionChecker, PermissionCheckerRef, PermissionReq};
+use catalog::process_manager::ProcessManagerRef;
 use catalog::CatalogManagerRef;
 use client::OutputData;
 use common_base::Plugins;
@@ -80,6 +81,7 @@ use crate::error::{
 };
 use crate::limiter::LimiterRef;
 use crate::slow_query_recorder::SlowQueryRecorder;
+use crate::stream_wrapper::StreamWrapper;

 /// The frontend instance contains necessary components, and implements many
 /// traits, like [`servers::query_handler::grpc::GrpcQueryHandler`],
@@ -96,6 +98,7 @@ pub struct Instance {
    table_metadata_manager: TableMetadataManagerRef,
    slow_query_recorder: Option<SlowQueryRecorder>,
    limiter: Option<LimiterRef>,
+    process_manager: ProcessManagerRef,
 }

 impl Instance {
@@ -153,6 +156,10 @@ impl Instance {
    pub fn inserter(&self) -> &InserterRef {
        &self.inserter
    }
+
+    pub fn process_manager(&self) -> &ProcessManagerRef {
+        &self.process_manager
+    }
 }

 fn parse_stmt(sql: &str, dialect: &(dyn Dialect + Send + Sync)) -> Result<Vec<Statement>> {
@@ -172,6 +179,14 @@ impl Instance {
            None
        };

+        let ticket = self.process_manager.register_query(
+            query_ctx.current_catalog().to_string(),
+            vec![query_ctx.current_schema()],
+            stmt.to_string(),
+            "unknown".to_string(),
+            None,
+        );
+
        let output = match stmt {
            Statement::Query(_) | Statement::Explain(_) | Statement::Delete(_) => {
                // TODO: remove this when format is supported in datafusion
@@ -215,7 +230,20 @@ impl Instance {
            }
        };

-        output.context(TableOperationSnafu)
+        match output {
+            Ok(output) => {
+                let Output { meta, data } = output;
+
+                let data = match data {
+                    OutputData::Stream(stream) => {
+                        OutputData::Stream(Box::pin(StreamWrapper::new(stream, ticket)))
+                    }
+                    other => other,
+                };
+                Ok(Output { data, meta })
+            }
+            Err(e) => Err(e).context(TableOperationSnafu),
+        }
    }
 }

--- a/src/frontend/src/instance/builder.rs
+++ b/src/frontend/src/instance/builder.rs
@@ -15,6 +15,7 @@
 use std::sync::Arc;

 use cache::{TABLE_FLOWNODE_SET_CACHE_NAME, TABLE_ROUTE_CACHE_NAME};
+use catalog::process_manager::ProcessManagerRef;
 use catalog::CatalogManagerRef;
 use common_base::Plugins;
 use common_meta::cache::{LayeredCacheRegistryRef, TableRouteCacheRef};
@@ -54,9 +55,11 @@ pub struct FrontendBuilder {
    node_manager: NodeManagerRef,
    plugins: Option<Plugins>,
    procedure_executor: ProcedureExecutorRef,
+    process_manager: ProcessManagerRef,
 }

 impl FrontendBuilder {
+    #[allow(clippy::too_many_arguments)]
    pub fn new(
        options: FrontendOptions,
        kv_backend: KvBackendRef,
@@ -64,6 +67,7 @@ impl FrontendBuilder {
        catalog_manager: CatalogManagerRef,
        node_manager: NodeManagerRef,
        procedure_executor: ProcedureExecutorRef,
+        process_manager: ProcessManagerRef,
    ) -> Self {
        Self {
            options,
@@ -74,6 +78,7 @@ impl FrontendBuilder {
            node_manager,
            plugins: None,
            procedure_executor,
+            process_manager,
        }
    }

@@ -95,7 +100,7 @@ impl FrontendBuilder {
        let kv_backend = self.kv_backend;
        let node_manager = self.node_manager;
        let plugins = self.plugins.unwrap_or_default();
-
+        let process_manager = self.process_manager;
        let table_route_cache: TableRouteCacheRef =
            self.layered_cache_registry
                .get()
@@ -216,6 +221,7 @@ impl FrontendBuilder {
            table_metadata_manager: Arc::new(TableMetadataManager::new(kv_backend)),
            slow_query_recorder,
            limiter,
+            process_manager,
        })
    }
 }
--- a/src/frontend/src/instance/jaeger.rs
+++ b/src/frontend/src/instance/jaeger.rs
@@ -130,7 +130,13 @@ impl JaegerQueryHandler for Instance {
        .await?)
    }

-    async fn get_trace(&self, ctx: QueryContextRef, trace_id: &str) -> ServerResult<Output> {
+    async fn get_trace(
+        &self,
+        ctx: QueryContextRef,
+        trace_id: &str,
+        start_time: Option<i64>,
+        end_time: Option<i64>,
+    ) -> ServerResult<Output> {
        // It's equivalent to
        //
        // ```
@@ -139,13 +145,25 @@ impl JaegerQueryHandler for Instance {
        // FROM
        //   {db}.{trace_table}
        // WHERE
-        //   trace_id = '{trace_id}'
+        //   trace_id = '{trace_id}' AND
+        //   timestamp >= {start_time} AND
+        //   timestamp <= {end_time}
        // ORDER BY
        //   timestamp DESC
        // ```.
        let selects = vec![wildcard()];

-        let filters = vec![col(TRACE_ID_COLUMN).eq(lit(trace_id))];
+        let mut filters = vec![col(TRACE_ID_COLUMN).eq(lit(trace_id))];
+
+        if let Some(start_time) = start_time {
+            // Microseconds to nanoseconds.
+            filters.push(col(TIMESTAMP_COLUMN).gt_eq(lit_timestamp_nano(start_time * 1_000)));
+        }
+
+        if let Some(end_time) = end_time {
+            // Microseconds to nanoseconds.
+            filters.push(col(TIMESTAMP_COLUMN).lt_eq(lit_timestamp_nano(end_time * 1_000)));
+        }

        Ok(query_trace_table(
            ctx,
--- a/src/frontend/src/lib.rs
+++ b/src/frontend/src/lib.rs
@@ -23,3 +23,4 @@ pub(crate) mod metrics;
 pub mod server;
 pub mod service_config;
 pub(crate) mod slow_query_recorder;
+mod stream_wrapper;
--- a/src/frontend/src/server.rs
+++ b/src/frontend/src/server.rs
@@ -20,6 +20,7 @@ use common_base::Plugins;
 use common_config::Configurable;
 use servers::error::Error as ServerError;
 use servers::grpc::builder::GrpcServerBuilder;
+use servers::grpc::frontend_grpc_handler::FrontendGrpcHandler;
 use servers::grpc::greptime_handler::GreptimeRequestHandler;
 use servers::grpc::{GrpcOptions, GrpcServer, GrpcServerConfig};
 use servers::http::event::LogValidatorRef;
@@ -154,13 +155,17 @@ where
            ServerGrpcQueryHandlerAdapter::arc(self.instance.clone()),
            user_provider.clone(),
            runtime,
+            opts.grpc.flight_compression,
        );

+        let frontend_grpc_handler =
+            FrontendGrpcHandler::new(self.instance.process_manager().clone());
        let grpc_server = builder
            .database_handler(greptime_request_handler.clone())
            .prometheus_handler(self.instance.clone(), user_provider.clone())
            .otel_arrow_handler(OtelArrowServiceHandler(self.instance.clone()))
            .flight_handler(Arc::new(greptime_request_handler))
+            .frontend_grpc_handler(frontend_grpc_handler)
            .build();
        Ok(grpc_server)
    }
--- a/src/frontend/src/stream_wrapper.rs
+++ b/src/frontend/src/stream_wrapper.rs
@@ -0,0 +1,60 @@
+// 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::pin::Pin;
+use std::task::{Context, Poll};
+
+use common_recordbatch::adapter::RecordBatchMetrics;
+use common_recordbatch::{OrderOption, RecordBatch, RecordBatchStream, SendableRecordBatchStream};
+use datatypes::schema::SchemaRef;
+use futures::Stream;
+
+pub struct StreamWrapper<T> {
+    inner: SendableRecordBatchStream,
+    _attachment: T,
+}
+
+impl<T> Unpin for StreamWrapper<T> {}
+
+impl<T> StreamWrapper<T> {
+    pub fn new(stream: SendableRecordBatchStream, attachment: T) -> Self {
+        Self {
+            inner: stream,
+            _attachment: attachment,
+        }
+    }
+}
+
+impl<T> Stream for StreamWrapper<T> {
+    type Item = common_recordbatch::error::Result<RecordBatch>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let this = &mut *self;
+        Pin::new(&mut this.inner).poll_next(cx)
+    }
+}
+
+impl<T> RecordBatchStream for StreamWrapper<T> {
+    fn schema(&self) -> SchemaRef {
+        self.inner.schema()
+    }
+
+    fn output_ordering(&self) -> Option<&[OrderOption]> {
+        self.inner.output_ordering()
+    }
+
+    fn metrics(&self) -> Option<RecordBatchMetrics> {
+        self.inner.metrics()
+    }
+}
--- a/src/meta-srv/src/bootstrap.rs
+++ b/src/meta-srv/src/bootstrap.rs
@@ -31,8 +31,6 @@ use common_meta::kv_backend::rds::MySqlStore;
 #[cfg(feature = "pg_kvbackend")]
 use common_meta::kv_backend::rds::PgStore;
 use common_meta::kv_backend::{KvBackendRef, ResettableKvBackendRef};
-#[cfg(feature = "pg_kvbackend")]
-use common_telemetry::error;
 use common_telemetry::info;
 #[cfg(feature = "pg_kvbackend")]
 use deadpool_postgres::{Config, Runtime};
@@ -144,7 +142,8 @@ impl MetasrvInstance {
        let (serve_state_tx, serve_state_rx) = oneshot::channel();

        let socket_addr =
-            bootstrap_metasrv_with_router(&self.opts.bind_addr, router, serve_state_tx, rx).await?;
+            bootstrap_metasrv_with_router(&self.opts.grpc.bind_addr, router, serve_state_tx, rx)
+                .await?;
        self.bind_addr = Some(socket_addr);

        let addr = self.opts.http.addr.parse().context(error::ParseAddrSnafu {
@@ -260,7 +259,7 @@ pub async fn metasrv_builder(
            let etcd_client = create_etcd_client(&opts.store_addrs).await?;
            let kv_backend = EtcdStore::with_etcd_client(etcd_client.clone(), opts.max_txn_ops);
            let election = EtcdElection::with_etcd_client(
-                &opts.server_addr,
+                &opts.grpc.server_addr,
                etcd_client,
                opts.store_key_prefix.clone(),
            )
@@ -270,22 +269,41 @@ pub async fn metasrv_builder(
        }
        #[cfg(feature = "pg_kvbackend")]
        (None, BackendImpl::PostgresStore) => {
-            let pool = create_postgres_pool(&opts.store_addrs).await?;
-            let kv_backend = PgStore::with_pg_pool(pool, &opts.meta_table_name, opts.max_txn_ops)
-                .await
-                .context(error::KvBackendSnafu)?;
-            // Client for election should be created separately since we need a different session keep-alive idle time.
-            let election_client = create_postgres_client(opts).await?;
+            use std::time::Duration;
+
+            use common_meta::distributed_time_constants::POSTGRES_KEEP_ALIVE_SECS;
+
+            use crate::election::rds::postgres::ElectionPgClient;
+
+            let candidate_lease_ttl = Duration::from_secs(CANDIDATE_LEASE_SECS);
+            let execution_timeout = Duration::from_secs(META_LEASE_SECS);
+            let statement_timeout = Duration::from_secs(META_LEASE_SECS);
+            let meta_lease_ttl = Duration::from_secs(META_LEASE_SECS);
+
+            let mut cfg = Config::new();
+            cfg.keepalives = Some(true);
+            cfg.keepalives_idle = Some(Duration::from_secs(POSTGRES_KEEP_ALIVE_SECS));
+            // We use a separate pool for election since we need a different session keep-alive idle time.
+            let pool = create_postgres_pool_with(&opts.store_addrs, cfg).await?;
+
+            let election_client =
+                ElectionPgClient::new(pool, execution_timeout, meta_lease_ttl, statement_timeout)?;
            let election = PgElection::with_pg_client(
-                opts.server_addr.clone(),
+                opts.grpc.server_addr.clone(),
                election_client,
                opts.store_key_prefix.clone(),
-                CANDIDATE_LEASE_SECS,
-                META_LEASE_SECS,
+                candidate_lease_ttl,
+                meta_lease_ttl,
                &opts.meta_table_name,
                opts.meta_election_lock_id,
            )
            .await?;
+
+            let pool = create_postgres_pool(&opts.store_addrs).await?;
+            let kv_backend = PgStore::with_pg_pool(pool, &opts.meta_table_name, opts.max_txn_ops)
+                .await
+                .context(error::KvBackendSnafu)?;
+
            (kv_backend, Some(election))
        }
        #[cfg(feature = "mysql_kvbackend")]
@@ -299,7 +317,7 @@ pub async fn metasrv_builder(
            let election_table_name = opts.meta_table_name.clone() + "_election";
            let election_client = create_mysql_client(opts).await?;
            let election = MySqlElection::with_mysql_client(
-                opts.server_addr.clone(),
+                opts.grpc.server_addr.clone(),
                election_client,
                opts.store_key_prefix.clone(),
                CANDIDATE_LEASE_SECS,
@@ -372,31 +390,24 @@ pub async fn create_etcd_client(store_addrs: &[String]) -> Result<Client> {
 }

 #[cfg(feature = "pg_kvbackend")]
-async fn create_postgres_client(opts: &MetasrvOptions) -> Result<tokio_postgres::Client> {
-    let postgres_url = opts
-        .store_addrs
-        .first()
-        .context(error::InvalidArgumentsSnafu {
-            err_msg: "empty store addrs",
-        })?;
-    let (client, connection) = tokio_postgres::connect(postgres_url, NoTls)
-        .await
-        .context(error::ConnectPostgresSnafu)?;
-
-    tokio::spawn(async move {
-        if let Err(e) = connection.await {
-            error!(e; "connection error");
-        }
-    });
-    Ok(client)
+/// Creates a pool for the Postgres backend.
+///
+/// It only use first store addr to create a pool.
+pub async fn create_postgres_pool(store_addrs: &[String]) -> Result<deadpool_postgres::Pool> {
+    create_postgres_pool_with(store_addrs, Config::new()).await
 }

 #[cfg(feature = "pg_kvbackend")]
-pub async fn create_postgres_pool(store_addrs: &[String]) -> Result<deadpool_postgres::Pool> {
+/// Creates a pool for the Postgres backend.
+///
+/// It only use first store addr to create a pool, and use the given config to create a pool.
+pub async fn create_postgres_pool_with(
+    store_addrs: &[String],
+    mut cfg: Config,
+) -> Result<deadpool_postgres::Pool> {
    let postgres_url = store_addrs.first().context(error::InvalidArgumentsSnafu {
        err_msg: "empty store addrs",
    })?;
-    let mut cfg = Config::new();
    cfg.url = Some(postgres_url.to_string());
    let pool = cfg
        .create_pool(Some(Runtime::Tokio1), NoTls)
--- a/Show More
+++ b/Show More