chore/enable-flight-encoder:

### Add Flight Compression Support - **Configuration Updates**: - Added `grpc.flight_compression` option to `config/config.md`, `config/datanode.example.toml`, and `config/frontend.example.toml` to specify compression modes for Arrow IPC service. - **Code Enhancements**: - Updated `FlightEncoder` in `src/common/grpc/src/flight.rs` to support compression modes. - Modified `RegionServer` and `DatanodeBuilder` in `src/datanode/src/datanode.rs` and `src/datanode/src/region_server.rs` to handle `FlightCompression`. - Integrated `FlightCompression` in `src/servers/src/grpc.rs` and `src/servers/src/grpc/flight.rs` to manage compression settings. - **Testing and Integration**: - Updated test utilities and integration tests in `tests-integration/src/grpc/flight.rs` and `tests-integration/src/test_util.rs` to include `FlightCompression`. Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
feat: truly limit time range by split window
2025-12-26 08:00:01 +00:00 · 2025-06-10 23:15:01 +08:00 · 2025-06-10 23:15:01 +08:00 · 2025-06-10 23:15:01 +08:00 · 2025-06-10 23:15:01 +08:00 · 2025-06-10 23:15:01 +08:00
234 changed files with 2055 additions and 5264 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.backendStorage.etcd.endpoints=${{ inputs.etcd-endpoints }} \
+        --set meta.etcdEndpoints=${{ inputs.etcd-endpoints }} \
        --set meta.enableRegionFailover=${{ inputs.enable-region-failover }} \
        --set image.registry=${{ inputs.image-registry }} \
        --set image.repository=${{ inputs.image-repository }}  \
@@ -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.backendStorage.etcd.endpoints="etcd.$install_namespace:2379" \
+    --set meta.etcdEndpoints="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.backendStorage.etcd.endpoints="etcd.$install_namespace:2379" \
+    --set meta.etcdEndpoints="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 -s -m "chore: Update GreptimeDB version to ${VERSION}"
+  git commit -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 -s -m "chore: Update GreptimeDB version to ${VERSION}"
+  git commit -m "chore: Update GreptimeDB version to ${VERSION}"
  git push origin $BRANCH_NAME

  # Create a Pull Request.
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1621,10 +1621,8 @@ dependencies = [
 "cache",
 "catalog",
 "chrono",
- "common-base",
 "common-catalog",
 "common-error",
- "common-frontend",
 "common-macro",
 "common-meta",
 "common-procedure",
@@ -2295,14 +2293,8 @@ 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]]
@@ -3260,7 +3252,7 @@ dependencies = [
 [[package]]
 name = "datafusion"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "arrow 54.2.1",
 "arrow-array 54.2.1",
@@ -3311,7 +3303,7 @@ dependencies = [
 [[package]]
 name = "datafusion-catalog"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "arrow 54.2.1",
 "async-trait",
@@ -3331,7 +3323,7 @@ dependencies = [
 [[package]]
 name = "datafusion-catalog-listing"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "arrow 54.2.1",
 "arrow-schema 54.3.1",
@@ -3354,7 +3346,7 @@ dependencies = [
 [[package]]
 name = "datafusion-common"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "ahash 0.8.11",
 "arrow 54.2.1",
@@ -3379,7 +3371,7 @@ dependencies = [
 [[package]]
 name = "datafusion-common-runtime"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "log",
 "tokio",
@@ -3388,12 +3380,12 @@ dependencies = [
 [[package]]
 name = "datafusion-doc"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"

 [[package]]
 name = "datafusion-execution"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "arrow 54.2.1",
 "dashmap",
@@ -3411,7 +3403,7 @@ dependencies = [
 [[package]]
 name = "datafusion-expr"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "arrow 54.2.1",
 "chrono",
@@ -3431,7 +3423,7 @@ dependencies = [
 [[package]]
 name = "datafusion-expr-common"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "arrow 54.2.1",
 "datafusion-common",
@@ -3442,7 +3434,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "arrow 54.2.1",
 "arrow-buffer 54.3.1",
@@ -3471,7 +3463,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-aggregate"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "ahash 0.8.11",
 "arrow 54.2.1",
@@ -3492,7 +3484,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-aggregate-common"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "ahash 0.8.11",
 "arrow 54.2.1",
@@ -3504,7 +3496,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-nested"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "arrow 54.2.1",
 "arrow-array 54.2.1",
@@ -3526,7 +3518,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-table"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "arrow 54.2.1",
 "async-trait",
@@ -3541,7 +3533,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-window"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "datafusion-common",
 "datafusion-doc",
@@ -3557,7 +3549,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-window-common"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "datafusion-common",
 "datafusion-physical-expr-common",
@@ -3566,7 +3558,7 @@ dependencies = [
 [[package]]
 name = "datafusion-macros"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "datafusion-expr",
 "quote",
@@ -3576,7 +3568,7 @@ dependencies = [
 [[package]]
 name = "datafusion-optimizer"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "arrow 54.2.1",
 "chrono",
@@ -3594,7 +3586,7 @@ dependencies = [
 [[package]]
 name = "datafusion-physical-expr"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "ahash 0.8.11",
 "arrow 54.2.1",
@@ -3617,7 +3609,7 @@ dependencies = [
 [[package]]
 name = "datafusion-physical-expr-common"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "ahash 0.8.11",
 "arrow 54.2.1",
@@ -3630,7 +3622,7 @@ dependencies = [
 [[package]]
 name = "datafusion-physical-optimizer"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "arrow 54.2.1",
 "arrow-schema 54.3.1",
@@ -3651,7 +3643,7 @@ dependencies = [
 [[package]]
 name = "datafusion-physical-plan"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "ahash 0.8.11",
 "arrow 54.2.1",
@@ -3681,7 +3673,7 @@ dependencies = [
 [[package]]
 name = "datafusion-sql"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "arrow 54.2.1",
 "arrow-array 54.2.1",
@@ -3699,7 +3691,7 @@ dependencies = [
 [[package]]
 name = "datafusion-substrait"
 version = "45.0.0"
-source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=12c0381babd52c681043957e9d6ee083a03f7646#12c0381babd52c681043957e9d6ee083a03f7646"
+source = "git+https://github.com/waynexia/arrow-datafusion.git?rev=e104c7cf62b11dd5fe41461b82514978234326b4#e104c7cf62b11dd5fe41461b82514978234326b4"
 dependencies = [
 "async-recursion",
 "async-trait",
@@ -4709,7 +4701,6 @@ dependencies = [
 "common-config",
 "common-datasource",
 "common-error",
- "common-frontend",
 "common-function",
 "common-grpc",
 "common-macro",
@@ -4756,7 +4747,6 @@ dependencies = [
 "substrait 0.15.0",
 "table",
 "tokio",
- "tokio-util",
 "toml 0.8.19",
 "tonic 0.12.3",
 "tower 0.5.2",
@@ -5143,7 +5133,7 @@ dependencies = [
 [[package]]
 name = "greptime-proto"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=17971523673f4fbc982510d3c9d6647ff642e16f#17971523673f4fbc982510d3c9d6647ff642e16f"
+source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=2dca1dc67862d7b410838aef81232274c019b3f6#2dca1dc67862d7b410838aef81232274c019b3f6"
 dependencies = [
 "prost 0.13.5",
 "serde",
@@ -7250,7 +7240,6 @@ dependencies = [
 "humantime-serde",
 "itertools 0.14.0",
 "lazy_static",
- "mito-codec",
 "mito2",
 "mur3",
 "object-store",
@@ -7316,29 +7305,6 @@ dependencies = [
 "windows-sys 0.52.0",
 ]

-[[package]]
-name = "mito-codec"
-version = "0.15.0"
-dependencies = [
- "api",
- "bytes",
- "common-base",
- "common-decimal",
- "common-error",
- "common-macro",
- "common-recordbatch",
- "common-telemetry",
- "common-time",
- "datafusion-common",
- "datafusion-expr",
- "datatypes",
- "memcomparable",
- "paste",
- "serde",
- "snafu 0.8.5",
- "store-api",
-]
-
 [[package]]
 name = "mito2"
 version = "0.15.0"
@@ -7381,7 +7347,6 @@ dependencies = [
 "lazy_static",
 "log-store",
 "memcomparable",
- "mito-codec",
 "moka",
 "object-store",
 "parquet",
@@ -8421,7 +8386,6 @@ dependencies = [
 "common-catalog",
 "common-datasource",
 "common-error",
- "common-frontend",
 "common-function",
 "common-grpc",
 "common-grpc-expr",
@@ -8921,8 +8885,9 @@ dependencies = [

 [[package]]
 name = "pgwire"
-version = "0.30.2"
-source = "git+https://github.com/sunng87/pgwire?rev=127573d997228cfb70c7699881c568eae8131270#127573d997228cfb70c7699881c568eae8131270"
+version = "0.30.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ec79ee18e6cafde8698885646780b967ecc905120798b8359dd0da64f9688e89"
 dependencies = [
 "async-trait",
 "bytes",
@@ -11170,7 +11135,6 @@ dependencies = [
 "common-catalog",
 "common-config",
 "common-error",
- "common-frontend",
 "common-grpc",
 "common-macro",
 "common-mem-prof",
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -49,7 +49,6 @@ members = [
    "src/meta-client",
    "src/meta-srv",
    "src/metric-engine",
-    "src/mito-codec",
    "src/mito2",
    "src/object-store",
    "src/operator",
@@ -117,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 = "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" }
+datafusion = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
+datafusion-common = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
+datafusion-expr = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
+datafusion-functions = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
+datafusion-optimizer = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
+datafusion-physical-expr = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
+datafusion-physical-plan = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
+datafusion-sql = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
+datafusion-substrait = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "e104c7cf62b11dd5fe41461b82514978234326b4" }
 deadpool = "0.12"
 deadpool-postgres = "0.14"
 derive_builder = "0.20"
@@ -134,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 = "17971523673f4fbc982510d3c9d6647ff642e16f" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "2dca1dc67862d7b410838aef81232274c019b3f6" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"
@@ -275,7 +274,6 @@ log-store = { path = "src/log-store" }
 meta-client = { path = "src/meta-client" }
 meta-srv = { path = "src/meta-srv" }
 metric-engine = { path = "src/metric-engine" }
-mito-codec = { path = "src/mito-codec" }
 mito2 = { path = "src/mito2" }
 object-store = { path = "src/object-store" }
 operator = { path = "src/operator" }
--- a/config/config.md
+++ b/config/config.md
@@ -232,7 +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.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. |
 | `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. |
@@ -405,7 +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.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. |
 | `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. |
--- a/config/datanode.example.toml
+++ b/config/datanode.example.toml
@@ -49,7 +49,6 @@ max_send_message_size = "512MB"
 ## - `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.
--- a/config/frontend.example.toml
+++ b/config/frontend.example.toml
@@ -59,7 +59,6 @@ runtime_size = 8
 ## - `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.
--- 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/user-guide/deployments-administration/deploy-on-kubernetes/getting-started).
+The standalone GreptimeDB instance will collect metrics from your cluster, and the dashboard will be available in the Grafana UI. For detailed deployment instructions, please refer to our [Kubernetes deployment guide](https://docs.greptime.com/nightly/user-guide/deployments/deploy-on-kubernetes/getting-started).

 ### Self-host Prometheus and import dashboards manually

--- a/src/catalog/Cargo.toml
+++ b/src/catalog/Cargo.toml
@@ -17,10 +17,8 @@ arrow-schema.workspace = true
 async-stream.workspace = true
 async-trait.workspace = true
 bytes.workspace = true
-common-base.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,26 +277,6 @@ pub enum Error {
        #[snafu(implicit)]
        location: Location,
    },
-
-    #[snafu(display("Failed to invoke frontend services"))]
-    InvokeFrontend {
-        source: common_frontend::error::Error,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
-    #[snafu(display("Meta client is not provided"))]
-    MetaClientMissing {
-        #[snafu(implicit)]
-        location: Location,
-    },
-
-    #[snafu(display("Failed to find frontend node: {}", addr))]
-    FrontendNotFound {
-        addr: String,
-        #[snafu(implicit)]
-        location: Location,
-    },
 }

 impl Error {
@@ -365,10 +345,6 @@ impl ErrorExt for Error {
            Error::GetViewCache { source, .. } | Error::GetTableCache { source, .. } => {
                source.status_code()
            }
-            Error::InvokeFrontend { source, .. } => source.status_code(),
-            Error::FrontendNotFound { .. } | Error::MetaClientMissing { .. } => {
-                StatusCode::Unexpected
-            }
        }
    }

--- a/src/catalog/src/kvbackend/manager.rs
+++ b/src/catalog/src/kvbackend/manager.rs
@@ -51,7 +51,6 @@ 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;
@@ -85,7 +84,6 @@ impl KvBackendCatalogManager {
        backend: KvBackendRef,
        cache_registry: LayeredCacheRegistryRef,
        procedure_manager: Option<ProcedureManagerRef>,
-        process_manager: Option<ProcessManagerRef>,
    ) -> Arc<Self> {
        Arc::new_cyclic(|me| Self {
            information_extension,
@@ -104,14 +102,12 @@ 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,
@@ -423,7 +419,6 @@ struct SystemCatalog {
    information_schema_provider: Arc<InformationSchemaProvider>,
    pg_catalog_provider: Arc<PGCatalogProvider>,
    backend: KvBackendRef,
-    process_manager: Option<ProcessManagerRef>,
 }

 impl SystemCatalog {
@@ -491,7 +486,6 @@ 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,7 +40,6 @@ 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,7 +356,6 @@ 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/metrics.rs
+++ b/src/catalog/src/metrics.rs
@@ -34,20 +34,4 @@ lazy_static! {
        register_histogram!("greptime_catalog_kv_get", "catalog kv get").unwrap();
    pub static ref METRIC_CATALOG_KV_BATCH_GET: Histogram =
        register_histogram!("greptime_catalog_kv_batch_get", "catalog kv batch get").unwrap();
-
-    /// Count of running process in each catalog.
-    pub static ref PROCESS_LIST_COUNT: IntGaugeVec = register_int_gauge_vec!(
-        "greptime_process_list_count",
-        "Running process count per catalog",
-        &["catalog"]
-    )
-    .unwrap();
-
-    /// Count of killed process in each catalog.
-    pub static ref PROCESS_KILL_COUNT: IntCounterVec = register_int_counter_vec!(
-        "greptime_process_kill_count",
-        "Completed kill process requests count",
-        &["catalog"]
-    )
-    .unwrap();
 }
--- a/src/catalog/src/process_manager.rs
+++ b/src/catalog/src/process_manager.rs
@@ -1,488 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::collections::hash_map::Entry;
-use std::collections::HashMap;
-use std::fmt::{Debug, Formatter};
-use std::sync::atomic::{AtomicU64, Ordering};
-use std::sync::{Arc, RwLock};
-
-use api::v1::frontend::{KillProcessRequest, ListProcessRequest, ProcessInfo};
-use common_base::cancellation::CancellationHandle;
-use common_frontend::selector::{FrontendSelector, MetaClientSelector};
-use common_telemetry::{debug, info};
-use common_time::util::current_time_millis;
-use meta_client::MetaClientRef;
-use snafu::{ensure, OptionExt, ResultExt};
-
-use crate::error;
-use crate::metrics::{PROCESS_KILL_COUNT, PROCESS_LIST_COUNT};
-
-pub type ProcessManagerRef = Arc<ProcessManager>;
-
-/// Query process manager.
-pub struct ProcessManager {
-    /// Local frontend server address,
-    server_addr: String,
-    /// Next process id for local queries.
-    next_id: AtomicU64,
-    /// Running process per catalog.
-    catalogs: RwLock<HashMap<String, HashMap<u64, CancellableProcess>>>,
-    /// Frontend selector to locate frontend nodes.
-    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.
-    #[must_use]
-    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(),
-        };
-        let cancellation_handle = Arc::new(CancellationHandle::default());
-        let cancellable_process = CancellableProcess::new(cancellation_handle.clone(), process);
-
-        self.catalogs
-            .write()
-            .unwrap()
-            .entry(catalog.clone())
-            .or_default()
-            .insert(id, cancellable_process);
-
-        Ticket {
-            catalog,
-            manager: self.clone(),
-            id,
-            cancellation_handle,
-        }
-    }
-
-    /// 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().is_empty() {
-                o.remove();
-            }
-        }
-    }
-
-    /// 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().map(|p| p.process.clone()).collect()
-            } else {
-                vec![]
-            }
-        } else {
-            catalogs
-                .values()
-                .flat_map(|v| v.values().map(|p| p.process.clone()))
-                .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::InvokeFrontendSnafu)?;
-            for mut f in frontends {
-                processes.extend(
-                    f.list_process(ListProcessRequest {
-                        catalog: catalog.unwrap_or_default().to_string(),
-                    })
-                    .await
-                    .context(error::InvokeFrontendSnafu)?
-                    .processes,
-                );
-            }
-        }
-        processes.extend(self.local_processes(catalog)?);
-        Ok(processes)
-    }
-
-    /// Kills query with provided catalog and id.
-    pub async fn kill_process(
-        &self,
-        server_addr: String,
-        catalog: String,
-        id: u64,
-    ) -> error::Result<bool> {
-        if server_addr == self.server_addr {
-            if let Some(catalogs) = self.catalogs.write().unwrap().get_mut(&catalog) {
-                if let Some(process) = catalogs.remove(&id) {
-                    process.handle.cancel();
-                    info!(
-                        "Killed process, catalog: {}, id: {:?}",
-                        process.process.catalog, process.process.id
-                    );
-                    PROCESS_KILL_COUNT.with_label_values(&[&catalog]).inc();
-                    Ok(true)
-                } else {
-                    debug!("Failed to kill process, id not found: {}", id);
-                    Ok(false)
-                }
-            } else {
-                debug!("Failed to kill process, catalog not found: {}", catalog);
-                Ok(false)
-            }
-        } else {
-            let mut nodes = self
-                .frontend_selector
-                .as_ref()
-                .context(error::MetaClientMissingSnafu)?
-                .select(|node| node.peer.addr == server_addr)
-                .await
-                .context(error::InvokeFrontendSnafu)?;
-            ensure!(
-                !nodes.is_empty(),
-                error::FrontendNotFoundSnafu { addr: server_addr }
-            );
-
-            let request = KillProcessRequest {
-                server_addr,
-                catalog,
-                process_id: id,
-            };
-            nodes[0]
-                .kill_process(request)
-                .await
-                .context(error::InvokeFrontendSnafu)?;
-            Ok(true)
-        }
-    }
-}
-
-pub struct Ticket {
-    pub(crate) catalog: String,
-    pub(crate) manager: ProcessManagerRef,
-    pub(crate) id: u64,
-    pub cancellation_handle: Arc<CancellationHandle>,
-}
-
-impl Drop for Ticket {
-    fn drop(&mut self) {
-        self.manager
-            .deregister_query(std::mem::take(&mut self.catalog), self.id);
-    }
-}
-
-struct CancellableProcess {
-    handle: Arc<CancellationHandle>,
-    process: ProcessInfo,
-}
-
-impl Drop for CancellableProcess {
-    fn drop(&mut self) {
-        PROCESS_LIST_COUNT
-            .with_label_values(&[&self.process.catalog])
-            .dec();
-    }
-}
-
-impl CancellableProcess {
-    fn new(handle: Arc<CancellationHandle>, process: ProcessInfo) -> Self {
-        PROCESS_LIST_COUNT
-            .with_label_values(&[&process.catalog])
-            .inc();
-        Self { handle, process }
-    }
-}
-
-impl Debug for CancellableProcess {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("CancellableProcess")
-            .field("cancelled", &self.handle.is_cancelled())
-            .field("process", &self.process)
-            .finish()
-    }
-}
-
-#[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);
-    }
-
-    #[tokio::test]
-    async fn test_register_query_with_custom_id() {
-        let process_manager = Arc::new(ProcessManager::new("127.0.0.1:8000".to_string(), None));
-        let custom_id = 12345;
-
-        let ticket = process_manager.clone().register_query(
-            "public".to_string(),
-            vec!["test".to_string()],
-            "SELECT * FROM table".to_string(),
-            "client1".to_string(),
-            Some(custom_id),
-        );
-
-        assert_eq!(ticket.id, custom_id);
-
-        let running_processes = process_manager.local_processes(None).unwrap();
-        assert_eq!(running_processes.len(), 1);
-        assert_eq!(running_processes[0].id, custom_id);
-        assert_eq!(&running_processes[0].client, "client1");
-    }
-
-    #[tokio::test]
-    async fn test_multiple_queries_same_catalog() {
-        let process_manager = Arc::new(ProcessManager::new("127.0.0.1:8000".to_string(), None));
-
-        let ticket1 = process_manager.clone().register_query(
-            "public".to_string(),
-            vec!["schema1".to_string()],
-            "SELECT * FROM table1".to_string(),
-            "client1".to_string(),
-            None,
-        );
-
-        let ticket2 = process_manager.clone().register_query(
-            "public".to_string(),
-            vec!["schema2".to_string()],
-            "SELECT * FROM table2".to_string(),
-            "client2".to_string(),
-            None,
-        );
-
-        let running_processes = process_manager.local_processes(Some("public")).unwrap();
-        assert_eq!(running_processes.len(), 2);
-
-        // Verify both processes are present
-        let ids: Vec<u64> = running_processes.iter().map(|p| p.id).collect();
-        assert!(ids.contains(&ticket1.id));
-        assert!(ids.contains(&ticket2.id));
-    }
-
-    #[tokio::test]
-    async fn test_multiple_catalogs() {
-        let process_manager = Arc::new(ProcessManager::new("127.0.0.1:8000".to_string(), None));
-
-        let _ticket1 = process_manager.clone().register_query(
-            "catalog1".to_string(),
-            vec!["schema1".to_string()],
-            "SELECT * FROM table1".to_string(),
-            "client1".to_string(),
-            None,
-        );
-
-        let _ticket2 = process_manager.clone().register_query(
-            "catalog2".to_string(),
-            vec!["schema2".to_string()],
-            "SELECT * FROM table2".to_string(),
-            "client2".to_string(),
-            None,
-        );
-
-        // Test listing processes for specific catalog
-        let catalog1_processes = process_manager.local_processes(Some("catalog1")).unwrap();
-        assert_eq!(catalog1_processes.len(), 1);
-        assert_eq!(&catalog1_processes[0].catalog, "catalog1");
-
-        let catalog2_processes = process_manager.local_processes(Some("catalog2")).unwrap();
-        assert_eq!(catalog2_processes.len(), 1);
-        assert_eq!(&catalog2_processes[0].catalog, "catalog2");
-
-        // Test listing all processes
-        let all_processes = process_manager.local_processes(None).unwrap();
-        assert_eq!(all_processes.len(), 2);
-    }
-
-    #[tokio::test]
-    async fn test_deregister_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(),
-            "client1".to_string(),
-            None,
-        );
-        assert_eq!(process_manager.local_processes(None).unwrap().len(), 1);
-        process_manager.deregister_query("public".to_string(), ticket.id);
-        assert_eq!(process_manager.local_processes(None).unwrap().len(), 0);
-    }
-
-    #[tokio::test]
-    async fn test_cancellation_handle() {
-        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(),
-            "client1".to_string(),
-            None,
-        );
-
-        assert!(!ticket.cancellation_handle.is_cancelled());
-        ticket.cancellation_handle.cancel();
-        assert!(ticket.cancellation_handle.is_cancelled());
-    }
-
-    #[tokio::test]
-    async fn test_kill_local_process() {
-        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(),
-            "client1".to_string(),
-            None,
-        );
-        assert!(!ticket.cancellation_handle.is_cancelled());
-        let killed = process_manager
-            .kill_process(
-                "127.0.0.1:8000".to_string(),
-                "public".to_string(),
-                ticket.id,
-            )
-            .await
-            .unwrap();
-
-        assert!(killed);
-        assert_eq!(process_manager.local_processes(None).unwrap().len(), 0);
-    }
-
-    #[tokio::test]
-    async fn test_kill_nonexistent_process() {
-        let process_manager = Arc::new(ProcessManager::new("127.0.0.1:8000".to_string(), None));
-        let killed = process_manager
-            .kill_process("127.0.0.1:8000".to_string(), "public".to_string(), 999)
-            .await
-            .unwrap();
-        assert!(!killed);
-    }
-
-    #[tokio::test]
-    async fn test_kill_process_nonexistent_catalog() {
-        let process_manager = Arc::new(ProcessManager::new("127.0.0.1:8000".to_string(), None));
-        let killed = process_manager
-            .kill_process("127.0.0.1:8000".to_string(), "nonexistent".to_string(), 1)
-            .await
-            .unwrap();
-        assert!(!killed);
-    }
-
-    #[tokio::test]
-    async fn test_process_info_fields() {
-        let process_manager = Arc::new(ProcessManager::new("127.0.0.1:8000".to_string(), None));
-
-        let _ticket = process_manager.clone().register_query(
-            "test_catalog".to_string(),
-            vec!["schema1".to_string(), "schema2".to_string()],
-            "SELECT COUNT(*) FROM users WHERE age > 18".to_string(),
-            "test_client".to_string(),
-            Some(42),
-        );
-
-        let processes = process_manager.local_processes(None).unwrap();
-        assert_eq!(processes.len(), 1);
-
-        let process = &processes[0];
-        assert_eq!(process.id, 42);
-        assert_eq!(&process.catalog, "test_catalog");
-        assert_eq!(process.schemas, vec!["schema1", "schema2"]);
-        assert_eq!(&process.query, "SELECT COUNT(*) FROM users WHERE age > 18");
-        assert_eq!(&process.client, "test_client");
-        assert_eq!(&process.frontend, "127.0.0.1:8000");
-        assert!(process.start_timestamp > 0);
-    }
-
-    #[tokio::test]
-    async fn test_ticket_drop_deregisters_process() {
-        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(),
-                "client1".to_string(),
-                None,
-            );
-
-            // Process should be registered
-            assert_eq!(process_manager.local_processes(None).unwrap().len(), 1);
-        } // ticket goes out of scope here
-
-        // Process should be automatically deregistered
-        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,7 +19,6 @@ 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;
@@ -43,7 +42,6 @@ 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;
@@ -52,7 +50,6 @@ 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;
@@ -116,7 +113,6 @@ 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>,
 }
@@ -211,10 +207,6 @@ 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,
        }
    }
@@ -225,13 +217,11 @@ 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(),
        };

@@ -287,9 +277,6 @@ 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
@@ -1,189 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::sync::Arc;
-
-use 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,4 +47,3 @@ 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,7 +328,6 @@ 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,7 +58,6 @@ 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
@@ -1,39 +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.
-
-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/error.rs
+++ b/src/cli/src/error.rs
@@ -30,7 +30,6 @@ pub enum Error {
        location: Location,
        msg: String,
    },
-
    #[snafu(display("Failed to create default catalog and schema"))]
    InitMetadata {
        #[snafu(implicit)]
@@ -229,41 +228,19 @@ 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("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(display("KV backend not set: {}", backend))]
+    KvBackendNotSet {
+        backend: String,
        #[snafu(implicit)]
        location: Location,
    },
@@ -285,9 +262,6 @@ impl ErrorExt for Error {
            | Error::ConnectEtcd { .. }
            | Error::CreateDir { .. }
            | Error::EmptyResult { .. }
-            | Error::InvalidFilePath { .. }
-            | Error::UnsupportedMemoryBackend { .. }
-            | Error::InvalidArguments { .. }
            | Error::ParseProxyOpts { .. } => StatusCode::InvalidArguments,

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

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

--- a/src/cli/src/data/export.rs
+++ b/src/cli/src/data/export.rs
@@ -50,7 +50,6 @@ 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,7 +40,6 @@ 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/lib.rs
+++ b/src/cli/src/lib.rs
@@ -13,10 +13,11 @@
 // limitations under the License.

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

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

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

 #[async_trait]
 pub trait Tool: Send + Sync {
--- a/src/cli/src/metadata/snapshot.rs
+++ b/src/cli/src/metadata/snapshot.rs
@@ -12,38 +12,96 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

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

 use async_trait::async_trait;
-use clap::{Parser, Subcommand};
+use clap::Parser;
 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::{OptionExt, ResultExt};
+use snafu::ResultExt;

-use crate::error::{InvalidFilePathSnafu, OpenDalSnafu, S3ConfigNotSetSnafu};
-use crate::metadata::common::StoreConfig;
+use crate::error::{KvBackendNotSetSnafu, OpenDalSnafu, S3ConfigNotSetSnafu};
 use crate::Tool;
-
-/// 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),
+#[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,
 }

-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,
+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))
+            }
        }
    }
 }
@@ -112,10 +170,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 SaveCommand {
-    /// The store configuration.
+pub struct MetaSnapshotCommand {
+    /// The connection to the metadata store.
    #[clap(flatten)]
-    store: StoreConfig,
+    connection: MetaConnection,
    /// The s3 config.
    #[clap(flatten)]
    s3_config: S3Config,
@@ -138,9 +196,9 @@ fn create_local_file_object_store(root: &str) -> Result<ObjectStore, BoxedError>
    Ok(object_store)
 }

-impl SaveCommand {
+impl MetaSnapshotCommand {
    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
-        let kvbackend = self.store.build().await?;
+        let kvbackend = self.connection.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 {
@@ -160,7 +218,7 @@ impl SaveCommand {
    }
 }

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

-/// 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.
+/// 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.
 #[derive(Debug, Default, Parser)]
-pub struct RestoreCommand {
-    /// The store configuration.
+pub struct MetaRestoreCommand {
+    /// The connection to the metadata store.
    #[clap(flatten)]
-    store: StoreConfig,
+    connection: MetaConnection,
    /// The s3 config.
    #[clap(flatten)]
    s3_config: S3Config,
@@ -199,9 +255,9 @@ pub struct RestoreCommand {
    force: bool,
 }

-impl RestoreCommand {
+impl MetaRestoreCommand {
    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
-        let kvbackend = self.store.build().await?;
+        let kvbackend = self.connection.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 {
@@ -223,7 +279,7 @@ impl RestoreCommand {
    }
 }

-struct MetaRestoreTool {
+pub struct MetaRestoreTool {
    inner: MetadataSnapshotManager,
    source_file: String,
    force: bool,
@@ -271,93 +327,3 @@ 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/cli/src/metadata.rs
+++ b/src/cli/src/metadata.rs
@@ -1,42 +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.
-
-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
@@ -1,116 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::sync::Arc;
-
-use 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
@@ -1,38 +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.
-
-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
@@ -1,242 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::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
@@ -1,36 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use 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/client/src/client.rs
+++ b/src/client/src/client.rs
@@ -162,23 +162,12 @@ impl Client {
            .as_bytes() as usize
    }

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

-        let mut client = FlightServiceClient::new(channel)
+        let 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,16 +49,7 @@ impl NodeManager for NodeClients {
    async fn datanode(&self, datanode: &Peer) -> DatanodeRef {
        let client = self.get_client(datanode).await;

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

    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(false, false)?;
+        let mut client = self.client.make_flight_client()?;

        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(false, false)?;
+        let mut client = self.client.make_flight_client()?;
        let response = client.mut_inner().do_put(request).await?;
        let response = response
            .into_inner()
--- a/src/client/src/flow.rs
+++ b/src/client/src/flow.rs
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use api::v1::flow::{DirtyWindowRequest, DirtyWindowRequests, FlowRequest, FlowResponse};
+use api::v1::flow::{FlowRequest, FlowResponse};
 use api::v1::region::InsertRequests;
 use common_error::ext::BoxedError;
 use common_meta::node_manager::Flownode;
@@ -44,16 +44,6 @@ impl Flownode for FlowRequester {
            .map_err(BoxedError::new)
            .context(common_meta::error::ExternalSnafu)
    }
-
-    async fn handle_mark_window_dirty(
-        &self,
-        req: DirtyWindowRequest,
-    ) -> common_meta::error::Result<FlowResponse> {
-        self.handle_mark_window_dirty(req)
-            .await
-            .map_err(BoxedError::new)
-            .context(common_meta::error::ExternalSnafu)
-    }
 }

 impl FlowRequester {
@@ -101,20 +91,4 @@ impl FlowRequester {
            .into_inner();
        Ok(response)
    }
-
-    async fn handle_mark_window_dirty(&self, req: DirtyWindowRequest) -> Result<FlowResponse> {
-        let (addr, mut client) = self.client.raw_flow_client()?;
-        let response = client
-            .handle_mark_dirty_time_window(DirtyWindowRequests {
-                requests: vec![req],
-            })
-            .await
-            .or_else(|e| {
-                let code = e.code();
-                let err: crate::error::Error = e.into();
-                Err(BoxedError::new(err)).context(FlowServerSnafu { addr, code })
-            })?
-            .into_inner();
-        Ok(response)
-    }
 }
--- a/src/client/src/region.rs
+++ b/src/client/src/region.rs
@@ -46,8 +46,6 @@ use crate::{metrics, Client, Error};
 #[derive(Debug)]
 pub struct RegionRequester {
    client: Client,
-    send_compression: bool,
-    accept_compression: bool,
 }

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

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

    pub async fn do_get_inner(&self, ticket: Ticket) -> Result<SendableRecordBatchStream> {
-        let mut flight_client = self
-            .client
-            .make_flight_client(self.send_compression, self.accept_compression)?;
+        let mut flight_client = self.client.make_flight_client()?;
        let response = flight_client
            .mut_inner()
            .do_get(ticket)
--- a/src/cmd/src/cli.rs
+++ b/src/cmd/src/cli.rs
@@ -146,7 +146,6 @@ 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,7 +323,6 @@ 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,7 +20,6 @@ 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;
@@ -39,7 +38,6 @@ 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};
@@ -344,17 +342,11 @@ 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![
@@ -372,16 +364,12 @@ impl StartCommand {

        // frontend to datanode need not timeout.
        // Some queries are expected to take long time.
-        let mut channel_config = ChannelConfig {
+        let 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(
@@ -391,7 +379,6 @@ 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/standalone.rs
+++ b/src/cmd/src/standalone.rs
@@ -21,7 +21,6 @@ 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;
@@ -527,14 +526,11 @@ 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 =
@@ -624,7 +620,6 @@ impl StartCommand {
            catalog_manager.clone(),
            node_manager.clone(),
            ddl_task_executor.clone(),
-            process_manager,
        )
        .with_plugin(plugins.clone())
        .try_build()
@@ -652,7 +647,7 @@ impl StartCommand {
            node_manager,
        )
        .await
-        .context(StartFlownodeSnafu)?;
+        .context(error::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
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use std::path::Path;
 use std::time::Duration;

 use cmd::options::GreptimeOptions;
@@ -57,7 +58,12 @@ fn test_load_datanode_example_config() {
                metadata_cache_tti: Duration::from_secs(300),
            }),
            wal: DatanodeWalConfig::RaftEngine(RaftEngineConfig {
-                dir: Some(format!("{}/{}", DEFAULT_DATA_HOME, WAL_DIR)),
+                dir: Some(
+                    Path::new(DEFAULT_DATA_HOME)
+                        .join(WAL_DIR)
+                        .to_string_lossy()
+                        .to_string(),
+                ),
                sync_period: Some(Duration::from_secs(10)),
                recovery_parallelism: 2,
                ..Default::default()
@@ -80,7 +86,10 @@ fn test_load_datanode_example_config() {
            ],
            logging: LoggingOptions {
                level: Some("info".to_string()),
-                dir: format!("{}/{}", DEFAULT_DATA_HOME, DEFAULT_LOGGING_DIR),
+                dir: Path::new(DEFAULT_DATA_HOME)
+                    .join(DEFAULT_LOGGING_DIR)
+                    .to_string_lossy()
+                    .to_string(),
                otlp_endpoint: Some(DEFAULT_OTLP_ENDPOINT.to_string()),
                tracing_sample_ratio: Some(Default::default()),
                ..Default::default()
@@ -123,7 +132,10 @@ fn test_load_frontend_example_config() {
            }),
            logging: LoggingOptions {
                level: Some("info".to_string()),
-                dir: format!("{}/{}", DEFAULT_DATA_HOME, DEFAULT_LOGGING_DIR),
+                dir: Path::new(DEFAULT_DATA_HOME)
+                    .join(DEFAULT_LOGGING_DIR)
+                    .to_string_lossy()
+                    .to_string(),
                otlp_endpoint: Some(DEFAULT_OTLP_ENDPOINT.to_string()),
                tracing_sample_ratio: Some(Default::default()),
                ..Default::default()
@@ -170,7 +182,10 @@ fn test_load_metasrv_example_config() {
                ..Default::default()
            },
            logging: LoggingOptions {
-                dir: format!("{}/{}", DEFAULT_DATA_HOME, DEFAULT_LOGGING_DIR),
+                dir: Path::new(DEFAULT_DATA_HOME)
+                    .join(DEFAULT_LOGGING_DIR)
+                    .to_string_lossy()
+                    .to_string(),
                level: Some("info".to_string()),
                otlp_endpoint: Some(DEFAULT_OTLP_ENDPOINT.to_string()),
                tracing_sample_ratio: Some(Default::default()),
@@ -205,7 +220,12 @@ fn test_load_standalone_example_config() {
        component: StandaloneOptions {
            default_timezone: Some("UTC".to_string()),
            wal: DatanodeWalConfig::RaftEngine(RaftEngineConfig {
-                dir: Some(format!("{}/{}", DEFAULT_DATA_HOME, WAL_DIR)),
+                dir: Some(
+                    Path::new(DEFAULT_DATA_HOME)
+                        .join(WAL_DIR)
+                        .to_string_lossy()
+                        .to_string(),
+                ),
                sync_period: Some(Duration::from_secs(10)),
                recovery_parallelism: 2,
                ..Default::default()
@@ -228,7 +248,10 @@ fn test_load_standalone_example_config() {
            },
            logging: LoggingOptions {
                level: Some("info".to_string()),
-                dir: format!("{}/{}", DEFAULT_DATA_HOME, DEFAULT_LOGGING_DIR),
+                dir: Path::new(DEFAULT_DATA_HOME)
+                    .join(DEFAULT_LOGGING_DIR)
+                    .to_string_lossy()
+                    .to_string(),
                otlp_endpoint: Some(DEFAULT_OTLP_ENDPOINT.to_string()),
                tracing_sample_ratio: Some(Default::default()),
                ..Default::default()
--- a/src/common/base/src/cancellation.rs
+++ b/src/common/base/src/cancellation.rs
@@ -1,240 +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.
-
-//! [CancellationHandle] is used to compose with manual implementation of [futures::future::Future]
-//! or [futures::stream::Stream] to facilitate cancellation.
-//! See example in [frontend::stream_wrapper::CancellableStreamWrapper] and [CancellableFuture].
-
-use std::fmt::{Debug, Display, Formatter};
-use std::future::Future;
-use std::pin::Pin;
-use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::Arc;
-use std::task::{Context, Poll};
-
-use futures::task::AtomicWaker;
-use pin_project::pin_project;
-
-#[derive(Default)]
-pub struct CancellationHandle {
-    waker: AtomicWaker,
-    cancelled: AtomicBool,
-}
-
-impl Debug for CancellationHandle {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("CancellationHandle")
-            .field("cancelled", &self.is_cancelled())
-            .finish()
-    }
-}
-
-impl CancellationHandle {
-    pub fn waker(&self) -> &AtomicWaker {
-        &self.waker
-    }
-
-    /// Cancels a future or stream.
-    pub fn cancel(&self) {
-        if self
-            .cancelled
-            .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
-            .is_ok()
-        {
-            self.waker.wake();
-        }
-    }
-
-    /// Is this handle cancelled.
-    pub fn is_cancelled(&self) -> bool {
-        self.cancelled.load(Ordering::Relaxed)
-    }
-}
-
-#[pin_project]
-#[derive(Debug, Clone)]
-pub struct CancellableFuture<T> {
-    #[pin]
-    fut: T,
-    handle: Arc<CancellationHandle>,
-}
-
-impl<T> CancellableFuture<T> {
-    pub fn new(fut: T, handle: Arc<CancellationHandle>) -> Self {
-        Self { fut, handle }
-    }
-}
-
-impl<T> Future for CancellableFuture<T>
-where
-    T: Future,
-{
-    type Output = Result<T::Output, Cancelled>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let this = self.as_mut().project();
-        // Check if the task has been aborted
-        if this.handle.is_cancelled() {
-            return Poll::Ready(Err(Cancelled));
-        }
-
-        if let Poll::Ready(x) = this.fut.poll(cx) {
-            return Poll::Ready(Ok(x));
-        }
-
-        this.handle.waker().register(cx.waker());
-        if this.handle.is_cancelled() {
-            return Poll::Ready(Err(Cancelled));
-        }
-        Poll::Pending
-    }
-}
-
-#[derive(Copy, Clone, Debug)]
-pub struct Cancelled;
-
-impl Display for Cancelled {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "Future has been cancelled")
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::sync::Arc;
-    use std::time::Duration;
-
-    use tokio::time::{sleep, timeout};
-
-    use crate::cancellation::{CancellableFuture, CancellationHandle, Cancelled};
-
-    #[tokio::test]
-    async fn test_cancellable_future_completes_normally() {
-        let handle = Arc::new(CancellationHandle::default());
-        let future = async { 42 };
-        let cancellable = CancellableFuture::new(future, handle);
-
-        let result = cancellable.await;
-        assert!(result.is_ok());
-        assert_eq!(result.unwrap(), 42);
-    }
-
-    #[tokio::test]
-    async fn test_cancellable_future_cancelled_before_start() {
-        let handle = Arc::new(CancellationHandle::default());
-        handle.cancel();
-
-        let future = async { 42 };
-        let cancellable = CancellableFuture::new(future, handle);
-
-        let result = cancellable.await;
-        assert!(result.is_err());
-        assert!(matches!(result.unwrap_err(), Cancelled));
-    }
-
-    #[tokio::test]
-    async fn test_cancellable_future_cancelled_during_execution() {
-        let handle = Arc::new(CancellationHandle::default());
-        let handle_clone = handle.clone();
-
-        // Create a future that sleeps for a long time
-        let future = async {
-            sleep(Duration::from_secs(10)).await;
-            42
-        };
-        let cancellable = CancellableFuture::new(future, handle);
-
-        // Cancel the future after a short delay
-        tokio::spawn(async move {
-            sleep(Duration::from_millis(50)).await;
-            handle_clone.cancel();
-        });
-
-        let result = cancellable.await;
-        assert!(result.is_err());
-        assert!(matches!(result.unwrap_err(), Cancelled));
-    }
-
-    #[tokio::test]
-    async fn test_cancellable_future_completes_before_cancellation() {
-        let handle = Arc::new(CancellationHandle::default());
-        let handle_clone = handle.clone();
-
-        // Create a future that completes quickly
-        let future = async {
-            sleep(Duration::from_millis(10)).await;
-            42
-        };
-        let cancellable = CancellableFuture::new(future, handle);
-
-        // Try to cancel after the future should have completed
-        tokio::spawn(async move {
-            sleep(Duration::from_millis(100)).await;
-            handle_clone.cancel();
-        });
-
-        let result = cancellable.await;
-        assert!(result.is_ok());
-        assert_eq!(result.unwrap(), 42);
-    }
-
-    #[tokio::test]
-    async fn test_cancellation_handle_is_cancelled() {
-        let handle = CancellationHandle::default();
-        assert!(!handle.is_cancelled());
-
-        handle.cancel();
-        assert!(handle.is_cancelled());
-    }
-
-    #[tokio::test]
-    async fn test_multiple_cancellable_futures_with_same_handle() {
-        let handle = Arc::new(CancellationHandle::default());
-
-        let future1 = CancellableFuture::new(async { 1 }, handle.clone());
-        let future2 = CancellableFuture::new(async { 2 }, handle.clone());
-
-        // Cancel before starting
-        handle.cancel();
-
-        let (result1, result2) = tokio::join!(future1, future2);
-
-        assert!(result1.is_err());
-        assert!(result2.is_err());
-        assert!(matches!(result1.unwrap_err(), Cancelled));
-        assert!(matches!(result2.unwrap_err(), Cancelled));
-    }
-
-    #[tokio::test]
-    async fn test_cancellable_future_with_timeout() {
-        let handle = Arc::new(CancellationHandle::default());
-        let future = async {
-            sleep(Duration::from_secs(1)).await;
-            42
-        };
-        let cancellable = CancellableFuture::new(future, handle.clone());
-
-        // Use timeout to ensure the test doesn't hang
-        let result = timeout(Duration::from_millis(100), cancellable).await;
-
-        // Should timeout because the future takes 1 second but we timeout after 100ms
-        assert!(result.is_err());
-    }
-
-    #[tokio::test]
-    async fn test_cancelled_display() {
-        let cancelled = Cancelled;
-        assert_eq!(format!("{}", cancelled), "Future has been cancelled");
-    }
-}
--- a/src/common/base/src/lib.rs
+++ b/src/common/base/src/lib.rs
@@ -14,7 +14,6 @@

 pub mod bit_vec;
 pub mod bytes;
-pub mod cancellation;
 pub mod plugins;
 pub mod range_read;
 #[allow(clippy::all)]
--- a/src/common/catalog/src/consts.rs
+++ b/src/common/catalog/src/consts.rs
@@ -102,8 +102,6 @@ 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,13 +7,5 @@ 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,35 +27,6 @@ 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 frontend service"))]
-    InvokeFrontend {
-        #[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>;
@@ -65,10 +36,6 @@ impl ErrorExt for Error {
        use Error::*;
        match self {
            External { source, .. } => source.status_code(),
-            Meta { source, .. } => source.status_code(),
-            ParseProcessId { .. } => StatusCode::InvalidArguments,
-            InvokeFrontend { .. } => StatusCode::Unexpected,
-            CreateChannel { source, .. } => source.status_code(),
        }
    }

--- a/src/common/frontend/src/lib.rs
+++ b/src/common/frontend/src/lib.rs
@@ -12,41 +12,4 @@
 // 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
@@ -1,112 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::time::Duration;
-
-use common_grpc::channel_manager::{ChannelConfig, ChannelManager};
-use common_meta::cluster::{ClusterInfo, NodeInfo, Role};
-use greptime_proto::v1::frontend::{
-    frontend_client, KillProcessRequest, KillProcessResponse, ListProcessRequest,
-    ListProcessResponse,
-};
-use meta_client::MetaClientRef;
-use snafu::ResultExt;
-use tonic::Response;
-
-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 fn kill_process(&mut self, req: KillProcessRequest) -> Result<KillProcessResponse>;
-}
-
-#[async_trait::async_trait]
-impl FrontendClient for frontend_client::FrontendClient<tonic::transport::channel::Channel> {
-    async fn list_process(&mut self, req: ListProcessRequest) -> Result<ListProcessResponse> {
-        frontend_client::FrontendClient::<tonic::transport::channel::Channel>::list_process(
-            self, req,
-        )
-        .await
-        .context(error::InvokeFrontendSnafu)
-        .map(Response::into_inner)
-    }
-
-    async fn kill_process(&mut self, req: KillProcessRequest) -> Result<KillProcessResponse> {
-        frontend_client::FrontendClient::<tonic::transport::channel::Channel>::kill_process(
-            self, req,
-        )
-        .await
-        .context(error::InvokeFrontendSnafu)
-        .map(Response::into_inner)
-    }
-}
-
-#[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/adjust_flow.rs
+++ b/src/common/function/src/adjust_flow.rs
@@ -0,0 +1,90 @@
+// 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_macro::admin_fn;
+use common_query::error::{
+    InvalidFuncArgsSnafu, MissingFlowServiceHandlerSnafu, Result, UnsupportedInputDataTypeSnafu,
+};
+use common_query::prelude::Signature;
+use datafusion::logical_expr::Volatility;
+use datatypes::value::{Value, ValueRef};
+use session::context::QueryContextRef;
+use snafu::ensure;
+use store_api::storage::ConcreteDataType;
+
+use crate::handlers::FlowServiceHandlerRef;
+use crate::helper::parse_catalog_flow;
+
+fn adjust_signature() -> Signature {
+    Signature::exact(
+        vec![
+            ConcreteDataType::string_datatype(), // flow name
+            ConcreteDataType::uint64_datatype(), // min_run_interval in seconds
+            ConcreteDataType::uint64_datatype(), // max filter number per query
+        ],
+        Volatility::Immutable,
+    )
+}
+
+#[admin_fn(
+    name = AdjustFlowFunction,
+    display_name = adjust_flow,
+    sig_fn = adjust_signature,
+    ret = uint64
+)]
+pub(crate) async fn adjust_flow(
+    flow_service_handler: &FlowServiceHandlerRef,
+    query_ctx: &QueryContextRef,
+    params: &[ValueRef<'_>],
+) -> Result<Value> {
+    ensure!(
+        params.len() == 3,
+        InvalidFuncArgsSnafu {
+            err_msg: format!(
+                "The length of the args is not correct, expect 3, have: {}",
+                params.len()
+            ),
+        }
+    );
+
+    let (flow_name, min_run_interval, max_filter_num) = match (params[0], params[1], params[2]) {
+        (
+            ValueRef::String(flow_name),
+            ValueRef::UInt64(min_run_interval),
+            ValueRef::UInt64(max_filter_num),
+        ) => (flow_name, min_run_interval, max_filter_num),
+        _ => {
+            return UnsupportedInputDataTypeSnafu {
+                function: "adjust_flow",
+                datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
+            }
+            .fail();
+        }
+    };
+
+    let (catalog_name, flow_name) = parse_catalog_flow(flow_name, query_ctx)?;
+
+    let res = flow_service_handler
+        .adjust(
+            &catalog_name,
+            &flow_name,
+            min_run_interval,
+            max_filter_num as usize,
+            query_ctx.clone(),
+        )
+        .await?;
+    let affected_rows = res.affected_rows;
+
+    Ok(Value::from(affected_rows))
+}
--- a/src/common/function/src/admin.rs
+++ b/src/common/function/src/admin.rs
@@ -26,6 +26,7 @@ use flush_compact_table::{CompactTableFunction, FlushTableFunction};
 use migrate_region::MigrateRegionFunction;
 use remove_region_follower::RemoveRegionFollowerFunction;

+use crate::adjust_flow::AdjustFlowFunction;
 use crate::flush_flow::FlushFlowFunction;
 use crate::function_registry::FunctionRegistry;

@@ -43,5 +44,6 @@ impl AdminFunction {
        registry.register_async(Arc::new(FlushTableFunction));
        registry.register_async(Arc::new(CompactTableFunction));
        registry.register_async(Arc::new(FlushFlowFunction));
+        registry.register_async(Arc::new(AdjustFlowFunction));
    }
 }
--- a/src/common/function/src/flush_flow.rs
+++ b/src/common/function/src/flush_flow.rs
@@ -12,21 +12,19 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use common_error::ext::BoxedError;
 use common_macro::admin_fn;
 use common_query::error::{
-    ExecuteSnafu, InvalidFuncArgsSnafu, MissingFlowServiceHandlerSnafu, Result,
-    UnsupportedInputDataTypeSnafu,
+    InvalidFuncArgsSnafu, MissingFlowServiceHandlerSnafu, Result, UnsupportedInputDataTypeSnafu,
 };
 use common_query::prelude::Signature;
 use datafusion::logical_expr::Volatility;
 use datatypes::value::{Value, ValueRef};
 use session::context::QueryContextRef;
-use snafu::{ensure, ResultExt};
-use sql::parser::ParserContext;
+use snafu::ensure;
 use store_api::storage::ConcreteDataType;

 use crate::handlers::FlowServiceHandlerRef;
+use crate::helper::parse_catalog_flow;

 fn flush_signature() -> Signature {
    Signature::uniform(
@@ -47,20 +45,6 @@ pub(crate) async fn flush_flow(
    query_ctx: &QueryContextRef,
    params: &[ValueRef<'_>],
 ) -> Result<Value> {
-    let (catalog_name, flow_name) = parse_flush_flow(params, query_ctx)?;
-
-    let res = flow_service_handler
-        .flush(&catalog_name, &flow_name, query_ctx.clone())
-        .await?;
-    let affected_rows = res.affected_rows;
-
-    Ok(Value::from(affected_rows))
-}
-
-fn parse_flush_flow(
-    params: &[ValueRef<'_>],
-    query_ctx: &QueryContextRef,
-) -> Result<(String, String)> {
    ensure!(
        params.len() == 1,
        InvalidFuncArgsSnafu {
@@ -70,7 +54,6 @@ fn parse_flush_flow(
            ),
        }
    );
-
    let ValueRef::String(flow_name) = params[0] else {
        return UnsupportedInputDataTypeSnafu {
            function: "flush_flow",
@@ -78,27 +61,14 @@ fn parse_flush_flow(
        }
        .fail();
    };
-    let obj_name = ParserContext::parse_table_name(flow_name, query_ctx.sql_dialect())
-        .map_err(BoxedError::new)
-        .context(ExecuteSnafu)?;
+    let (catalog_name, flow_name) = parse_catalog_flow(flow_name, query_ctx)?;

-    let (catalog_name, flow_name) = match &obj_name.0[..] {
-        [flow_name] => (
-            query_ctx.current_catalog().to_string(),
-            flow_name.value.clone(),
-        ),
-        [catalog, flow_name] => (catalog.value.clone(), flow_name.value.clone()),
-        _ => {
-            return InvalidFuncArgsSnafu {
-                err_msg: format!(
-                    "expect flow name to be <catalog>.<flow-name> or <flow-name>, actual: {}",
-                    obj_name
-                ),
-            }
-            .fail()
-        }
-    };
-    Ok((catalog_name, flow_name))
+    let res = flow_service_handler
+        .flush(&catalog_name, &flow_name, query_ctx.clone())
+        .await?;
+    let affected_rows = res.affected_rows;
+
+    Ok(Value::from(affected_rows))
 }

 #[cfg(test)]
@@ -154,10 +124,7 @@ mod test {
            ("catalog.flow_name", ("catalog", "flow_name")),
        ];
        for (input, expected) in testcases.iter() {
-            let args = vec![*input];
-            let args = args.into_iter().map(ValueRef::String).collect::<Vec<_>>();
-
-            let result = parse_flush_flow(&args, &QueryContext::arc()).unwrap();
+            let result = parse_catalog_flow(input, &QueryContext::arc()).unwrap();
            assert_eq!(*expected, (result.0.as_str(), result.1.as_str()));
        }
    }
--- a/src/common/function/src/handlers.rs
+++ b/src/common/function/src/handlers.rs
@@ -87,6 +87,15 @@ pub trait FlowServiceHandler: Send + Sync {
        flow: &str,
        ctx: QueryContextRef,
    ) -> Result<api::v1::flow::FlowResponse>;
+
+    async fn adjust(
+        &self,
+        catalog: &str,
+        flow: &str,
+        min_run_interval_secs: u64,
+        max_filter_num_per_query: usize,
+        ctx: QueryContextRef,
+    ) -> Result<api::v1::flow::FlowResponse>;
 }

 pub type TableMutationHandlerRef = Arc<dyn TableMutationHandler>;
--- a/src/common/function/src/helper.rs
+++ b/src/common/function/src/helper.rs
@@ -12,12 +12,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use common_query::error::{InvalidInputTypeSnafu, Result};
+use common_error::ext::BoxedError;
+use common_query::error::{ExecuteSnafu, InvalidFuncArgsSnafu, InvalidInputTypeSnafu, Result};
 use common_query::prelude::{Signature, TypeSignature, Volatility};
 use datatypes::prelude::ConcreteDataType;
 use datatypes::types::cast::cast;
 use datatypes::value::ValueRef;
+use session::context::QueryContextRef;
 use snafu::ResultExt;
+use sql::parser::ParserContext;

 /// Create a function signature with oneof signatures of interleaving two arguments.
 pub fn one_of_sigs2(args1: Vec<ConcreteDataType>, args2: Vec<ConcreteDataType>) -> Signature {
@@ -43,3 +46,30 @@ pub fn cast_u64(value: &ValueRef) -> Result<Option<u64>> {
        })
        .map(|v| v.as_u64())
 }
+
+pub fn parse_catalog_flow(
+    flow_name: &str,
+    query_ctx: &QueryContextRef,
+) -> Result<(String, String)> {
+    let obj_name = ParserContext::parse_table_name(flow_name, query_ctx.sql_dialect())
+        .map_err(BoxedError::new)
+        .context(ExecuteSnafu)?;
+
+    let (catalog_name, flow_name) = match &obj_name.0[..] {
+        [flow_name] => (
+            query_ctx.current_catalog().to_string(),
+            flow_name.value.clone(),
+        ),
+        [catalog, flow_name] => (catalog.value.clone(), flow_name.value.clone()),
+        _ => {
+            return InvalidFuncArgsSnafu {
+                err_msg: format!(
+                    "expect flow name to be <catalog>.<flow-name> or <flow-name>, actual: {}",
+                    obj_name
+                ),
+            }
+            .fail()
+        }
+    };
+    Ok((catalog_name, flow_name))
+}
--- a/src/common/function/src/lib.rs
+++ b/src/common/function/src/lib.rs
@@ -15,6 +15,7 @@
 #![feature(let_chains)]
 #![feature(try_blocks)]

+mod adjust_flow;
 mod admin;
 mod flush_flow;
 mod macros;
--- a/src/common/function/src/state.rs
+++ b/src/common/function/src/state.rs
@@ -148,6 +148,17 @@ impl FunctionState {
            ) -> Result<api::v1::flow::FlowResponse> {
                todo!()
            }
+
+            async fn adjust(
+                &self,
+                _catalog: &str,
+                _flow: &str,
+                _min_run_interval_secs: u64,
+                _max_filter_num_per_query: usize,
+                _ctx: QueryContextRef,
+            ) -> Result<api::v1::flow::FlowResponse> {
+                todo!()
+            }
        }

        Self {
--- a/src/common/function/src/system.rs
+++ b/src/common/function/src/system.rs
@@ -23,8 +23,7 @@ use std::sync::Arc;

 use build::BuildFunction;
 use database::{
-    ConnectionIdFunction, CurrentSchemaFunction, DatabaseFunction, PgBackendPidFunction,
-    ReadPreferenceFunction, SessionUserFunction,
+    CurrentSchemaFunction, DatabaseFunction, ReadPreferenceFunction, SessionUserFunction,
 };
 use pg_catalog::PGCatalogFunction;
 use procedure_state::ProcedureStateFunction;
@@ -43,8 +42,6 @@ impl SystemFunction {
        registry.register_scalar(DatabaseFunction);
        registry.register_scalar(SessionUserFunction);
        registry.register_scalar(ReadPreferenceFunction);
-        registry.register_scalar(PgBackendPidFunction);
-        registry.register_scalar(ConnectionIdFunction);
        registry.register_scalar(TimezoneFunction);
        registry.register_async(Arc::new(ProcedureStateFunction));
        PGCatalogFunction::register(registry);
--- a/src/common/function/src/system/database.rs
+++ b/src/common/function/src/system/database.rs
@@ -18,8 +18,7 @@ use std::sync::Arc;
 use common_query::error::Result;
 use common_query::prelude::{Signature, Volatility};
 use datatypes::prelude::{ConcreteDataType, ScalarVector};
-use datatypes::vectors::{StringVector, UInt64Vector, VectorRef};
-use derive_more::Display;
+use datatypes::vectors::{StringVector, VectorRef};

 use crate::function::{Function, FunctionContext};

@@ -33,20 +32,10 @@ pub struct SessionUserFunction;

 pub struct ReadPreferenceFunction;

-#[derive(Display)]
-#[display("{}", self.name())]
-pub struct PgBackendPidFunction;
-
-#[derive(Display)]
-#[display("{}", self.name())]
-pub struct ConnectionIdFunction;
-
 const DATABASE_FUNCTION_NAME: &str = "database";
 const CURRENT_SCHEMA_FUNCTION_NAME: &str = "current_schema";
 const SESSION_USER_FUNCTION_NAME: &str = "session_user";
 const READ_PREFERENCE_FUNCTION_NAME: &str = "read_preference";
-const PG_BACKEND_PID: &str = "pg_backend_pid";
-const CONNECTION_ID: &str = "connection_id";

 impl Function for DatabaseFunction {
    fn name(&self) -> &str {
@@ -128,46 +117,6 @@ impl Function for ReadPreferenceFunction {
    }
 }

-impl Function for PgBackendPidFunction {
-    fn name(&self) -> &str {
-        PG_BACKEND_PID
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::uint64_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        Signature::nullary(Volatility::Immutable)
-    }
-
-    fn eval(&self, func_ctx: &FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
-        let pid = func_ctx.query_ctx.process_id();
-
-        Ok(Arc::new(UInt64Vector::from_slice([pid])) as _)
-    }
-}
-
-impl Function for ConnectionIdFunction {
-    fn name(&self) -> &str {
-        CONNECTION_ID
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::uint64_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        Signature::nullary(Volatility::Immutable)
-    }
-
-    fn eval(&self, func_ctx: &FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
-        let pid = func_ctx.query_ctx.process_id();
-
-        Ok(Arc::new(UInt64Vector::from_slice([pid])) as _)
-    }
-}
-
 impl fmt::Display for DatabaseFunction {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "DATABASE")
--- a/src/common/grpc/src/channel_manager.rs
+++ b/src/common/grpc/src/channel_manager.rs
@@ -296,8 +296,6 @@ 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 {
@@ -318,8 +316,6 @@ 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,
        }
    }
 }
@@ -570,8 +566,6 @@ 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
        );
@@ -616,8 +610,6 @@ 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/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,
-            DatanodeTableInfoNotFound { .. } => StatusCode::Internal,
+            Error::DatanodeTableInfoNotFound { .. } => StatusCode::Internal,
        }
    }

--- a/src/common/meta/src/key/view_info.rs
+++ b/src/common/meta/src/key/view_info.rs
@@ -70,12 +70,11 @@ impl MetadataKey<'_, ViewInfoKey> for ViewInfoKey {
            }
            .build()
        })?;
-        let captures =
-            VIEW_INFO_KEY_PATTERN
-                .captures(key)
-                .with_context(|| InvalidViewInfoSnafu {
-                    err_msg: format!("Invalid ViewInfoKey '{key}'"),
-                })?;
+        let captures = VIEW_INFO_KEY_PATTERN
+            .captures(key)
+            .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/node_manager.rs
+++ b/src/common/meta/src/node_manager.rs
@@ -15,7 +15,7 @@
 use std::sync::Arc;

 use api::region::RegionResponse;
-use api::v1::flow::{DirtyWindowRequest, FlowRequest, FlowResponse};
+use api::v1::flow::{FlowRequest, FlowResponse};
 use api::v1::region::{InsertRequests, RegionRequest};
 pub use common_base::AffectedRows;
 use common_query::request::QueryRequest;
@@ -42,9 +42,6 @@ pub trait Flownode: Send + Sync {
    async fn handle(&self, request: FlowRequest) -> Result<FlowResponse>;

    async fn handle_inserts(&self, request: InsertRequests) -> Result<FlowResponse>;
-
-    /// Handles requests to mark time window as dirty.
-    async fn handle_mark_window_dirty(&self, req: DirtyWindowRequest) -> Result<FlowResponse>;
 }

 pub type FlownodeRef = Arc<dyn Flownode>;
--- a/src/common/meta/src/snapshot.rs
+++ b/src/common/meta/src/snapshot.rs
@@ -14,7 +14,6 @@

 pub mod file;

-use std::borrow::Cow;
 use std::fmt::{Display, Formatter};
 use std::path::{Path, PathBuf};
 use std::time::Instant;
@@ -272,49 +271,6 @@ 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,11 +111,6 @@ 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/common/meta/src/test_util.rs
+++ b/src/common/meta/src/test_util.rs
@@ -15,7 +15,7 @@
 use std::sync::Arc;

 use api::region::RegionResponse;
-use api::v1::flow::{DirtyWindowRequest, FlowRequest, FlowResponse};
+use api::v1::flow::{FlowRequest, FlowResponse};
 use api::v1::region::{InsertRequests, RegionRequest};
 pub use common_base::AffectedRows;
 use common_query::request::QueryRequest;
@@ -67,14 +67,6 @@ pub trait MockFlownodeHandler: Sync + Send + Clone {
    ) -> Result<FlowResponse> {
        unimplemented!()
    }
-
-    async fn handle_mark_window_dirty(
-        &self,
-        _peer: &Peer,
-        _req: DirtyWindowRequest,
-    ) -> Result<FlowResponse> {
-        unimplemented!()
-    }
 }

 /// A mock struct implements [NodeManager] only implement the `datanode` method.
@@ -142,10 +134,6 @@ impl<T: MockFlownodeHandler> Flownode for MockNode<T> {
    async fn handle_inserts(&self, requests: InsertRequests) -> Result<FlowResponse> {
        self.handler.handle_inserts(&self.peer, requests).await
    }
-
-    async fn handle_mark_window_dirty(&self, req: DirtyWindowRequest) -> Result<FlowResponse> {
-        self.handler.handle_mark_window_dirty(&self.peer, req).await
-    }
 }

 #[async_trait::async_trait]
--- a/src/common/recordbatch/src/error.rs
+++ b/src/common/recordbatch/src/error.rs
@@ -173,7 +173,6 @@ pub enum Error {
        #[snafu(implicit)]
        location: Location,
    },
-
    #[snafu(display("Stream timeout"))]
    StreamTimeout {
        #[snafu(implicit)]
@@ -181,7 +180,6 @@ pub enum Error {
        #[snafu(source)]
        error: tokio::time::error::Elapsed,
    },
-
    #[snafu(display("RecordBatch slice index overflow: {visit_index} > {size}"))]
    RecordBatchSliceIndexOverflow {
        #[snafu(implicit)]
@@ -189,12 +187,6 @@ pub enum Error {
        size: usize,
        visit_index: usize,
    },
-
-    #[snafu(display("Stream has been cancelled"))]
-    StreamCancelled {
-        #[snafu(implicit)]
-        location: Location,
-    },
 }

 impl ErrorExt for Error {
@@ -229,8 +221,6 @@ impl ErrorExt for Error {
            }

            Error::StreamTimeout { .. } => StatusCode::Cancelled,
-
-            Error::StreamCancelled { .. } => StatusCode::Cancelled,
        }
    }

--- a/src/datatypes/src/value.rs
+++ b/src/datatypes/src/value.rs
@@ -326,14 +326,6 @@ 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/Cargo.toml
+++ b/src/flow/Cargo.toml
@@ -61,6 +61,7 @@ prost.workspace = true
 query.workspace = true
 rand.workspace = true
 serde.workspace = true
+serde_json.workspace = true
 servers.workspace = true
 session.workspace = true
 smallvec.workspace = true
--- a/src/flow/src/adapter.rs
+++ b/src/flow/src/adapter.rs
@@ -316,7 +316,7 @@ impl StreamingEngine {
            );

            METRIC_FLOW_ROWS
-                .with_label_values(&["out-streaming"])
+                .with_label_values(&["out"])
                .inc_by(total_rows as u64);

            let now = self.tick_manager.tick();
--- a/src/flow/src/adapter/flownode_impl.rs
+++ b/src/flow/src/adapter/flownode_impl.rs
@@ -18,7 +18,7 @@ use std::sync::atomic::AtomicBool;
 use std::sync::Arc;

 use api::v1::flow::{
-    flow_request, CreateRequest, DropRequest, FlowRequest, FlowResponse, FlushFlow,
+    flow_request, AdjustFlow, CreateRequest, DropRequest, FlowRequest, FlowResponse, FlushFlow,
 };
 use api::v1::region::InsertRequests;
 use catalog::CatalogManager;
@@ -31,8 +31,8 @@ use common_runtime::JoinHandle;
 use common_telemetry::{error, info, trace, warn};
 use datatypes::value::Value;
 use futures::TryStreamExt;
-use greptime_proto::v1::flow::DirtyWindowRequest;
 use itertools::Itertools;
+use serde::{Deserialize, Serialize};
 use session::context::QueryContextBuilder;
 use snafu::{ensure, IntoError, OptionExt, ResultExt};
 use store_api::storage::{RegionId, TableId};
@@ -47,7 +47,7 @@ use crate::error::{
    IllegalCheckTaskStateSnafu, InsertIntoFlowSnafu, InternalSnafu, JoinTaskSnafu, ListFlowsSnafu,
    NoAvailableFrontendSnafu, SyncCheckTaskSnafu, UnexpectedSnafu,
 };
-use crate::metrics::{METRIC_FLOW_ROWS, METRIC_FLOW_TASK_COUNT};
+use crate::metrics::METRIC_FLOW_TASK_COUNT;
 use crate::repr::{self, DiffRow};
 use crate::{Error, FlowId};

@@ -690,9 +690,6 @@ impl FlowEngine for FlowDualEngine {
        let mut to_stream_engine = Vec::with_capacity(request.requests.len());
        let mut to_batch_engine = request.requests;

-        let mut batching_row_cnt = 0;
-        let mut streaming_row_cnt = 0;
-
        {
            // not locking this, or recover flows will be starved when also handling flow inserts
            let src_table2flow = self.src_table2flow.read().await;
@@ -702,11 +699,9 @@ impl FlowEngine for FlowDualEngine {
                let is_in_stream = src_table2flow.in_stream(table_id);
                let is_in_batch = src_table2flow.in_batch(table_id);
                if is_in_stream {
-                    streaming_row_cnt += req.rows.as_ref().map(|rs| rs.rows.len()).unwrap_or(0);
                    to_stream_engine.push(req.clone());
                }
                if is_in_batch {
-                    batching_row_cnt += req.rows.as_ref().map(|rs| rs.rows.len()).unwrap_or(0);
                    return true;
                }
                if !is_in_batch && !is_in_stream {
@@ -719,14 +714,6 @@ impl FlowEngine for FlowDualEngine {
            // can't use drop due to https://github.com/rust-lang/rust/pull/128846
        }

-        METRIC_FLOW_ROWS
-            .with_label_values(&["in-streaming"])
-            .inc_by(streaming_row_cnt as u64);
-
-        METRIC_FLOW_ROWS
-            .with_label_values(&["in-batching"])
-            .inc_by(batching_row_cnt as u64);
-
        let streaming_engine = self.streaming_engine.clone();
        let stream_handler: JoinHandle<Result<(), Error>> =
            common_runtime::spawn_global(async move {
@@ -823,6 +810,25 @@ impl common_meta::node_manager::Flownode for FlowDualEngine {
                    ..Default::default()
                })
            }
+            Some(flow_request::Body::Adjust(AdjustFlow { flow_id, options })) => {
+                #[derive(Debug, Serialize, Deserialize)]
+                struct Options {
+                    min_run_interval_secs: u64,
+                    max_filter_num_per_query: usize,
+                }
+                let options: Options = serde_json::from_str(&options).with_context(|_| {
+                    common_meta::error::DeserializeFromJsonSnafu { input: options }
+                })?;
+                self.batching_engine
+                    .adjust_flow(
+                        flow_id.unwrap().id as u64,
+                        options.min_run_interval_secs,
+                        options.max_filter_num_per_query,
+                    )
+                    .await
+                    .map_err(to_meta_err(snafu::location!()))?;
+                Ok(Default::default())
+            }
            other => common_meta::error::InvalidFlowRequestBodySnafu { body: other }.fail(),
        }
    }
@@ -833,10 +839,6 @@ impl common_meta::node_manager::Flownode for FlowDualEngine {
            .map(|_| Default::default())
            .map_err(to_meta_err(snafu::location!()))
    }
-
-    async fn handle_mark_window_dirty(&self, _req: DirtyWindowRequest) -> MetaResult<FlowResponse> {
-        unreachable!()
-    }
 }

 /// return a function to convert `crate::error::Error` to `common_meta::error::Error`
@@ -859,97 +861,6 @@ fn to_meta_err(
    }
 }

-#[async_trait::async_trait]
-impl common_meta::node_manager::Flownode for StreamingEngine {
-    async fn handle(&self, request: FlowRequest) -> MetaResult<FlowResponse> {
-        let query_ctx = request
-            .header
-            .and_then(|h| h.query_context)
-            .map(|ctx| ctx.into());
-        match request.body {
-            Some(flow_request::Body::Create(CreateRequest {
-                flow_id: Some(task_id),
-                source_table_ids,
-                sink_table_name: Some(sink_table_name),
-                create_if_not_exists,
-                expire_after,
-                comment,
-                sql,
-                flow_options,
-                or_replace,
-            })) => {
-                let source_table_ids = source_table_ids.into_iter().map(|id| id.id).collect_vec();
-                let sink_table_name = [
-                    sink_table_name.catalog_name,
-                    sink_table_name.schema_name,
-                    sink_table_name.table_name,
-                ];
-                let expire_after = expire_after.map(|e| e.value);
-                let args = CreateFlowArgs {
-                    flow_id: task_id.id as u64,
-                    sink_table_name,
-                    source_table_ids,
-                    create_if_not_exists,
-                    or_replace,
-                    expire_after,
-                    comment: Some(comment),
-                    sql: sql.clone(),
-                    flow_options,
-                    query_ctx,
-                };
-                let ret = self
-                    .create_flow(args)
-                    .await
-                    .map_err(BoxedError::new)
-                    .with_context(|_| CreateFlowSnafu { sql: sql.clone() })
-                    .map_err(to_meta_err(snafu::location!()))?;
-                METRIC_FLOW_TASK_COUNT.inc();
-                Ok(FlowResponse {
-                    affected_flows: ret
-                        .map(|id| greptime_proto::v1::FlowId { id: id as u32 })
-                        .into_iter()
-                        .collect_vec(),
-                    ..Default::default()
-                })
-            }
-            Some(flow_request::Body::Drop(DropRequest {
-                flow_id: Some(flow_id),
-            })) => {
-                self.remove_flow(flow_id.id as u64)
-                    .await
-                    .map_err(to_meta_err(snafu::location!()))?;
-                METRIC_FLOW_TASK_COUNT.dec();
-                Ok(Default::default())
-            }
-            Some(flow_request::Body::Flush(FlushFlow {
-                flow_id: Some(flow_id),
-            })) => {
-                let row = self
-                    .flush_flow_inner(flow_id.id as u64)
-                    .await
-                    .map_err(to_meta_err(snafu::location!()))?;
-                Ok(FlowResponse {
-                    affected_flows: vec![flow_id],
-                    affected_rows: row as u64,
-                    ..Default::default()
-                })
-            }
-            other => common_meta::error::InvalidFlowRequestBodySnafu { body: other }.fail(),
-        }
-    }
-
-    async fn handle_inserts(&self, request: InsertRequests) -> MetaResult<FlowResponse> {
-        self.handle_inserts_inner(request)
-            .await
-            .map(|_| Default::default())
-            .map_err(to_meta_err(snafu::location!()))
-    }
-
-    async fn handle_mark_window_dirty(&self, _req: DirtyWindowRequest) -> MetaResult<FlowResponse> {
-        unreachable!()
-    }
-}
-
 impl FlowEngine for StreamingEngine {
    async fn create_flow(&self, args: CreateFlowArgs) -> Result<Option<FlowId>, Error> {
        self.create_flow_inner(args).await
--- a/src/flow/src/batching_mode/engine.rs
+++ b/src/flow/src/batching_mode/engine.rs
@@ -17,7 +17,6 @@
 use std::collections::{BTreeMap, HashMap};
 use std::sync::Arc;

-use api::v1::flow::{DirtyWindowRequests, FlowResponse};
 use catalog::CatalogManagerRef;
 use common_error::ext::BoxedError;
 use common_meta::ddl::create_flow::FlowType;
@@ -30,7 +29,8 @@ use common_telemetry::{debug, info};
 use common_time::TimeToLive;
 use query::QueryEngineRef;
 use snafu::{ensure, OptionExt, ResultExt};
-use store_api::storage::{RegionId, TableId};
+use store_api::storage::RegionId;
+use table::metadata::TableId;
 use tokio::sync::{oneshot, RwLock};

 use crate::batching_mode::frontend_client::FrontendClient;
@@ -42,7 +42,6 @@ use crate::error::{
    ExternalSnafu, FlowAlreadyExistSnafu, FlowNotFoundSnafu, TableNotFoundMetaSnafu,
    UnexpectedSnafu, UnsupportedSnafu,
 };
-use crate::metrics::METRIC_FLOW_BATCHING_ENGINE_BULK_MARK_TIME_WINDOW;
 use crate::{CreateFlowArgs, Error, FlowId, TableName};

 /// Batching mode Engine, responsible for driving all the batching mode tasks
@@ -78,116 +77,6 @@ impl BatchingEngine {
        }
    }

-    pub async fn handle_mark_dirty_time_window(
-        &self,
-        reqs: DirtyWindowRequests,
-    ) -> Result<FlowResponse, Error> {
-        let table_info_mgr = self.table_meta.table_info_manager();
-
-        let mut group_by_table_id: HashMap<u32, Vec<_>> = HashMap::new();
-        for r in reqs.requests {
-            let tid = TableId::from(r.table_id);
-            let entry = group_by_table_id.entry(tid).or_default();
-            entry.extend(r.timestamps);
-        }
-        let tids = group_by_table_id.keys().cloned().collect::<Vec<TableId>>();
-        let table_infos =
-            table_info_mgr
-                .batch_get(&tids)
-                .await
-                .with_context(|_| TableNotFoundMetaSnafu {
-                    msg: format!("Failed to get table info for table ids: {:?}", tids),
-                })?;
-
-        let group_by_table_name = group_by_table_id
-            .into_iter()
-            .filter_map(|(id, timestamps)| {
-                let table_name = table_infos.get(&id).map(|info| info.table_name());
-                let Some(table_name) = table_name else {
-                    warn!("Failed to get table infos for table id: {:?}", id);
-                    return None;
-                };
-                let table_name = [
-                    table_name.catalog_name,
-                    table_name.schema_name,
-                    table_name.table_name,
-                ];
-                let schema = &table_infos.get(&id).unwrap().table_info.meta.schema;
-                let time_index_unit = schema.column_schemas[schema.timestamp_index.unwrap()]
-                    .data_type
-                    .as_timestamp()
-                    .unwrap()
-                    .unit();
-                Some((table_name, (timestamps, time_index_unit)))
-            })
-            .collect::<HashMap<_, _>>();
-
-        let group_by_table_name = Arc::new(group_by_table_name);
-
-        let mut handles = Vec::new();
-        let tasks = self.tasks.read().await;
-
-        for (_flow_id, task) in tasks.iter() {
-            let src_table_names = &task.config.source_table_names;
-
-            if src_table_names
-                .iter()
-                .all(|name| !group_by_table_name.contains_key(name))
-            {
-                continue;
-            }
-
-            let group_by_table_name = group_by_table_name.clone();
-            let task = task.clone();
-
-            let handle: JoinHandle<Result<(), Error>> = tokio::spawn(async move {
-                let src_table_names = &task.config.source_table_names;
-                let mut all_dirty_windows = vec![];
-                for src_table_name in src_table_names {
-                    if let Some((timestamps, unit)) = group_by_table_name.get(src_table_name) {
-                        let Some(expr) = &task.config.time_window_expr else {
-                            continue;
-                        };
-                        for timestamp in timestamps {
-                            let align_start = expr
-                                .eval(common_time::Timestamp::new(*timestamp, *unit))?
-                                .0
-                                .context(UnexpectedSnafu {
-                                    reason: "Failed to eval start value",
-                                })?;
-                            all_dirty_windows.push(align_start);
-                        }
-                    }
-                }
-                let mut state = task.state.write().unwrap();
-                let flow_id_label = task.config.flow_id.to_string();
-                for timestamp in all_dirty_windows {
-                    state.dirty_time_windows.add_window(timestamp, None);
-                }
-
-                METRIC_FLOW_BATCHING_ENGINE_BULK_MARK_TIME_WINDOW
-                    .with_label_values(&[&flow_id_label])
-                    .set(state.dirty_time_windows.len() as f64);
-                Ok(())
-            });
-            handles.push(handle);
-        }
-        drop(tasks);
-        for handle in handles {
-            match handle.await {
-                Err(e) => {
-                    warn!("Failed to handle inserts: {e}");
-                }
-                Ok(Ok(())) => (),
-                Ok(Err(e)) => {
-                    warn!("Failed to handle inserts: {e}");
-                }
-            }
-        }
-
-        Ok(Default::default())
-    }
-
    pub async fn handle_inserts_inner(
        &self,
        request: api::v1::region::InsertRequests,
@@ -499,6 +388,20 @@ impl BatchingEngine {
    pub async fn flow_exist_inner(&self, flow_id: FlowId) -> bool {
        self.tasks.read().await.contains_key(&flow_id)
    }
+
+    pub async fn adjust_flow(
+        &self,
+        flow_id: FlowId,
+        min_run_interval_secs: u64,
+        max_filter_num_per_query: usize,
+    ) -> Result<(), Error> {
+        let task = self.tasks.read().await.get(&flow_id).cloned();
+        let task = task.with_context(|| FlowNotFoundSnafu { id: flow_id })?;
+        debug!("Adjusting flow {flow_id} with min_run_interval_secs={} and max_filter_num_per_query={}", min_run_interval_secs, max_filter_num_per_query);
+        task.adjust(min_run_interval_secs, max_filter_num_per_query);
+
+        Ok(())
+    }
 }

 impl FlowEngine for BatchingEngine {
--- a/src/flow/src/batching_mode/frontend_client.rs
+++ b/src/flow/src/batching_mode/frontend_client.rs
@@ -14,8 +14,9 @@

 //! Frontend client to run flow as batching task which is time-window-aware normal query triggered every tick set by user

-use std::sync::{Arc, Weak};
-use std::time::SystemTime;
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex, Weak};
+use std::time::{Duration, Instant, SystemTime};

 use api::v1::greptime_request::Request;
 use api::v1::CreateTableExpr;
@@ -26,20 +27,21 @@ use common_meta::cluster::{NodeInfo, NodeInfoKey, Role};
 use common_meta::peer::Peer;
 use common_meta::rpc::store::RangeRequest;
 use common_query::Output;
-use common_telemetry::warn;
+use common_telemetry::{debug, warn};
+use itertools::Itertools;
 use meta_client::client::MetaClient;
-use rand::rng;
-use rand::seq::SliceRandom;
 use servers::query_handler::grpc::GrpcQueryHandler;
 use session::context::{QueryContextBuilder, QueryContextRef};
 use snafu::{OptionExt, ResultExt};

+use crate::batching_mode::task::BatchingTask;
 use crate::batching_mode::{
    DEFAULT_BATCHING_ENGINE_QUERY_TIMEOUT, FRONTEND_ACTIVITY_TIMEOUT, GRPC_CONN_TIMEOUT,
    GRPC_MAX_RETRIES,
 };
 use crate::error::{ExternalSnafu, InvalidRequestSnafu, NoAvailableFrontendSnafu, UnexpectedSnafu};
-use crate::{Error, FlowAuthHeader};
+use crate::metrics::METRIC_FLOW_BATCHING_ENGINE_GUESS_FE_LOAD;
+use crate::{Error, FlowAuthHeader, FlowId};

 /// Just like [`GrpcQueryHandler`] but use BoxedError
 ///
@@ -74,6 +76,105 @@ impl<

 type HandlerMutable = Arc<std::sync::Mutex<Option<Weak<dyn GrpcQueryHandlerWithBoxedError>>>>;

+/// Statistics about running query on this frontend from flownode
+#[derive(Debug, Default, Clone)]
+struct FrontendStat {
+    /// The query for flow id has been running since this timestamp
+    since: HashMap<FlowId, Instant>,
+    /// The average query time for each flow id
+    /// This is used to calculate the average query time for each flow id
+    past_query_avg: HashMap<FlowId, (usize, Duration)>,
+}
+
+#[derive(Debug, Default, Clone)]
+pub struct FrontendStats {
+    /// The statistics for each flow id
+    stats: Arc<Mutex<HashMap<String, FrontendStat>>>,
+}
+
+impl FrontendStats {
+    pub fn observe(&self, frontend_addr: &str, flow_id: FlowId) -> FrontendStatsGuard {
+        let mut stats = self.stats.lock().expect("Failed to lock frontend stats");
+        let stat = stats.entry(frontend_addr.to_string()).or_default();
+        stat.since.insert(flow_id, Instant::now());
+
+        FrontendStatsGuard {
+            stats: self.stats.clone(),
+            frontend_addr: frontend_addr.to_string(),
+            cur: flow_id,
+        }
+    }
+
+    /// return frontend addrs sorted by load, from lightest to heaviest
+    /// The load is calculated as the total average query time for each flow id plus running query's total running time elapsed
+    pub fn sort_by_load(&self) -> Vec<String> {
+        let stats = self.stats.lock().expect("Failed to lock frontend stats");
+        let fe_load_factor = stats
+            .iter()
+            .map(|(node_addr, stat)| {
+                // total expected avg running time for all currently running queries
+                let total_expect_avg_run_time = stat
+                    .since
+                    .keys()
+                    .map(|f| {
+                        let (count, total_duration) =
+                            stat.past_query_avg.get(f).unwrap_or(&(0, Duration::ZERO));
+                        if *count == 0 {
+                            0.0
+                        } else {
+                            total_duration.as_secs_f64() / *count as f64
+                        }
+                    })
+                    .sum::<f64>();
+                let total_cur_running_time = stat
+                    .since
+                    .values()
+                    .map(|since| since.elapsed().as_secs_f64())
+                    .sum::<f64>();
+                (
+                    node_addr.to_string(),
+                    total_expect_avg_run_time + total_cur_running_time,
+                )
+            })
+            .sorted_by(|(_, load_a), (_, load_b)| {
+                load_a
+                    .partial_cmp(load_b)
+                    .unwrap_or(std::cmp::Ordering::Equal)
+            })
+            .collect::<Vec<_>>();
+        debug!("Frontend load factor: {:?}", fe_load_factor);
+        for (node_addr, load) in &fe_load_factor {
+            METRIC_FLOW_BATCHING_ENGINE_GUESS_FE_LOAD
+                .with_label_values(&[&node_addr.to_string()])
+                .observe(*load);
+        }
+        fe_load_factor
+            .into_iter()
+            .map(|(addr, _)| addr)
+            .collect::<Vec<_>>()
+    }
+}
+
+pub struct FrontendStatsGuard {
+    stats: Arc<Mutex<HashMap<String, FrontendStat>>>,
+    frontend_addr: String,
+    cur: FlowId,
+}
+
+impl Drop for FrontendStatsGuard {
+    fn drop(&mut self) {
+        let mut stats = self.stats.lock().expect("Failed to lock frontend stats");
+        if let Some(stat) = stats.get_mut(&self.frontend_addr) {
+            if let Some(since) = stat.since.remove(&self.cur) {
+                let elapsed = since.elapsed();
+                let (count, total_duration) = stat.past_query_avg.entry(self.cur).or_default();
+                *count += 1;
+                *total_duration += elapsed;
+            }
+        }
+    }
+}
+
 /// A simple frontend client able to execute sql using grpc protocol
 ///
 /// This is for computation-heavy query which need to offload computation to frontend, lifting the load from flownode
@@ -83,6 +184,7 @@ pub enum FrontendClient {
        meta_client: Arc<MetaClient>,
        chnl_mgr: ChannelManager,
        auth: Option<FlowAuthHeader>,
+        fe_stats: FrontendStats,
    },
    Standalone {
        /// for the sake of simplicity still use grpc even in standalone mode
@@ -114,6 +216,7 @@ impl FrontendClient {
                ChannelManager::with_config(cfg)
            },
            auth,
+            fe_stats: Default::default(),
        }
    }

@@ -183,7 +286,7 @@ impl FrontendClient {

    /// Get the frontend with recent enough(less than 1 minute from now) `last_activity_ts`
    /// and is able to process query
-    async fn get_random_active_frontend(
+    pub(crate) async fn get_random_active_frontend(
        &self,
        catalog: &str,
        schema: &str,
@@ -192,6 +295,7 @@ impl FrontendClient {
            meta_client: _,
            chnl_mgr,
            auth,
+            fe_stats,
        } = self
        else {
            return UnexpectedSnafu {
@@ -208,8 +312,21 @@ impl FrontendClient {
                .duration_since(SystemTime::UNIX_EPOCH)
                .unwrap()
                .as_millis() as i64;
-            // shuffle the frontends to avoid always pick the same one
-            frontends.shuffle(&mut rng());
+            let node_addrs_by_load = fe_stats.sort_by_load();
+            // index+1 to load order asc, so that the lightest node has load 1 and non-existent node has load 0
+            let addr2load = node_addrs_by_load
+                .iter()
+                .enumerate()
+                .map(|(i, id)| (id.clone(), i + 1))
+                .collect::<HashMap<_, _>>();
+            // sort frontends by load, from lightest to heaviest
+            frontends.sort_by(|(_, a), (_, b)| {
+                // if not even in stats, treat as 0 load since never been queried
+                let load_a = addr2load.get(&a.peer.addr).unwrap_or(&0);
+                let load_b = addr2load.get(&b.peer.addr).unwrap_or(&0);
+                load_a.cmp(load_b)
+            });
+            debug!("Frontend nodes sorted by load: {:?}", frontends);

            // found node with maximum last_activity_ts
            for (_, node_info) in frontends
@@ -257,6 +374,7 @@ impl FrontendClient {
        create: CreateTableExpr,
        catalog: &str,
        schema: &str,
+        task: Option<&BatchingTask>,
    ) -> Result<u32, Error> {
        self.handle(
            Request::Ddl(api::v1::DdlRequest {
@@ -264,7 +382,8 @@ impl FrontendClient {
            }),
            catalog,
            schema,
-            &mut None,
+            None,
+            task,
        )
        .await
    }
@@ -275,15 +394,31 @@ impl FrontendClient {
        req: api::v1::greptime_request::Request,
        catalog: &str,
        schema: &str,
-        peer_desc: &mut Option<PeerDesc>,
+        use_peer: Option<Peer>,
+        task: Option<&BatchingTask>,
    ) -> Result<u32, Error> {
        match self {
-            FrontendClient::Distributed { .. } => {
-                let db = self.get_random_active_frontend(catalog, schema).await?;
+            FrontendClient::Distributed {
+                fe_stats, chnl_mgr, ..
+            } => {
+                let db = if let Some(peer) = use_peer {
+                    DatabaseWithPeer::new(
+                        Database::new(
+                            catalog,
+                            schema,
+                            Client::with_manager_and_urls(
+                                chnl_mgr.clone(),
+                                vec![peer.addr.clone()],
+                            ),
+                        ),
+                        peer,
+                    )
+                } else {
+                    self.get_random_active_frontend(catalog, schema).await?
+                };

-                *peer_desc = Some(PeerDesc::Dist {
-                    peer: db.peer.clone(),
-                });
+                let flow_id = task.map(|t| t.config.flow_id).unwrap_or_default();
+                let _guard = fe_stats.observe(&db.peer.addr, flow_id);

                db.database
                    .handle_with_retry(req.clone(), GRPC_MAX_RETRIES)
--- a/src/flow/src/batching_mode/state.rs
+++ b/src/flow/src/batching_mode/state.rs
@@ -32,6 +32,7 @@ use crate::batching_mode::MIN_REFRESH_DURATION;
 use crate::error::{DatatypesSnafu, InternalSnafu, TimeSnafu, UnexpectedSnafu};
 use crate::metrics::{
    METRIC_FLOW_BATCHING_ENGINE_QUERY_TIME_RANGE, METRIC_FLOW_BATCHING_ENGINE_QUERY_WINDOW_CNT,
+    METRIC_FLOW_BATCHING_ENGINE_STALLED_QUERY_WINDOW_CNT,
 };
 use crate::{Error, FlowId};

@@ -52,6 +53,13 @@ pub struct TaskState {
    pub(crate) shutdown_rx: oneshot::Receiver<()>,
    /// Task handle
    pub(crate) task_handle: Option<tokio::task::JoinHandle<()>>,
+    /// Slow Query metrics update task handle
+    pub(crate) slow_query_metric_task: Option<tokio::task::JoinHandle<()>>,
+
+    /// min run interval in seconds
+    pub(crate) min_run_interval: Option<u64>,
+    /// max filter number per query
+    pub(crate) max_filter_num: Option<usize>,
 }
 impl TaskState {
    pub fn new(query_ctx: QueryContextRef, shutdown_rx: oneshot::Receiver<()>) -> Self {
@@ -63,6 +71,9 @@ impl TaskState {
            exec_state: ExecState::Idle,
            shutdown_rx,
            task_handle: None,
+            slow_query_metric_task: None,
+            min_run_interval: None,
+            max_filter_num: None,
        }
    }

@@ -87,20 +98,17 @@ impl TaskState {
    pub fn get_next_start_query_time(
        &self,
        flow_id: FlowId,
-        time_window_size: &Option<Duration>,
+        _time_window_size: &Option<Duration>,
        max_timeout: Option<Duration>,
    ) -> Instant {
-        let last_duration = max_timeout
+        let next_duration = max_timeout
            .unwrap_or(self.last_query_duration)
            .min(self.last_query_duration)
-            .max(MIN_REFRESH_DURATION);
-
-        let next_duration = time_window_size
-            .map(|t| {
-                let half = t / 2;
-                half.max(last_duration)
-            })
-            .unwrap_or(last_duration);
+            .max(
+                self.min_run_interval
+                    .map(Duration::from_secs)
+                    .unwrap_or(MIN_REFRESH_DURATION),
+            );

        // if have dirty time window, execute immediately to clean dirty time window
        if self.dirty_time_windows.windows.is_empty() {
@@ -156,11 +164,6 @@ impl DirtyTimeWindows {
        self.windows.clear();
    }

-    /// Number of dirty windows.
-    pub fn len(&self) -> usize {
-        self.windows.len()
-    }
-
    /// Generate all filter expressions consuming all time windows
    ///
    /// there is two limits:
@@ -211,47 +214,69 @@ impl DirtyTimeWindows {

        // get the first `window_cnt` time windows
        let max_time_range = window_size * window_cnt as i32;
-        let nth = {
-            let mut cur_time_range = chrono::Duration::zero();
-            let mut nth_key = None;
-            for (idx, (start, end)) in self.windows.iter().enumerate() {
-                // if time range is too long, stop
-                if cur_time_range > max_time_range {
-                    nth_key = Some(*start);
-                    break;
-                }

-                // if we have enough time windows, stop
-                if idx >= window_cnt {
-                    nth_key = Some(*start);
-                    break;
-                }
+        let mut to_be_query = BTreeMap::new();
+        let mut new_windows = self.windows.clone();
+        let mut cur_time_range = chrono::Duration::zero();
+        for (idx, (start, end)) in self.windows.iter().enumerate() {
+            let first_end = start
+                .add_duration(window_size.to_std().unwrap())
+                .context(TimeSnafu)?;
+            let end = end.unwrap_or(first_end);

-                if let Some(end) = end {
-                    if let Some(x) = end.sub(start) {
-                        cur_time_range += x;
-                    }
-                }
+            // if time range is too long, stop
+            if cur_time_range >= max_time_range {
+                break;
            }

-            nth_key
-        };
-        let first_nth = {
-            if let Some(nth) = nth {
-                let mut after = self.windows.split_off(&nth);
-                std::mem::swap(&mut self.windows, &mut after);
-
-                after
-            } else {
-                std::mem::take(&mut self.windows)
+            // if we have enough time windows, stop
+            if idx >= window_cnt {
+                break;
            }
-        };
+
+            if let Some(x) = end.sub(start) {
+                if cur_time_range + x <= max_time_range {
+                    to_be_query.insert(*start, Some(end));
+                    new_windows.remove(start);
+                    cur_time_range += x;
+                } else {
+                    // too large a window, split it
+                    // split at window_size * times
+                    let surplus = max_time_range - cur_time_range;
+                    let times = surplus.num_seconds() / window_size.num_seconds();
+
+                    let split_offset = window_size * times as i32;
+                    let split_at = start
+                        .add_duration(split_offset.to_std().unwrap())
+                        .context(TimeSnafu)?;
+                    to_be_query.insert(*start, Some(split_at));
+
+                    // remove the original window
+                    new_windows.remove(start);
+                    new_windows.insert(split_at, Some(end));
+                    cur_time_range += split_offset;
+                    break;
+                }
+            }
+        }
+
+        self.windows = new_windows;

        METRIC_FLOW_BATCHING_ENGINE_QUERY_WINDOW_CNT
-            .with_label_values(&[flow_id.to_string().as_str()])
-            .observe(first_nth.len() as f64);
+            .with_label_values(&[
+                flow_id.to_string().as_str(),
+                format!("{}", window_size).as_str(),
+            ])
+            .observe(to_be_query.len() as f64);

-        let full_time_range = first_nth
+        METRIC_FLOW_BATCHING_ENGINE_STALLED_QUERY_WINDOW_CNT
+            .with_label_values(&[
+                flow_id.to_string().as_str(),
+                format!("{}", window_size).as_str(),
+            ])
+            .observe(self.windows.len() as f64);
+
+        let full_time_range = to_be_query
            .iter()
            .fold(chrono::Duration::zero(), |acc, (start, end)| {
                if let Some(end) = end {
@@ -262,11 +287,14 @@ impl DirtyTimeWindows {
            })
            .num_seconds() as f64;
        METRIC_FLOW_BATCHING_ENGINE_QUERY_TIME_RANGE
-            .with_label_values(&[flow_id.to_string().as_str()])
+            .with_label_values(&[
+                flow_id.to_string().as_str(),
+                format!("{}", window_size).as_str(),
+            ])
            .observe(full_time_range);

        let mut expr_lst = vec![];
-        for (start, end) in first_nth.into_iter() {
+        for (start, end) in to_be_query.into_iter() {
            // align using time window exprs
            let (start, end) = if let Some(ctx) = task_ctx {
                let Some(time_window_expr) = &ctx.config.time_window_expr else {
@@ -500,6 +528,64 @@ mod test {
                    "((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts < CAST('1970-01-01 00:00:21' AS TIMESTAMP)))",
                )
            ),
+            // split range
+            (
+                Vec::from_iter((0..20).map(|i|Timestamp::new_second(i*3)).chain(std::iter::once(
+                    Timestamp::new_second(60 + 3 * (DirtyTimeWindows::MERGE_DIST as i64 + 1)),
+                ))),
+                (chrono::Duration::seconds(3), None),
+                BTreeMap::from([
+                (
+                    Timestamp::new_second(0),
+                    Some(Timestamp::new_second(
+                        60
+                    )),
+                ),
+                (
+                    Timestamp::new_second(60 + 3 * (DirtyTimeWindows::MERGE_DIST as i64 + 1)),
+                    Some(Timestamp::new_second(
+                        60 + 3 * (DirtyTimeWindows::MERGE_DIST as i64 + 1) + 3
+                    )),
+                )]),
+                Some(
+                    "((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts < CAST('1970-01-01 00:01:00' AS TIMESTAMP)))",
+                )
+            ),
+            // split 2 min into 1 min
+            (
+                Vec::from_iter((0..40).map(|i|Timestamp::new_second(i*3))),
+                (chrono::Duration::seconds(3), None),
+                BTreeMap::from([
+                (
+                    Timestamp::new_second(0),
+                    Some(Timestamp::new_second(
+                        40 * 3
+                    )),
+                )]),
+                Some(
+                    "((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts < CAST('1970-01-01 00:01:00' AS TIMESTAMP)))",
+                )
+            ),
+            // split 3s + 1min into 3s + 57s
+            (
+                Vec::from_iter(std::iter::once(Timestamp::new_second(0)).chain((0..40).map(|i|Timestamp::new_second(20+i*3)))),
+                (chrono::Duration::seconds(3), None),
+                BTreeMap::from([
+                (
+                    Timestamp::new_second(0),
+                    Some(Timestamp::new_second(
+                        3
+                    )),
+                ),(
+                    Timestamp::new_second(20),
+                    Some(Timestamp::new_second(
+                        140
+                    )),
+                )]),
+                Some(
+                    "(((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts < CAST('1970-01-01 00:00:03' AS TIMESTAMP))) OR ((ts >= CAST('1970-01-01 00:00:20' AS TIMESTAMP)) AND (ts < CAST('1970-01-01 00:01:17' AS TIMESTAMP))))",
+                )
+            ),
            // expired
            (
                vec![
@@ -516,6 +602,8 @@ mod test {
                None
            ),
        ];
+        // let len = testcases.len();
+        // let testcases = testcases[(len - 2)..(len - 1)].to_vec();
        for (lower_bounds, (window_size, expire_lower_bound), expected, expected_filter_expr) in
            testcases
        {
--- a/src/flow/src/batching_mode/task.rs
+++ b/src/flow/src/batching_mode/task.rs
@@ -61,9 +61,8 @@ use crate::error::{
    SubstraitEncodeLogicalPlanSnafu, UnexpectedSnafu,
 };
 use crate::metrics::{
-    METRIC_FLOW_BATCHING_ENGINE_ERROR_CNT, METRIC_FLOW_BATCHING_ENGINE_QUERY_TIME,
-    METRIC_FLOW_BATCHING_ENGINE_SLOW_QUERY, METRIC_FLOW_BATCHING_ENGINE_START_QUERY_CNT,
-    METRIC_FLOW_ROWS,
+    METRIC_FLOW_BATCHING_ENGINE_QUERY_TIME, METRIC_FLOW_BATCHING_ENGINE_REAL_TIME_SLOW_QUERY_CNT,
+    METRIC_FLOW_BATCHING_ENGINE_SLOW_QUERY,
 };
 use crate::{Error, FlowId};

@@ -83,6 +82,14 @@ pub struct TaskConfig {
    query_type: QueryType,
 }

+impl TaskConfig {
+    pub fn time_window_size(&self) -> Option<Duration> {
+        self.time_window_expr
+            .as_ref()
+            .and_then(|expr| *expr.time_window_size())
+    }
+}
+
 fn determine_query_type(query: &str, query_ctx: &QueryContextRef) -> Result<QueryType, Error> {
    let stmts =
        ParserContext::create_with_dialect(query, query_ctx.sql_dialect(), ParseOptions::default())
@@ -146,6 +153,12 @@ impl BatchingTask {
        })
    }

+    pub fn adjust(&self, min_run_interval_secs: u64, max_filter_num_per_query: usize) {
+        let mut state = self.state.write().unwrap();
+        state.min_run_interval = Some(min_run_interval_secs);
+        state.max_filter_num = Some(max_filter_num_per_query);
+    }
+
    /// mark time window range (now - expire_after, now) as dirty (or (0, now) if expire_after not set)
    ///
    /// useful for flush_flow to flush dirty time windows range
@@ -282,7 +295,7 @@ impl BatchingTask {
        let catalog = &self.config.sink_table_name[0];
        let schema = &self.config.sink_table_name[1];
        frontend_client
-            .create(expr.clone(), catalog, schema)
+            .create(expr.clone(), catalog, schema, Some(self))
            .await?;
        Ok(())
    }
@@ -330,11 +343,53 @@ impl BatchingTask {
            })?;

        let plan = expanded_plan;
-        let mut peer_desc = None;
+
+        let db = frontend_client
+            .get_random_active_frontend(catalog, schema)
+            .await?;
+        let peer_desc = db.peer.clone();
+
+        let (tx, mut rx) = oneshot::channel();
+        let peer_inner = peer_desc.clone();
+        let window_size_pretty = format!(
+            "{}s",
+            self.config.time_window_size().unwrap_or_default().as_secs()
+        );
+        let inner_window_size_pretty = window_size_pretty.clone();
+        let flow_id = self.config.flow_id;
+        let slow_query_metric_task = tokio::task::spawn(async move {
+            tokio::time::sleep(SLOW_QUERY_THRESHOLD).await;
+            METRIC_FLOW_BATCHING_ENGINE_REAL_TIME_SLOW_QUERY_CNT
+                .with_label_values(&[
+                    flow_id.to_string().as_str(),
+                    &peer_inner.to_string(),
+                    inner_window_size_pretty.as_str(),
+                ])
+                .add(1.0);
+            while rx.try_recv() == Err(TryRecvError::Empty) {
+                // sleep for a while before next update
+                tokio::time::sleep(MIN_REFRESH_DURATION).await;
+            }
+            METRIC_FLOW_BATCHING_ENGINE_REAL_TIME_SLOW_QUERY_CNT
+                .with_label_values(&[
+                    flow_id.to_string().as_str(),
+                    &peer_inner.to_string(),
+                    inner_window_size_pretty.as_str(),
+                ])
+                .sub(1.0);
+        });
+        self.state.write().unwrap().slow_query_metric_task = Some(slow_query_metric_task);

        let res = {
            let _timer = METRIC_FLOW_BATCHING_ENGINE_QUERY_TIME
-                .with_label_values(&[flow_id.to_string().as_str()])
+                .with_label_values(&[
+                    flow_id.to_string().as_str(),
+                    format!(
+                        "{}s",
+                        self.config.time_window_size().unwrap_or_default().as_secs()
+                    )
+                    .as_str(),
+                ])
                .start_timer();

            // hack and special handling the insert logical plan
@@ -363,19 +418,18 @@ impl BatchingTask {
            };

            frontend_client
-                .handle(req, catalog, schema, &mut peer_desc)
+                .handle(req, catalog, schema, Some(db.peer), Some(self))
                .await
        };

+        // signaling the slow query metric task to stop
+        let _ = tx.send(());
        let elapsed = instant.elapsed();
        if let Ok(affected_rows) = &res {
            debug!(
                "Flow {flow_id} executed, affected_rows: {affected_rows:?}, elapsed: {:?}",
                elapsed
            );
-            METRIC_FLOW_ROWS
-                .with_label_values(&[format!("{}-out-batching", flow_id).as_str()])
-                .inc_by(*affected_rows as _);
        } else if let Err(err) = &res {
            warn!(
                "Failed to execute Flow {flow_id} on frontend {:?}, result: {err:?}, elapsed: {:?} with query: {}",
@@ -392,7 +446,12 @@ impl BatchingTask {
            METRIC_FLOW_BATCHING_ENGINE_SLOW_QUERY
                .with_label_values(&[
                    flow_id.to_string().as_str(),
-                    &peer_desc.unwrap_or_default().to_string(),
+                    &peer_desc.to_string(),
+                    format!(
+                        "{}s",
+                        self.config.time_window_size().unwrap_or_default().as_secs()
+                    )
+                    .as_str(),
                ])
                .observe(elapsed.as_secs_f64());
        }
@@ -415,7 +474,6 @@ impl BatchingTask {
        engine: QueryEngineRef,
        frontend_client: Arc<FrontendClient>,
    ) {
-        let flow_id_str = self.config.flow_id.to_string();
        loop {
            // first check if shutdown signal is received
            // if so, break the loop
@@ -433,9 +491,6 @@ impl BatchingTask {
                    Err(TryRecvError::Empty) => (),
                }
            }
-            METRIC_FLOW_BATCHING_ENGINE_START_QUERY_CNT
-                .with_label_values(&[&flow_id_str])
-                .inc();

            let new_query = match self.gen_insert_plan(&engine).await {
                Ok(new_query) => new_query,
@@ -482,9 +537,6 @@ impl BatchingTask {
                }
                // TODO(discord9): this error should have better place to go, but for now just print error, also more context is needed
                Err(err) => {
-                    METRIC_FLOW_BATCHING_ENGINE_ERROR_CNT
-                        .with_label_values(&[&flow_id_str])
-                        .inc();
                    match new_query {
                        Some(query) => {
                            common_telemetry::error!(err; "Failed to execute query for flow={} with query: {query}", self.config.flow_id)
@@ -592,19 +644,20 @@ impl BatchingTask {
                ),
            })?;

-        let expr = self
-            .state
-            .write()
-            .unwrap()
-            .dirty_time_windows
-            .gen_filter_exprs(
+        let expr = {
+            let mut state = self.state.write().unwrap();
+            let max_window_cnt = state
+                .max_filter_num
+                .unwrap_or(DirtyTimeWindows::MAX_FILTER_NUM);
+            state.dirty_time_windows.gen_filter_exprs(
                &col_name,
                Some(l),
                window_size,
-                DirtyTimeWindows::MAX_FILTER_NUM,
+                max_window_cnt,
                self.config.flow_id,
                Some(self),
-            )?;
+            )?
+        };

        debug!(
            "Flow id={:?}, Generated filter expr: {:?}",
--- a/src/flow/src/metrics.rs
+++ b/src/flow/src/metrics.rs
@@ -31,22 +31,37 @@ lazy_static! {
    pub static ref METRIC_FLOW_BATCHING_ENGINE_QUERY_TIME: HistogramVec = register_histogram_vec!(
        "greptime_flow_batching_engine_query_time_secs",
        "flow batching engine query time(seconds)",
-        &["flow_id"],
+        &["flow_id", "time_window_granularity"],
        vec![0.0, 5., 10., 20., 40., 80., 160., 320., 640.,]
    )
    .unwrap();
    pub static ref METRIC_FLOW_BATCHING_ENGINE_SLOW_QUERY: HistogramVec = register_histogram_vec!(
        "greptime_flow_batching_engine_slow_query_secs",
-        "flow batching engine slow query(seconds)",
-        &["flow_id", "peer"],
+        "flow batching engine slow query(seconds), updated after query finished",
+        &["flow_id", "peer", "time_window_granularity"],
        vec![60., 2. * 60., 3. * 60., 5. * 60., 10. * 60.]
    )
    .unwrap();
+    pub static ref METRIC_FLOW_BATCHING_ENGINE_REAL_TIME_SLOW_QUERY_CNT: GaugeVec =
+        register_gauge_vec!(
+            "greptime_flow_batching_engine_real_time_slow_query_number",
+            "flow batching engine real time slow query number, updated in real time",
+            &["flow_id", "peer", "time_window_granularity"],
+        )
+        .unwrap();
+    pub static ref METRIC_FLOW_BATCHING_ENGINE_STALLED_QUERY_WINDOW_CNT: HistogramVec =
+        register_histogram_vec!(
+            "greptime_flow_batching_engine_stalled_query_window_cnt",
+            "flow batching engine stalled query time window count",
+            &["flow_id", "time_window_granularity"],
+            vec![0.0, 5., 10., 20., 40.]
+        )
+        .unwrap();
    pub static ref METRIC_FLOW_BATCHING_ENGINE_QUERY_WINDOW_CNT: HistogramVec =
        register_histogram_vec!(
            "greptime_flow_batching_engine_query_window_cnt",
            "flow batching engine query time window count",
-            &["flow_id"],
+            &["flow_id", "time_window_granularity"],
            vec![0.0, 5., 10., 20., 40.]
        )
        .unwrap();
@@ -54,36 +69,23 @@ lazy_static! {
        register_histogram_vec!(
            "greptime_flow_batching_engine_query_time_range_secs",
            "flow batching engine query time range(seconds)",
-            &["flow_id"],
+            &["flow_id", "time_window_granularity"],
            vec![60., 4. * 60., 16. * 60., 64. * 60., 256. * 60.]
        )
        .unwrap();
-    pub static ref METRIC_FLOW_BATCHING_ENGINE_BULK_MARK_TIME_WINDOW: GaugeVec =
-        register_gauge_vec!(
-            "greptime_flow_batching_engine_bulk_mark_time_window",
-            "flow batching engine query time window count marked by bulk inserts",
-            &["flow_id"],
-        )
-        .unwrap();
-    pub static ref METRIC_FLOW_BATCHING_ENGINE_START_QUERY_CNT: IntCounterVec =
-        register_int_counter_vec!(
-            "greptime_flow_batching_start_query_count",
-            "flow batching engine started query count",
-            &["flow_id"],
-        )
-        .unwrap();
-    pub static ref METRIC_FLOW_BATCHING_ENGINE_ERROR_CNT: IntCounterVec =
-        register_int_counter_vec!(
-            "greptime_flow_batching_error_count",
-            "flow batching engine error count per flow id",
-            &["flow_id"],
+    pub static ref METRIC_FLOW_BATCHING_ENGINE_GUESS_FE_LOAD: HistogramVec =
+        register_histogram_vec!(
+            "greptime_flow_batching_engine_guess_fe_load",
+            "flow batching engine guessed frontend load",
+            &["fe_addr"],
+            vec![60., 4. * 60., 16. * 60., 64. * 60., 256. * 60.]
        )
        .unwrap();
    pub static ref METRIC_FLOW_RUN_INTERVAL_MS: IntGauge =
        register_int_gauge!("greptime_flow_run_interval_ms", "flow run interval in ms").unwrap();
    pub static ref METRIC_FLOW_ROWS: IntCounterVec = register_int_counter_vec!(
        "greptime_flow_processed_rows",
-        "Count of rows flowing through the system.",
+        "Count of rows flowing through the system",
        &["direction"]
    )
    .unwrap();
--- a/src/flow/src/server.rs
+++ b/src/flow/src/server.rs
@@ -17,7 +17,6 @@
 use std::net::SocketAddr;
 use std::sync::Arc;

-use api::v1::flow::DirtyWindowRequests;
 use api::v1::{RowDeleteRequests, RowInsertRequests};
 use cache::{TABLE_FLOWNODE_SET_CACHE_NAME, TABLE_ROUTE_CACHE_NAME};
 use catalog::CatalogManagerRef;
@@ -137,18 +136,6 @@ impl flow_server::Flow for FlowService {
            .map(Response::new)
            .map_err(to_status_with_last_err)
    }
-
-    async fn handle_mark_dirty_time_window(
-        &self,
-        reqs: Request<DirtyWindowRequests>,
-    ) -> Result<Response<FlowResponse>, Status> {
-        self.dual_engine
-            .batching_engine()
-            .handle_mark_dirty_time_window(reqs.into_inner())
-            .await
-            .map(Response::new)
-            .map_err(to_status_with_last_err)
-    }
 }

 #[derive(Clone)]
@@ -591,7 +578,6 @@ impl FrontendInvoker {
            layered_cache_registry.clone(),
            inserter.clone(),
            table_route_cache,
-            None,
        ));

        let invoker = FrontendInvoker::new(inserter, deleter, statement_executor);
--- a/src/frontend/Cargo.toml
+++ b/src/frontend/Cargo.toml
@@ -25,7 +25,6 @@ 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
@@ -70,7 +69,6 @@ store-api.workspace = true
 substrait.workspace = true
 table.workspace = true
 tokio.workspace = true
-tokio-util.workspace = true
 toml.workspace = true
 tonic.workspace = true

--- a/src/frontend/src/error.rs
+++ b/src/frontend/src/error.rs
@@ -357,12 +357,6 @@ pub enum Error {
        #[snafu(implicit)]
        location: Location,
    },
-
-    #[snafu(display("Query has been cancelled"))]
-    Cancelled {
-        #[snafu(implicit)]
-        location: Location,
-    },
 }

 pub type Result<T> = std::result::Result<T, Error>;
@@ -441,8 +435,6 @@ impl ErrorExt for Error {
            Error::InFlightWriteBytesExceeded { .. } => StatusCode::RateLimited,

            Error::DataFusion { error, .. } => datafusion_status_code::<Self>(error, None),
-
-            Error::Cancelled { .. } => StatusCode::Cancelled,
        }
    }

--- a/src/frontend/src/instance.rs
+++ b/src/frontend/src/instance.rs
@@ -30,10 +30,8 @@ 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::cancellation::CancellableFuture;
 use common_base::Plugins;
 use common_config::KvBackendConfig;
 use common_error::ext::{BoxedError, ErrorExt};
@@ -82,7 +80,6 @@ use crate::error::{
 };
 use crate::limiter::LimiterRef;
 use crate::slow_query_recorder::SlowQueryRecorder;
-use crate::stream_wrapper::CancellableStreamWrapper;

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

 impl Instance {
@@ -157,10 +153,6 @@ 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>> {
@@ -180,79 +172,50 @@ impl Instance {
            None
        };

-        let ticket = self.process_manager.register_query(
-            query_ctx.current_catalog().to_string(),
-            vec![query_ctx.current_schema()],
-            stmt.to_string(),
-            query_ctx.conn_info().to_string(),
-            None,
-        );
-
-        let query_fut = async {
-            match stmt {
-                Statement::Query(_) | Statement::Explain(_) | Statement::Delete(_) => {
-                    // TODO: remove this when format is supported in datafusion
-                    if let Statement::Explain(explain) = &stmt {
-                        if let Some(format) = explain.format() {
-                            query_ctx.set_explain_format(format.to_string());
-                        }
+        let output = match stmt {
+            Statement::Query(_) | Statement::Explain(_) | Statement::Delete(_) => {
+                // TODO: remove this when format is supported in datafusion
+                if let Statement::Explain(explain) = &stmt {
+                    if let Some(format) = explain.format() {
+                        query_ctx.set_explain_format(format.to_string());
                    }
-
-                    let stmt = QueryStatement::Sql(stmt);
-                    let plan = self
-                        .statement_executor
-                        .plan(&stmt, query_ctx.clone())
-                        .await?;
-
-                    let QueryStatement::Sql(stmt) = stmt else {
-                        unreachable!()
-                    };
-                    query_interceptor.pre_execute(&stmt, Some(&plan), query_ctx.clone())?;
-                    self.statement_executor
-                        .exec_plan(plan, query_ctx)
-                        .await
-                        .context(TableOperationSnafu)
                }
-                Statement::Tql(tql) => {
-                    let plan = self
-                        .statement_executor
-                        .plan_tql(tql.clone(), &query_ctx)
-                        .await?;

-                    query_interceptor.pre_execute(
-                        &Statement::Tql(tql),
-                        Some(&plan),
-                        query_ctx.clone(),
-                    )?;
-                    self.statement_executor
-                        .exec_plan(plan, query_ctx)
-                        .await
-                        .context(TableOperationSnafu)
-                }
-                _ => {
-                    query_interceptor.pre_execute(&stmt, None, query_ctx.clone())?;
-                    self.statement_executor
-                        .execute_sql(stmt, query_ctx)
-                        .await
-                        .context(TableOperationSnafu)
-                }
+                let stmt = QueryStatement::Sql(stmt);
+                let plan = self
+                    .statement_executor
+                    .plan(&stmt, query_ctx.clone())
+                    .await?;
+
+                let QueryStatement::Sql(stmt) = stmt else {
+                    unreachable!()
+                };
+                query_interceptor.pre_execute(&stmt, Some(&plan), query_ctx.clone())?;
+
+                self.statement_executor.exec_plan(plan, query_ctx).await
+            }
+            Statement::Tql(tql) => {
+                let plan = self
+                    .statement_executor
+                    .plan_tql(tql.clone(), &query_ctx)
+                    .await?;
+
+                query_interceptor.pre_execute(
+                    &Statement::Tql(tql),
+                    Some(&plan),
+                    query_ctx.clone(),
+                )?;
+
+                self.statement_executor.exec_plan(plan, query_ctx).await
+            }
+            _ => {
+                query_interceptor.pre_execute(&stmt, None, query_ctx.clone())?;
+
+                self.statement_executor.execute_sql(stmt, query_ctx).await
            }
        };

-        CancellableFuture::new(query_fut, ticket.cancellation_handle.clone())
-            .await
-            .map_err(|_| error::CancelledSnafu.build())?
-            .map(|output| {
-                let Output { meta, data } = output;
-
-                let data = match data {
-                    OutputData::Stream(stream) => {
-                        OutputData::Stream(Box::pin(CancellableStreamWrapper::new(stream, ticket)))
-                    }
-                    other => other,
-                };
-                Output { data, meta }
-            })
+        output.context(TableOperationSnafu)
    }
 }

@@ -614,8 +577,6 @@ pub fn check_permission(
        }
        // cursor operations are always allowed once it's created
        Statement::FetchCursor(_) | Statement::CloseCursor(_) => {}
-        // User can only kill process in their own catalog.
-        Statement::Kill(_) => {}
    }
    Ok(())
 }
--- a/src/frontend/src/instance/builder.rs
+++ b/src/frontend/src/instance/builder.rs
@@ -15,7 +15,6 @@
 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};
@@ -55,11 +54,9 @@ 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,
@@ -67,7 +64,6 @@ impl FrontendBuilder {
        catalog_manager: CatalogManagerRef,
        node_manager: NodeManagerRef,
        procedure_executor: ProcedureExecutorRef,
-        process_manager: ProcessManagerRef,
    ) -> Self {
        Self {
            options,
@@ -78,7 +74,6 @@ impl FrontendBuilder {
            node_manager,
            plugins: None,
            procedure_executor,
-            process_manager,
        }
    }

@@ -100,7 +95,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()
@@ -180,7 +175,6 @@ impl FrontendBuilder {
            local_cache_invalidator,
            inserter.clone(),
            table_route_cache,
-            Some(process_manager.clone()),
        ));

        let pipeline_operator = Arc::new(PipelineOperator::new(
@@ -222,7 +216,6 @@ impl FrontendBuilder {
            table_metadata_manager: Arc::new(TableMetadataManager::new(kv_backend)),
            slow_query_recorder,
            limiter,
-            process_manager,
        })
    }
 }
--- a/src/frontend/src/instance/grpc.rs
+++ b/src/frontend/src/instance/grpc.rs
@@ -35,8 +35,8 @@ use servers::query_handler::grpc::GrpcQueryHandler;
 use servers::query_handler::sql::SqlQueryHandler;
 use session::context::QueryContextRef;
 use snafu::{ensure, OptionExt, ResultExt};
+use table::metadata::TableId;
 use table::table_name::TableName;
-use table::TableRef;

 use crate::error::{
    CatalogSnafu, DataFusionSnafu, Error, InFlightWriteBytesExceededSnafu,
@@ -235,33 +235,34 @@ impl GrpcQueryHandler for Instance {

    async fn put_record_batch(
        &self,
-        table_name: &TableName,
-        table_ref: &mut Option<TableRef>,
+        table: &TableName,
+        table_id: &mut Option<TableId>,
        decoder: &mut FlightDecoder,
        data: FlightData,
    ) -> Result<AffectedRows> {
-        let table = if let Some(table) = table_ref {
-            table.clone()
+        let table_id = if let Some(table_id) = table_id {
+            *table_id
        } else {
            let table = self
                .catalog_manager()
                .table(
-                    &table_name.catalog_name,
-                    &table_name.schema_name,
-                    &table_name.table_name,
+                    &table.catalog_name,
+                    &table.schema_name,
+                    &table.table_name,
                    None,
                )
                .await
                .context(CatalogSnafu)?
                .with_context(|| TableNotFoundSnafu {
-                    table_name: table_name.to_string(),
+                    table_name: table.to_string(),
                })?;
-            *table_ref = Some(table.clone());
-            table
+            let id = table.table_info().table_id();
+            *table_id = Some(id);
+            id
        };

        self.inserter
-            .handle_bulk_insert(table, decoder, data)
+            .handle_bulk_insert(table_id, decoder, data)
            .await
            .context(TableOperationSnafu)
    }
--- a/src/frontend/src/lib.rs
+++ b/src/frontend/src/lib.rs
@@ -23,4 +23,3 @@ 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,7 +20,6 @@ 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;
@@ -158,14 +157,11 @@ where
            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
@@ -1,367 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::pin::Pin;
-use std::task::{Context, Poll};
-
-use catalog::process_manager::Ticket;
-use common_recordbatch::adapter::RecordBatchMetrics;
-use common_recordbatch::{OrderOption, RecordBatch, RecordBatchStream, SendableRecordBatchStream};
-use datatypes::schema::SchemaRef;
-use futures::Stream;
-
-pub struct CancellableStreamWrapper {
-    inner: SendableRecordBatchStream,
-    ticket: Ticket,
-}
-
-impl Unpin for CancellableStreamWrapper {}
-
-impl CancellableStreamWrapper {
-    pub fn new(stream: SendableRecordBatchStream, ticket: Ticket) -> Self {
-        Self {
-            inner: stream,
-            ticket,
-        }
-    }
-}
-
-impl Stream for CancellableStreamWrapper {
-    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;
-        if this.ticket.cancellation_handle.is_cancelled() {
-            return Poll::Ready(Some(common_recordbatch::error::StreamCancelledSnafu.fail()));
-        }
-
-        if let Poll::Ready(res) = Pin::new(&mut this.inner).poll_next(cx) {
-            return Poll::Ready(res);
-        }
-
-        // on pending, register cancellation waker.
-        this.ticket.cancellation_handle.waker().register(cx.waker());
-        // check if canceled again.
-        if this.ticket.cancellation_handle.is_cancelled() {
-            return Poll::Ready(Some(common_recordbatch::error::StreamCancelledSnafu.fail()));
-        }
-        Poll::Pending
-    }
-}
-
-impl RecordBatchStream for CancellableStreamWrapper {
-    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()
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::pin::Pin;
-    use std::sync::Arc;
-    use std::task::{Context, Poll};
-    use std::time::Duration;
-
-    use catalog::process_manager::ProcessManager;
-    use common_recordbatch::adapter::RecordBatchMetrics;
-    use common_recordbatch::{OrderOption, RecordBatch, RecordBatchStream};
-    use datatypes::data_type::ConcreteDataType;
-    use datatypes::prelude::VectorRef;
-    use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
-    use datatypes::vectors::Int32Vector;
-    use futures::{Stream, StreamExt};
-    use tokio::time::{sleep, timeout};
-
-    use super::CancellableStreamWrapper;
-
-    // Mock stream for testing
-    struct MockRecordBatchStream {
-        schema: SchemaRef,
-        batches: Vec<common_recordbatch::error::Result<RecordBatch>>,
-        current: usize,
-        delay: Option<Duration>,
-    }
-
-    impl MockRecordBatchStream {
-        fn new(batches: Vec<common_recordbatch::error::Result<RecordBatch>>) -> Self {
-            let schema = Arc::new(Schema::new(vec![ColumnSchema::new(
-                "test_col",
-                ConcreteDataType::int32_datatype(),
-                false,
-            )]));
-
-            Self {
-                schema,
-                batches,
-                current: 0,
-                delay: None,
-            }
-        }
-
-        fn with_delay(mut self, delay: Duration) -> Self {
-            self.delay = Some(delay);
-            self
-        }
-    }
-
-    impl Stream for MockRecordBatchStream {
-        type Item = common_recordbatch::error::Result<RecordBatch>;
-
-        fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-            if let Some(delay) = self.delay {
-                // Simulate async delay
-                let waker = cx.waker().clone();
-                let delay_clone = delay;
-                tokio::spawn(async move {
-                    sleep(delay_clone).await;
-                    waker.wake();
-                });
-                self.delay = None; // Only delay once
-                return Poll::Pending;
-            }
-
-            if self.current >= self.batches.len() {
-                return Poll::Ready(None);
-            }
-
-            let batch = self.batches[self.current].as_ref().unwrap().clone();
-            self.current += 1;
-            Poll::Ready(Some(Ok(batch)))
-        }
-    }
-
-    impl RecordBatchStream for MockRecordBatchStream {
-        fn schema(&self) -> SchemaRef {
-            self.schema.clone()
-        }
-
-        fn output_ordering(&self) -> Option<&[OrderOption]> {
-            None
-        }
-
-        fn metrics(&self) -> Option<RecordBatchMetrics> {
-            None
-        }
-    }
-
-    fn create_test_batch() -> RecordBatch {
-        let schema = Arc::new(Schema::new(vec![ColumnSchema::new(
-            "test_col",
-            ConcreteDataType::int32_datatype(),
-            false,
-        )]));
-        RecordBatch::new(
-            schema,
-            vec![Arc::new(Int32Vector::from_values(0..3)) as VectorRef],
-        )
-        .unwrap()
-    }
-
-    #[tokio::test]
-    async fn test_stream_completes_normally() {
-        let batch = create_test_batch();
-        let mock_stream = MockRecordBatchStream::new(vec![Ok(batch.clone())]);
-        let process_manager = Arc::new(ProcessManager::new("".to_string(), None));
-        let ticket = process_manager.register_query(
-            "catalog".to_string(),
-            vec![],
-            "query".to_string(),
-            "client".to_string(),
-            None,
-        );
-
-        let mut cancellable_stream = CancellableStreamWrapper::new(Box::pin(mock_stream), ticket);
-
-        let result = cancellable_stream.next().await;
-        assert!(result.is_some());
-        assert!(result.unwrap().is_ok());
-
-        let end_result = cancellable_stream.next().await;
-        assert!(end_result.is_none());
-    }
-
-    #[tokio::test]
-    async fn test_stream_cancelled_before_start() {
-        let batch = create_test_batch();
-        let mock_stream = MockRecordBatchStream::new(vec![Ok(batch)]);
-        let process_manager = Arc::new(ProcessManager::new("".to_string(), None));
-        let ticket = process_manager.register_query(
-            "catalog".to_string(),
-            vec![],
-            "query".to_string(),
-            "client".to_string(),
-            None,
-        );
-
-        // Cancel before creating the wrapper
-        ticket.cancellation_handle.cancel();
-
-        let mut cancellable_stream = CancellableStreamWrapper::new(Box::pin(mock_stream), ticket);
-
-        let result = cancellable_stream.next().await;
-        assert!(result.is_some());
-        assert!(result.unwrap().is_err());
-    }
-
-    #[tokio::test]
-    async fn test_stream_cancelled_during_execution() {
-        let batch = create_test_batch();
-        let mock_stream =
-            MockRecordBatchStream::new(vec![Ok(batch)]).with_delay(Duration::from_millis(100));
-        let process_manager = Arc::new(ProcessManager::new("".to_string(), None));
-        let ticket = process_manager.register_query(
-            "catalog".to_string(),
-            vec![],
-            "query".to_string(),
-            "client".to_string(),
-            None,
-        );
-        let cancellation_handle = ticket.cancellation_handle.clone();
-
-        let mut cancellable_stream = CancellableStreamWrapper::new(Box::pin(mock_stream), ticket);
-
-        // Cancel after a short delay
-        tokio::spawn(async move {
-            sleep(Duration::from_millis(50)).await;
-            cancellation_handle.cancel();
-        });
-
-        let result = cancellable_stream.next().await;
-        assert!(result.is_some());
-        assert!(result.unwrap().is_err());
-    }
-
-    #[tokio::test]
-    async fn test_stream_completes_before_cancellation() {
-        let batch = create_test_batch();
-        let mock_stream = MockRecordBatchStream::new(vec![Ok(batch.clone())]);
-        let process_manager = Arc::new(ProcessManager::new("".to_string(), None));
-        let ticket = process_manager.register_query(
-            "catalog".to_string(),
-            vec![],
-            "query".to_string(),
-            "client".to_string(),
-            None,
-        );
-        let cancellation_handle = ticket.cancellation_handle.clone();
-
-        let mut cancellable_stream = CancellableStreamWrapper::new(Box::pin(mock_stream), ticket);
-
-        // Try to cancel after the stream should have completed
-        tokio::spawn(async move {
-            sleep(Duration::from_millis(100)).await;
-            cancellation_handle.cancel();
-        });
-
-        let result = cancellable_stream.next().await;
-        assert!(result.is_some());
-        assert!(result.unwrap().is_ok());
-    }
-
-    #[tokio::test]
-    async fn test_multiple_batches() {
-        let batch1 = create_test_batch();
-        let batch2 = create_test_batch();
-        let mock_stream = MockRecordBatchStream::new(vec![Ok(batch1), Ok(batch2)]);
-        let process_manager = Arc::new(ProcessManager::new("".to_string(), None));
-        let ticket = process_manager.register_query(
-            "catalog".to_string(),
-            vec![],
-            "query".to_string(),
-            "client".to_string(),
-            None,
-        );
-
-        let mut cancellable_stream = CancellableStreamWrapper::new(Box::pin(mock_stream), ticket);
-
-        // First batch
-        let result1 = cancellable_stream.next().await;
-        assert!(result1.is_some());
-        assert!(result1.unwrap().is_ok());
-
-        // Second batch
-        let result2 = cancellable_stream.next().await;
-        assert!(result2.is_some());
-        assert!(result2.unwrap().is_ok());
-
-        // End of stream
-        let end_result = cancellable_stream.next().await;
-        assert!(end_result.is_none());
-    }
-
-    #[tokio::test]
-    async fn test_record_batch_stream_methods() {
-        let batch = create_test_batch();
-        let mock_stream = MockRecordBatchStream::new(vec![Ok(batch)]);
-        let process_manager = Arc::new(ProcessManager::new("".to_string(), None));
-        let ticket = process_manager.register_query(
-            "catalog".to_string(),
-            vec![],
-            "query".to_string(),
-            "client".to_string(),
-            None,
-        );
-
-        let cancellable_stream = CancellableStreamWrapper::new(Box::pin(mock_stream), ticket);
-
-        // Test schema method
-        let schema = cancellable_stream.schema();
-        assert_eq!(schema.column_schemas().len(), 1);
-        assert_eq!(schema.column_schemas()[0].name, "test_col");
-
-        // Test output_ordering method
-        assert!(cancellable_stream.output_ordering().is_none());
-
-        // Test metrics method
-        assert!(cancellable_stream.metrics().is_none());
-    }
-
-    #[tokio::test]
-    async fn test_cancellation_during_pending_poll() {
-        let batch = create_test_batch();
-        let mock_stream =
-            MockRecordBatchStream::new(vec![Ok(batch)]).with_delay(Duration::from_millis(200));
-        let process_manager = Arc::new(ProcessManager::new("".to_string(), None));
-        let ticket = process_manager.register_query(
-            "catalog".to_string(),
-            vec![],
-            "query".to_string(),
-            "client".to_string(),
-            None,
-        );
-        let cancellation_handle = ticket.cancellation_handle.clone();
-
-        let mut cancellable_stream = CancellableStreamWrapper::new(Box::pin(mock_stream), ticket);
-
-        // Cancel while the stream is pending
-        tokio::spawn(async move {
-            sleep(Duration::from_millis(50)).await;
-            cancellation_handle.cancel();
-        });
-
-        let result = timeout(Duration::from_millis(300), cancellable_stream.next()).await;
-        assert!(result.is_ok());
-        let stream_result = result.unwrap();
-        assert!(stream_result.is_some());
-        assert!(stream_result.unwrap().is_err());
-    }
-}
--- a/src/metric-engine/Cargo.toml
+++ b/src/metric-engine/Cargo.toml
@@ -27,7 +27,6 @@ futures-util.workspace = true
 humantime-serde.workspace = true
 itertools.workspace = true
 lazy_static = "1.4"
-mito-codec.workspace = true
 mito2.workspace = true
 mur3 = "0.1"
 object-store.workspace = true
--- a/src/metric-engine/src/error.rs
+++ b/src/metric-engine/src/error.rs
@@ -115,7 +115,7 @@ pub enum Error {

    #[snafu(display("Failed to encode primary key"))]
    EncodePrimaryKey {
-        source: mito_codec::error::Error,
+        source: mito2::error::Error,
        #[snafu(implicit)]
        location: Location,
    },
--- a/src/metric-engine/src/row_modifier.rs
+++ b/src/metric-engine/src/row_modifier.rs
@@ -18,7 +18,7 @@ use std::hash::Hash;
 use api::v1::value::ValueData;
 use api::v1::{ColumnDataType, ColumnSchema, Row, Rows, SemanticType, Value};
 use datatypes::value::ValueRef;
-use mito_codec::row_converter::SparsePrimaryKeyCodec;
+use mito2::row_converter::SparsePrimaryKeyCodec;
 use smallvec::SmallVec;
 use snafu::ResultExt;
 use store_api::codec::PrimaryKeyEncoding;
--- a/src/mito-codec/Cargo.toml
+++ b/src/mito-codec/Cargo.toml
@@ -1,30 +0,0 @@
-[package]
-name = "mito-codec"
-version.workspace = true
-edition.workspace = true
-license.workspace = true
-
-[features]
-default = []
-testing = []
-
-[dependencies]
-api.workspace = true
-bytes.workspace = true
-common-base.workspace = true
-common-decimal.workspace = true
-common-error.workspace = true
-common-macro.workspace = true
-common-recordbatch.workspace = true
-common-telemetry.workspace = true
-common-time.workspace = true
-datatypes.workspace = true
-memcomparable = "0.2"
-paste.workspace = true
-serde.workspace = true
-snafu.workspace = true
-store-api.workspace = true
-
-[dev-dependencies]
-datafusion-common.workspace = true
-datafusion-expr.workspace = true
--- a/src/mito-codec/src/error.rs
+++ b/src/mito-codec/src/error.rs
@@ -1,95 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::any::Any;
-
-use common_error::ext::ErrorExt;
-use common_error::status_code::StatusCode;
-use common_macro::stack_trace_debug;
-use datatypes::prelude::ConcreteDataType;
-use snafu::{Location, Snafu};
-
-/// Error definitions for mito encoding.
-#[derive(Snafu)]
-#[snafu(visibility(pub))]
-#[stack_trace_debug]
-pub enum Error {
-    #[snafu(display("Row value mismatches field data type"))]
-    FieldTypeMismatch {
-        // Box the source to reduce the size of the error.
-        #[snafu(source(from(datatypes::error::Error, Box::new)))]
-        source: Box<datatypes::error::Error>,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
-    #[snafu(display("Failed to serialize field"))]
-    SerializeField {
-        #[snafu(source)]
-        error: memcomparable::Error,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
-    #[snafu(display(
-        "Data type: {} does not support serialization/deserialization",
-        data_type,
-    ))]
-    NotSupportedField {
-        data_type: ConcreteDataType,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
-    #[snafu(display("Failed to deserialize field"))]
-    DeserializeField {
-        #[snafu(source)]
-        error: memcomparable::Error,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
-    #[snafu(display("Operation not supported: {}", err_msg))]
-    UnsupportedOperation {
-        err_msg: String,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
-    #[snafu(display("Encode null value"))]
-    IndexEncodeNull {
-        #[snafu(implicit)]
-        location: Location,
-    },
-}
-
-pub type Result<T, E = Error> = std::result::Result<T, E>;
-
-impl ErrorExt for Error {
-    fn status_code(&self) -> StatusCode {
-        use Error::*;
-
-        match self {
-            FieldTypeMismatch { source, .. } => source.status_code(),
-            SerializeField { .. } | DeserializeField { .. } | IndexEncodeNull { .. } => {
-                StatusCode::InvalidArguments
-            }
-            NotSupportedField { .. } | UnsupportedOperation { .. } => StatusCode::Unsupported,
-        }
-    }
-
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-}
--- a/src/mito-codec/src/lib.rs
+++ b/src/mito-codec/src/lib.rs
@@ -1,24 +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.
-
-//! Codec utilities for the Mito protocol.
-
-pub mod error;
-pub mod index;
-pub mod key_values;
-pub mod primary_key_filter;
-pub mod row_converter;
-
-#[cfg(any(test, feature = "testing"))]
-pub mod test_util;
--- a/src/mito-codec/src/row_converter.rs
+++ b/src/mito-codec/src/row_converter.rs
@@ -1,164 +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.
-
-pub mod dense;
-pub mod sparse;
-
-use std::fmt::Debug;
-use std::sync::Arc;
-
-use common_recordbatch::filter::SimpleFilterEvaluator;
-use datatypes::value::{Value, ValueRef};
-pub use dense::{DensePrimaryKeyCodec, SortField};
-pub use sparse::{SparsePrimaryKeyCodec, SparseValues, COLUMN_ID_ENCODE_SIZE};
-use store_api::codec::PrimaryKeyEncoding;
-use store_api::metadata::{RegionMetadata, RegionMetadataRef};
-use store_api::storage::ColumnId;
-
-use crate::error::Result;
-use crate::key_values::KeyValue;
-
-/// Row value encoder/decoder.
-pub trait PrimaryKeyCodecExt {
-    /// Encodes rows to bytes.
-    /// # Note
-    /// Ensure the length of row iterator matches the length of fields.
-    fn encode<'a, I>(&self, row: I) -> Result<Vec<u8>>
-    where
-        I: Iterator<Item = ValueRef<'a>>,
-    {
-        let mut buffer = Vec::new();
-        self.encode_to_vec(row, &mut buffer)?;
-        Ok(buffer)
-    }
-
-    /// Encodes rows to specific vec.
-    /// # Note
-    /// Ensure the length of row iterator matches the length of fields.
-    fn encode_to_vec<'a, I>(&self, row: I, buffer: &mut Vec<u8>) -> Result<()>
-    where
-        I: Iterator<Item = ValueRef<'a>>;
-}
-
-pub trait PrimaryKeyFilter: Send + Sync {
-    /// Returns true if the primary key matches the filter.
-    fn matches(&mut self, pk: &[u8]) -> bool;
-}
-
-/// Composite values decoded from primary key bytes.
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub enum CompositeValues {
-    Dense(Vec<(ColumnId, Value)>),
-    Sparse(SparseValues),
-}
-
-impl CompositeValues {
-    /// Extends the composite values with the given values.
-    pub fn extend(&mut self, values: &[(ColumnId, Value)]) {
-        match self {
-            CompositeValues::Dense(dense_values) => {
-                for (column_id, value) in values {
-                    dense_values.push((*column_id, value.clone()));
-                }
-            }
-            CompositeValues::Sparse(sprase_value) => {
-                for (column_id, value) in values {
-                    sprase_value.insert(*column_id, value.clone());
-                }
-            }
-        }
-    }
-}
-
-#[cfg(any(test, feature = "testing"))]
-impl CompositeValues {
-    pub fn into_sparse(self) -> SparseValues {
-        match self {
-            CompositeValues::Sparse(v) => v,
-            _ => panic!("CompositeValues is not sparse"),
-        }
-    }
-
-    pub fn into_dense(self) -> Vec<Value> {
-        match self {
-            CompositeValues::Dense(v) => v.into_iter().map(|(_, v)| v).collect(),
-            _ => panic!("CompositeValues is not dense"),
-        }
-    }
-}
-
-pub trait PrimaryKeyCodec: Send + Sync + Debug {
-    /// Encodes a key value to bytes.
-    fn encode_key_value(&self, key_value: &KeyValue, buffer: &mut Vec<u8>) -> Result<()>;
-
-    /// Encodes values to bytes.
-    fn encode_values(&self, values: &[(ColumnId, Value)], buffer: &mut Vec<u8>) -> Result<()>;
-
-    /// Encodes values to bytes.
-    fn encode_value_refs(
-        &self,
-        values: &[(ColumnId, ValueRef)],
-        buffer: &mut Vec<u8>,
-    ) -> Result<()>;
-
-    /// Returns the number of fields in the primary key.
-    fn num_fields(&self) -> Option<usize>;
-
-    /// Returns a primary key filter factory.
-    fn primary_key_filter(
-        &self,
-        metadata: &RegionMetadataRef,
-        filters: Arc<Vec<SimpleFilterEvaluator>>,
-    ) -> Box<dyn PrimaryKeyFilter>;
-
-    /// Returns the estimated size of the primary key.
-    fn estimated_size(&self) -> Option<usize> {
-        None
-    }
-
-    /// Returns the encoding type of the primary key.
-    fn encoding(&self) -> PrimaryKeyEncoding;
-
-    /// Decodes the primary key from the given bytes.
-    ///
-    /// Returns a [`CompositeValues`] that follows the primary key ordering.
-    fn decode(&self, bytes: &[u8]) -> Result<CompositeValues>;
-
-    /// Decode the leftmost value from bytes.
-    fn decode_leftmost(&self, bytes: &[u8]) -> Result<Option<Value>>;
-}
-
-/// Builds a primary key codec from region metadata.
-pub fn build_primary_key_codec(region_metadata: &RegionMetadata) -> Arc<dyn PrimaryKeyCodec> {
-    let fields = region_metadata.primary_key_columns().map(|col| {
-        (
-            col.column_id,
-            SortField::new(col.column_schema.data_type.clone()),
-        )
-    });
-    build_primary_key_codec_with_fields(region_metadata.primary_key_encoding, fields)
-}
-
-/// Builds a primary key codec from region metadata.
-pub fn build_primary_key_codec_with_fields(
-    encoding: PrimaryKeyEncoding,
-    fields: impl Iterator<Item = (ColumnId, SortField)>,
-) -> Arc<dyn PrimaryKeyCodec> {
-    match encoding {
-        PrimaryKeyEncoding::Dense => Arc::new(DensePrimaryKeyCodec::with_fields(fields.collect())),
-        PrimaryKeyEncoding::Sparse => {
-            Arc::new(SparsePrimaryKeyCodec::with_fields(fields.collect()))
-        }
-    }
-}
--- a/src/mito2/Cargo.toml
+++ b/src/mito2/Cargo.toml
@@ -48,7 +48,6 @@ itertools.workspace = true
 lazy_static = "1.4"
 log-store = { workspace = true }
 memcomparable = "0.2"
-mito-codec.workspace = true
 moka = { workspace = true, features = ["sync", "future"] }
 object-store.workspace = true
 parquet = { workspace = true, features = ["async"] }
@@ -83,7 +82,6 @@ common-test-util.workspace = true
 criterion = "0.4"
 dotenv.workspace = true
 log-store.workspace = true
-mito-codec = { workspace = true, features = ["testing"] }
 object-store = { workspace = true, features = ["services-memory"] }
 rskafka.workspace = true
 rstest.workspace = true
--- a/src/mito2/benches/memtable_bench.rs
+++ b/src/mito2/benches/memtable_bench.rs
@@ -25,8 +25,8 @@ use mito2::memtable::partition_tree::{PartitionTreeConfig, PartitionTreeMemtable
 use mito2::memtable::time_series::TimeSeriesMemtable;
 use mito2::memtable::{KeyValues, Memtable};
 use mito2::region::options::MergeMode;
+use mito2::row_converter::DensePrimaryKeyCodec;
 use mito2::test_util::memtable_util::{self, region_metadata_to_row_schema};
-use mito_codec::row_converter::DensePrimaryKeyCodec;
 use rand::rngs::ThreadRng;
 use rand::seq::IndexedRandom;
 use rand::Rng;
--- a/src/mito2/src/compaction/run.rs
+++ b/src/mito2/src/compaction/run.rs
@@ -18,9 +18,8 @@
 use common_base::readable_size::ReadableSize;
 use common_base::BitVec;
 use common_time::Timestamp;
-use smallvec::{smallvec, SmallVec};

-use crate::sst::file::{FileHandle, FileId};
+use crate::sst::file::FileHandle;

 /// Default max compaction output file size when not specified.
 const DEFAULT_MAX_OUTPUT_SIZE: u64 = ReadableSize::gb(2).as_bytes();
@@ -126,68 +125,17 @@ pub trait Item: Ranged + Clone {
    fn size(&self) -> usize;
 }

-/// A group of files that are created by the same compaction task.
-#[derive(Debug, Clone)]
-pub struct FileGroup {
-    files: SmallVec<[FileHandle; 2]>,
-    size: usize,
-    num_rows: usize,
-    min_timestamp: Timestamp,
-    max_timestamp: Timestamp,
-}
-
-impl FileGroup {
-    pub(crate) fn new_with_file(file: FileHandle) -> Self {
-        let size = file.size() as usize;
-        let (min_timestamp, max_timestamp) = file.time_range();
-        let num_rows = file.num_rows();
-        Self {
-            files: smallvec![file],
-            size,
-            num_rows,
-            min_timestamp,
-            max_timestamp,
-        }
-    }
-
-    pub(crate) fn num_rows(&self) -> usize {
-        self.num_rows
-    }
-
-    pub(crate) fn add_file(&mut self, file: FileHandle) {
-        self.size += file.size() as usize;
-        self.num_rows += file.num_rows();
-        let (min_timestamp, max_timestamp) = file.time_range();
-        self.min_timestamp = self.min_timestamp.min(min_timestamp);
-        self.max_timestamp = self.max_timestamp.max(max_timestamp);
-        self.files.push(file);
-    }
-
-    #[cfg(test)]
-    pub(crate) fn files(&self) -> &[FileHandle] {
-        &self.files[..]
-    }
-
-    pub(crate) fn file_ids(&self) -> SmallVec<[FileId; 2]> {
-        SmallVec::from_iter(self.files.iter().map(|f| f.file_id()))
-    }
-
-    pub(crate) fn into_files(self) -> impl Iterator<Item = FileHandle> {
-        self.files.into_iter()
-    }
-}
-
-impl Ranged for FileGroup {
+impl Ranged for FileHandle {
    type BoundType = Timestamp;

    fn range(&self) -> (Self::BoundType, Self::BoundType) {
-        (self.min_timestamp, self.max_timestamp)
+        self.time_range()
    }
 }

-impl Item for FileGroup {
+impl Item for FileHandle {
    fn size(&self) -> usize {
-        self.size
+        self.size() as usize
    }
 }

--- a/src/mito2/src/compaction/test_util.rs
+++ b/src/mito2/src/compaction/test_util.rs
@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use std::num::NonZeroU64;
-
 use common_time::Timestamp;

 use crate::sst::file::{FileHandle, FileId, FileMeta, Level};
@@ -25,23 +23,6 @@ pub fn new_file_handle(
    start_ts_millis: i64,
    end_ts_millis: i64,
    level: Level,
-) -> FileHandle {
-    new_file_handle_with_sequence(
-        file_id,
-        start_ts_millis,
-        end_ts_millis,
-        level,
-        start_ts_millis as u64,
-    )
-}
-
-/// Test util to create file handles.
-pub fn new_file_handle_with_sequence(
-    file_id: FileId,
-    start_ts_millis: i64,
-    end_ts_millis: i64,
-    level: Level,
-    sequence: u64,
 ) -> FileHandle {
    let file_purger = new_noop_file_purger();
    FileHandle::new(
@@ -58,7 +39,7 @@ pub fn new_file_handle_with_sequence(
            index_file_size: 0,
            num_rows: 0,
            num_row_groups: 0,
-            sequence: NonZeroU64::new(sequence),
+            sequence: None,
        },
        file_purger,
    )
--- a/src/mito2/src/compaction/twcs.rs
+++ b/src/mito2/src/compaction/twcs.rs
@@ -15,7 +15,6 @@
 use std::collections::hash_map::Entry;
 use std::collections::{BTreeMap, HashMap};
 use std::fmt::Debug;
-use std::num::NonZeroU64;

 use common_base::readable_size::ReadableSize;
 use common_telemetry::info;
@@ -27,9 +26,7 @@ use store_api::storage::RegionId;
 use crate::compaction::buckets::infer_time_bucket;
 use crate::compaction::compactor::CompactionRegion;
 use crate::compaction::picker::{Picker, PickerOutput};
-use crate::compaction::run::{
-    find_sorted_runs, merge_seq_files, reduce_runs, FileGroup, Item, Ranged,
-};
+use crate::compaction::run::{find_sorted_runs, merge_seq_files, reduce_runs};
 use crate::compaction::{get_expired_ssts, CompactionOutput};
 use crate::sst::file::{overlaps, FileHandle, Level};
 use crate::sst::version::LevelMeta;
@@ -63,8 +60,7 @@ impl TwcsPicker {
            if files.files.is_empty() {
                continue;
            }
-            let mut files_to_merge: Vec<_> = files.files().cloned().collect();
-            let sorted_runs = find_sorted_runs(&mut files_to_merge);
+            let sorted_runs = find_sorted_runs(&mut files.files);
            let found_runs = sorted_runs.len();
            // We only remove deletion markers if we found less than 2 runs and not in append mode.
            // because after compaction there will be no overlapping files.
@@ -94,7 +90,7 @@ impl TwcsPicker {
                );
                output.push(CompactionOutput {
                    output_level: LEVEL_COMPACTED, // always compact to l1
-                    inputs: inputs.into_iter().flat_map(|fg| fg.into_files()).collect(),
+                    inputs,
                    filter_deleted,
                    output_time_range: None, // we do not enforce output time range in twcs compactions.
                });
@@ -113,21 +109,21 @@ fn log_pick_result(
    file_num: usize,
    max_output_file_size: Option<u64>,
    filter_deleted: bool,
-    inputs: &[FileGroup],
+    inputs: &[FileHandle],
 ) {
    let input_file_str: Vec<String> = inputs
        .iter()
        .map(|f| {
-            let range = f.range();
+            let range = f.time_range();
            let start = range.0.to_iso8601_string();
            let end = range.1.to_iso8601_string();
            let num_rows = f.num_rows();
            format!(
-                "FileGroup{{id: {:?}, range: ({}, {}), size: {}, num rows: {} }}",
-                f.file_ids(),
+                "SST{{id: {}, range: ({}, {}), size: {}, num rows: {} }}",
+                f.file_id(),
                start,
                end,
-                ReadableSize(f.size() as u64),
+                ReadableSize(f.size()),
                num_rows
            )
        })
@@ -202,9 +198,7 @@ impl Picker for TwcsPicker {
 struct Window {
    start: Timestamp,
    end: Timestamp,
-    // Mapping from file sequence to file groups. Files with the same sequence is considered
-    // created from the same compaction task.
-    files: HashMap<Option<NonZeroU64>, FileGroup>,
+    files: Vec<FileHandle>,
    time_window: i64,
    overlapping: bool,
 }
@@ -213,11 +207,10 @@ impl Window {
    /// Creates a new [Window] with given file.
    fn new_with_file(file: FileHandle) -> Self {
        let (start, end) = file.time_range();
-        let files = HashMap::from([(file.meta_ref().sequence, FileGroup::new_with_file(file))]);
        Self {
            start,
            end,
-            files,
+            files: vec![file],
            time_window: 0,
            overlapping: false,
        }
@@ -233,19 +226,7 @@ impl Window {
        let (start, end) = file.time_range();
        self.start = self.start.min(start);
        self.end = self.end.max(end);
-
-        match self.files.entry(file.meta_ref().sequence) {
-            Entry::Occupied(mut o) => {
-                o.get_mut().add_file(file);
-            }
-            Entry::Vacant(v) => {
-                v.insert(FileGroup::new_with_file(file));
-            }
-        }
-    }
-
-    fn files(&self) -> impl Iterator<Item = &FileGroup> {
-        self.files.values()
+        self.files.push(file);
    }
 }

@@ -330,7 +311,7 @@ mod tests {
    use std::collections::HashSet;

    use super::*;
-    use crate::compaction::test_util::{new_file_handle, new_file_handle_with_sequence};
+    use crate::compaction::test_util::new_file_handle;
    use crate::sst::file::{FileId, Level};

    #[test]
@@ -390,9 +371,7 @@ mod tests {
            .iter(),
            3,
        );
-        let fgs = &windows.get(&0).unwrap().files;
-        assert_eq!(1, fgs.len());
-        assert_eq!(fgs.values().map(|f| f.files().len()).sum::<usize>(), 5);
+        assert_eq!(5, windows.get(&0).unwrap().files.len());

        let files = [FileId::random(); 3];
        let windows = assign_to_windows(
@@ -406,56 +385,15 @@ mod tests {
        );
        assert_eq!(
            files[0],
-            windows.get(&0).unwrap().files().next().unwrap().files()[0].file_id()
+            windows.get(&0).unwrap().files.first().unwrap().file_id()
        );
        assert_eq!(
            files[1],
-            windows.get(&3).unwrap().files().next().unwrap().files()[0].file_id()
+            windows.get(&3).unwrap().files.first().unwrap().file_id()
        );
        assert_eq!(
            files[2],
-            windows.get(&12).unwrap().files().next().unwrap().files()[0].file_id()
-        );
-    }
-
-    #[test]
-    fn test_assign_file_groups_to_windows() {
-        let files = [
-            FileId::random(),
-            FileId::random(),
-            FileId::random(),
-            FileId::random(),
-        ];
-        let windows = assign_to_windows(
-            [
-                new_file_handle_with_sequence(files[0], 0, 999, 0, 1),
-                new_file_handle_with_sequence(files[1], 0, 999, 0, 1),
-                new_file_handle_with_sequence(files[2], 0, 999, 0, 2),
-                new_file_handle_with_sequence(files[3], 0, 999, 0, 2),
-            ]
-            .iter(),
-            3,
-        );
-        assert_eq!(windows.len(), 1);
-        let fgs = &windows.get(&0).unwrap().files;
-        assert_eq!(2, fgs.len());
-        assert_eq!(
-            fgs.get(&NonZeroU64::new(1))
-                .unwrap()
-                .files()
-                .iter()
-                .map(|f| f.file_id())
-                .collect::<HashSet<_>>(),
-            [files[0], files[1]].into_iter().collect()
-        );
-        assert_eq!(
-            fgs.get(&NonZeroU64::new(2))
-                .unwrap()
-                .files()
-                .iter()
-                .map(|f| f.file_id())
-                .collect::<HashSet<_>>(),
-            [files[2], files[3]].into_iter().collect()
+            windows.get(&12).unwrap().files.first().unwrap().file_id()
        );
    }

@@ -470,22 +408,8 @@ mod tests {
        ];
        files[0].set_compacting(true);
        files[2].set_compacting(true);
-        let mut windows = assign_to_windows(files.iter(), 3);
-        let window0 = windows.remove(&0).unwrap();
-        assert_eq!(1, window0.files.len());
-        let candidates = window0
-            .files
-            .into_values()
-            .flat_map(|fg| fg.into_files())
-            .map(|f| f.file_id())
-            .collect::<HashSet<_>>();
-        assert_eq!(candidates.len(), 3);
-        assert_eq!(
-            candidates,
-            [files[1].file_id(), files[3].file_id(), files[4].file_id()]
-                .into_iter()
-                .collect::<HashSet<_>>()
-        );
+        let windows = assign_to_windows(files.iter(), 3);
+        assert_eq!(3, windows.get(&0).unwrap().files.len());
    }

    /// (Window value, overlapping, files' time ranges in window)
@@ -514,11 +438,9 @@ mod tests {
            let mut file_ranges = actual_window
                .files
                .iter()
-                .flat_map(|(_, f)| {
-                    f.files().iter().map(|f| {
-                        let (s, e) = f.time_range();
-                        (s.value(), e.value())
-                    })
+                .map(|f| {
+                    let (s, e) = f.time_range();
+                    (s.value(), e.value())
                })
                .collect::<Vec<_>>();
            file_ranges.sort_unstable_by(|l, r| l.0.cmp(&r.0).then(l.1.cmp(&r.1)));
@@ -685,10 +607,10 @@ mod tests {
        CompactionPickerTestCase {
            window_size: 3,
            input_files: [
-                new_file_handle_with_sequence(file_ids[0], -2000, -3, 0, 1),
-                new_file_handle_with_sequence(file_ids[1], -3000, -100, 0, 2),
-                new_file_handle_with_sequence(file_ids[2], 0, 2999, 0, 3), //active windows
-                new_file_handle_with_sequence(file_ids[3], 50, 2998, 0, 4), //active windows
+                new_file_handle(file_ids[0], -2000, -3, 0),
+                new_file_handle(file_ids[1], -3000, -100, 0),
+                new_file_handle(file_ids[2], 0, 2999, 0), //active windows
+                new_file_handle(file_ids[3], 50, 2998, 0), //active windows
            ]
            .to_vec(),
            expected_outputs: vec![
@@ -714,11 +636,11 @@ mod tests {
        CompactionPickerTestCase {
            window_size: 3,
            input_files: [
-                new_file_handle_with_sequence(file_ids[0], -2000, -3, 0, 1),
-                new_file_handle_with_sequence(file_ids[1], -3000, -100, 0, 2),
-                new_file_handle_with_sequence(file_ids[2], 0, 2999, 0, 3),
-                new_file_handle_with_sequence(file_ids[3], 50, 2998, 0, 4),
-                new_file_handle_with_sequence(file_ids[4], 11, 2990, 0, 5),
+                new_file_handle(file_ids[0], -2000, -3, 0),
+                new_file_handle(file_ids[1], -3000, -100, 0),
+                new_file_handle(file_ids[2], 0, 2999, 0),
+                new_file_handle(file_ids[3], 50, 2998, 0),
+                new_file_handle(file_ids[4], 11, 2990, 0),
            ]
            .to_vec(),
            expected_outputs: vec![
@@ -733,27 +655,6 @@ mod tests {
            ],
        }
        .check();
-
-        // Case 3:
-        // A compaction may split output into several files that have overlapping time ranges and same sequence,
-        // we should treat these files as one FileGroup.
-        let file_ids = (0..6).map(|_| FileId::random()).collect::<Vec<_>>();
-        CompactionPickerTestCase {
-            window_size: 3,
-            input_files: [
-                new_file_handle_with_sequence(file_ids[0], 0, 2999, 1, 1),
-                new_file_handle_with_sequence(file_ids[1], 0, 2998, 1, 1),
-                new_file_handle_with_sequence(file_ids[2], 3000, 5999, 1, 2),
-                new_file_handle_with_sequence(file_ids[3], 3000, 5000, 1, 2),
-                new_file_handle_with_sequence(file_ids[4], 11, 2990, 0, 3),
-            ]
-            .to_vec(),
-            expected_outputs: vec![ExpectedOutput {
-                input_files: vec![0, 1, 4],
-                output_level: 1,
-            }],
-        }
-        .check();
    }

    // TODO(hl): TTL tester that checks if get_expired_ssts function works as expected.
--- a/src/mito2/src/error.rs
+++ b/src/mito2/src/error.rs
@@ -42,13 +42,6 @@ use crate::worker::WorkerId;
 #[snafu(visibility(pub))]
 #[stack_trace_debug]
 pub enum Error {
-    #[snafu(display("Unexpected data type"))]
-    DataTypeMismatch {
-        source: datatypes::error::Error,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
    #[snafu(display("External error, context: {}", context))]
    External {
        source: BoxedError,
@@ -298,6 +291,35 @@ pub enum Error {
    #[snafu(display("Failed to write region"))]
    WriteGroup { source: Arc<Error> },

+    #[snafu(display("Row value mismatches field data type"))]
+    FieldTypeMismatch { source: datatypes::error::Error },
+
+    #[snafu(display("Failed to serialize field"))]
+    SerializeField {
+        #[snafu(source)]
+        error: memcomparable::Error,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display(
+        "Data type: {} does not support serialization/deserialization",
+        data_type,
+    ))]
+    NotSupportedField {
+        data_type: ConcreteDataType,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Failed to deserialize field"))]
+    DeserializeField {
+        #[snafu(source)]
+        error: memcomparable::Error,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
    #[snafu(display("Invalid parquet SST file {}, reason: {}", file, reason))]
    InvalidParquet {
        file: String,
@@ -1006,20 +1028,6 @@ pub enum Error {
        location: Location,
        source: common_grpc::Error,
    },
-
-    #[snafu(display("Failed to encode"))]
-    Encode {
-        #[snafu(implicit)]
-        location: Location,
-        source: mito_codec::error::Error,
-    },
-
-    #[snafu(display("Failed to decode"))]
-    Decode {
-        #[snafu(implicit)]
-        location: Location,
-        source: mito_codec::error::Error,
-    },
 }

 pub type Result<T, E = Error> = std::result::Result<T, E>;
@@ -1044,7 +1052,6 @@ impl ErrorExt for Error {
        use Error::*;

        match self {
-            DataTypeMismatch { source, .. } => source.status_code(),
            OpenDal { .. } | ReadParquet { .. } => StatusCode::StorageUnavailable,
            WriteWal { source, .. } | ReadWal { source, .. } | DeleteWal { source, .. } => {
                source.status_code()
@@ -1088,6 +1095,7 @@ impl ErrorExt for Error {
            | BiErrors { .. }
            | StopScheduler { .. }
            | ComputeVector { .. }
+            | SerializeField { .. }
            | EncodeMemtable { .. }
            | CreateDir { .. }
            | ReadDataPart { .. }
@@ -1099,7 +1107,9 @@ impl ErrorExt for Error {

            WriteParquet { .. } => StatusCode::StorageUnavailable,
            WriteGroup { source, .. } => source.status_code(),
-            EncodeSparsePrimaryKey { .. } => StatusCode::Unexpected,
+            FieldTypeMismatch { source, .. } => source.status_code(),
+            NotSupportedField { .. } => StatusCode::Unsupported,
+            DeserializeField { .. } | EncodeSparsePrimaryKey { .. } => StatusCode::Unexpected,
            InvalidBatch { .. } => StatusCode::InvalidArguments,
            InvalidRecordBatch { .. } => StatusCode::InvalidArguments,
            ConvertVector { source, .. } => source.status_code(),
@@ -1171,9 +1181,7 @@ impl ErrorExt for Error {
            ScanSeries { source, .. } => source.status_code(),

            ScanMultiTimes { .. } => StatusCode::InvalidArguments,
-            ConvertBulkWalEntry { source, .. } => source.status_code(),
-
-            Encode { source, .. } | Decode { source, .. } => source.status_code(),
+            Error::ConvertBulkWalEntry { source, .. } => source.status_code(),
        }
    }

--- a/src/mito2/src/lib.rs
+++ b/src/mito2/src/lib.rs
@@ -41,6 +41,7 @@ pub mod read;
 pub mod region;
 mod region_write_ctx;
 pub mod request;
+pub mod row_converter;
 pub mod schedule;
 pub mod sst;
 mod time_provider;
--- a/src/mito2/src/memtable.rs
+++ b/src/mito2/src/memtable.rs
@@ -21,8 +21,6 @@ use std::sync::Arc;

 pub use bulk::part::EncodedBulkPart;
 use common_time::Timestamp;
-use mito_codec::key_values::KeyValue;
-pub use mito_codec::key_values::KeyValues;
 use serde::{Deserialize, Serialize};
 use store_api::metadata::RegionMetadataRef;
 use store_api::storage::{ColumnId, SequenceNumber};
@@ -31,6 +29,8 @@ use table::predicate::Predicate;
 use crate::config::MitoConfig;
 use crate::error::Result;
 use crate::flush::WriteBufferManagerRef;
+use crate::memtable::key_values::KeyValue;
+pub use crate::memtable::key_values::KeyValues;
 use crate::memtable::partition_tree::{PartitionTreeConfig, PartitionTreeMemtableBuilder};
 use crate::memtable::time_series::TimeSeriesMemtableBuilder;
 use crate::metrics::WRITE_BUFFER_BYTES;
@@ -42,6 +42,7 @@ use crate::sst::file::FileTimeRange;

 mod builder;
 pub mod bulk;
+pub mod key_values;
 pub mod partition_tree;
 mod simple_bulk_memtable;
 mod stats;
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
Lei, HUANG	4534e4c31d	chore/enable-flight-encoder: ### Add Flight Compression Support - Configuration Updates: - Added `grpc.flight_compression` option to `config/config.md`, `config/datanode.example.toml`, and `config/frontend.example.toml` to specify compression modes for Arrow IPC service. - Code Enhancements: - Updated `FlightEncoder` in `src/common/grpc/src/flight.rs` to support compression modes. - Modified `RegionServer` and `DatanodeBuilder` in `src/datanode/src/datanode.rs` and `src/datanode/src/region_server.rs` to handle `FlightCompression`. - Integrated `FlightCompression` in `src/servers/src/grpc.rs` and `src/servers/src/grpc/flight.rs` to manage compression settings. - Testing and Integration: - Updated test utilities and integration tests in `tests-integration/src/grpc/flight.rs` and `tests-integration/src/test_util.rs` to include `FlightCompression`. Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>	2025-06-10 23:15:01 +08:00
discord9	3d8278dc4c	feat: truly limit time range by split window	2025-06-10 23:15:01 +08:00
discord9	e4328380b2	feat: better metrics	2025-06-10 23:15:01 +08:00
discord9	e962076207	feat(exp): adjust_flow admin function	2025-06-10 23:15:01 +08:00
discord9	9ef8ba6460	feat: flownode to frontend load balance with guess	2025-06-10 23:15:01 +08:00