chore: comment

Signed-off-by: discord9 <discord9@163.com>
chore: per review
2026-01-05 04:42:56 +00:00 · 2025-06-12 17:44:54 +08:00 · 2025-06-12 17:44:54 +08:00 · 2025-06-12 17:44:54 +08:00
408 changed files with 12271 additions and 26067 deletions
--- a/.github/labeler.yaml
+++ b/.github/labeler.yaml
@@ -1,15 +0,0 @@
-ci:
-  - changed-files:
-      - any-glob-to-any-file: .github/**
-
-docker:
-  - changed-files:
-      - any-glob-to-any-file: docker/**
-
-documentation:
-  - changed-files:
-      - any-glob-to-any-file: docs/**
-
-dashboard:
-  - changed-files:
-      - any-glob-to-any-file: grafana/**
--- a/.github/workflows/pr-labeling.yaml
+++ b/.github/workflows/pr-labeling.yaml
@@ -1,42 +0,0 @@
-name: 'PR Labeling'
-
-on:
-  pull_request_target:
-    types:
-      - opened
-      - synchronize
-      - reopened
-
-permissions:
-  contents: read
-  pull-requests: write
-  issues: write
-
-jobs:
-  labeler:
-    runs-on: ubuntu-latest
-    steps:
-      - name: Checkout sources
-        uses: actions/checkout@v4
-
-      - uses: actions/labeler@v5
-        with:
-          configuration-path: ".github/labeler.yaml"
-          repo-token: "${{ secrets.GITHUB_TOKEN }}"
-
-  size-label:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: pascalgn/size-label-action@v0.5.5
-        env:
-          GITHUB_TOKEN: "${{ secrets.GITHUB_TOKEN }}"
-        with:
-          sizes: >
-            {
-              "0": "XS",
-              "100": "S",
-              "300": "M",
-              "1000": "L",
-              "1500": "XL",
-              "2000": "XXL"
-            }
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1621,7 +1621,6 @@ dependencies = [
 "cache",
 "catalog",
 "chrono",
- "common-base",
 "common-catalog",
 "common-error",
 "common-frontend",
@@ -1670,9 +1669,9 @@ dependencies = [

 [[package]]
 name = "cc"
-version = "1.2.27"
+version = "1.1.24"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d487aa071b5f64da6f19a3e848e3578944b726ee5a4854b82172f02aa876bfdc"
+checksum = "812acba72f0a070b003d3697490d2b55b837230ae7c6c6497f05cc2ddbb8d938"
 dependencies = [
 "jobserver",
 "libc",
@@ -1950,7 +1949,6 @@ checksum = "1462739cb27611015575c0c11df5df7601141071f07518d56fcc1be504cbec97"
 name = "cli"
 version = "0.15.0"
 dependencies = [
- "async-stream",
 "async-trait",
 "auth",
 "base64 0.22.1",
@@ -1983,10 +1981,9 @@ dependencies = [
 "meta-srv",
 "nu-ansi-term",
 "object-store",
- "operator",
 "query",
 "rand 0.9.0",
- "reqwest 0.12.9",
+ "reqwest",
 "serde",
 "serde_json",
 "servers",
@@ -2118,14 +2115,13 @@ dependencies = [
 "mito2",
 "moka",
 "nu-ansi-term",
- "object-store",
 "plugins",
 "prometheus",
 "prost 0.13.5",
 "query",
 "rand 0.9.0",
 "regex",
- "reqwest 0.12.9",
+ "reqwest",
 "rexpect",
 "serde",
 "serde_json",
@@ -2221,7 +2217,6 @@ dependencies = [
 "humantime-serde",
 "meta-client",
 "num_cpus",
- "object-store",
 "serde",
 "serde_json",
 "serde_with",
@@ -2334,7 +2329,6 @@ dependencies = [
 "datafusion",
 "datafusion-common",
 "datafusion-expr",
- "datafusion-functions-aggregate-common",
 "datatypes",
 "derive_more",
 "geo",
@@ -2373,7 +2367,7 @@ dependencies = [
 "common-test-util",
 "common-version",
 "hyper 0.14.30",
- "reqwest 0.12.9",
+ "reqwest",
 "serde",
 "tempfile",
 "tokio",
@@ -2674,6 +2668,7 @@ dependencies = [
 name = "common-telemetry"
 version = "0.15.0"
 dependencies = [
+ "atty",
 "backtrace",
 "common-error",
 "console-subscriber",
@@ -3074,9 +3069,9 @@ dependencies = [

 [[package]]
 name = "crossbeam-channel"
-version = "0.5.15"
+version = "0.5.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "82b8f8f868b36967f9606790d1903570de9ceaf870a7bf9fbbd3016d636a2cb2"
+checksum = "33480d6946193aa8033910124896ca395333cae7e2d1113d1fef6c3272217df2"
 dependencies = [
 "crossbeam-utils",
 ]
@@ -3765,7 +3760,7 @@ dependencies = [
 "prometheus",
 "prost 0.13.5",
 "query",
- "reqwest 0.12.9",
+ "reqwest",
 "serde",
 "serde_json",
 "servers",
@@ -4738,7 +4733,6 @@ dependencies = [
 "log-store",
 "meta-client",
 "num_cpus",
- "object-store",
 "opentelemetry-proto 0.27.0",
 "operator",
 "otel-arrow-rust",
@@ -4761,7 +4755,6 @@ dependencies = [
 "substrait 0.15.0",
 "table",
 "tokio",
- "tokio-util",
 "toml 0.8.19",
 "tonic 0.12.3",
 "tower 0.5.2",
@@ -5148,7 +5141,7 @@ dependencies = [
 [[package]]
 name = "greptime-proto"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=464226cf8a4a22696503536a123d0b9e318582f4#464226cf8a4a22696503536a123d0b9e318582f4"
+source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=5f6119ac7952878d39dcde0343c4bf828d18ffc8#5f6119ac7952878d39dcde0343c4bf828d18ffc8"
 dependencies = [
 "prost 0.13.5",
 "serde",
@@ -7235,7 +7228,6 @@ dependencies = [
 name = "metric-engine"
 version = "0.15.0"
 dependencies = [
- "ahash 0.8.11",
 "api",
 "aquamarine",
 "async-stream",
@@ -7256,7 +7248,6 @@ dependencies = [
 "humantime-serde",
 "itertools 0.14.0",
 "lazy_static",
- "mito-codec",
 "mito2",
 "mur3",
 "object-store",
@@ -7322,29 +7313,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"
@@ -7387,7 +7355,6 @@ dependencies = [
 "lazy_static",
 "log-store",
 "memcomparable",
- "mito-codec",
 "moka",
 "object-store",
 "parquet",
@@ -8101,21 +8068,14 @@ version = "0.15.0"
 dependencies = [
 "anyhow",
 "bytes",
- "common-base",
- "common-error",
- "common-macro",
 "common-telemetry",
 "common-test-util",
 "futures",
- "humantime-serde",
 "lazy_static",
 "md5",
 "moka",
 "opendal",
 "prometheus",
- "reqwest 0.12.9",
- "serde",
- "snafu 0.8.5",
 "tokio",
 "uuid",
 ]
@@ -8250,7 +8210,7 @@ dependencies = [
 "prometheus",
 "quick-xml 0.36.2",
 "reqsign",
- "reqwest 0.12.9",
+ "reqwest",
 "serde",
 "serde_json",
 "sha2",
@@ -8322,19 +8282,6 @@ dependencies = [
 "tracing",
 ]

-[[package]]
-name = "opentelemetry-http"
-version = "0.10.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7f51189ce8be654f9b5f7e70e49967ed894e84a06fc35c6c042e64ac1fc5399e"
-dependencies = [
- "async-trait",
- "bytes",
- "http 0.2.12",
- "opentelemetry 0.21.0",
- "reqwest 0.11.27",
-]
-
 [[package]]
 name = "opentelemetry-otlp"
 version = "0.14.0"
@@ -8345,12 +8292,10 @@ dependencies = [
 "futures-core",
 "http 0.2.12",
 "opentelemetry 0.21.0",
- "opentelemetry-http",
 "opentelemetry-proto 0.4.0",
 "opentelemetry-semantic-conventions",
 "opentelemetry_sdk 0.21.2",
 "prost 0.11.9",
- "reqwest 0.11.27",
 "thiserror 1.0.64",
 "tokio",
 "tonic 0.9.2",
@@ -8449,7 +8394,6 @@ dependencies = [
 "common-catalog",
 "common-datasource",
 "common-error",
- "common-frontend",
 "common-function",
 "common-grpc",
 "common-grpc-expr",
@@ -8950,7 +8894,8 @@ dependencies = [
 [[package]]
 name = "pgwire"
 version = "0.30.2"
-source = "git+https://github.com/sunng87/pgwire?rev=127573d997228cfb70c7699881c568eae8131270#127573d997228cfb70c7699881c568eae8131270"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4ca6c26b25be998208a13ff2f0c55b567363f34675410e6d6f1c513a150583fd"
 dependencies = [
 "async-trait",
 "bytes",
@@ -10337,7 +10282,7 @@ dependencies = [
 "percent-encoding",
 "quick-xml 0.35.0",
 "rand 0.8.5",
- "reqwest 0.12.9",
+ "reqwest",
 "rsa",
 "rust-ini 0.21.1",
 "serde",
@@ -10346,42 +10291,6 @@ dependencies = [
 "sha2",
 ]

-[[package]]
-name = "reqwest"
-version = "0.11.27"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dd67538700a17451e7cba03ac727fb961abb7607553461627b97de0b89cf4a62"
-dependencies = [
- "base64 0.21.7",
- "bytes",
- "encoding_rs",
- "futures-core",
- "futures-util",
- "h2 0.3.26",
- "http 0.2.12",
- "http-body 0.4.6",
- "hyper 0.14.30",
- "ipnet",
- "js-sys",
- "log",
- "mime",
- "once_cell",
- "percent-encoding",
- "pin-project-lite",
- "serde",
- "serde_json",
- "serde_urlencoded",
- "sync_wrapper 0.1.2",
- "system-configuration",
- "tokio",
- "tower-service",
- "url",
- "wasm-bindgen",
- "wasm-bindgen-futures",
- "web-sys",
- "winreg",
-]
-
 [[package]]
 name = "reqwest"
 version = "0.12.9"
@@ -10452,14 +10361,15 @@ dependencies = [

 [[package]]
 name = "ring"
-version = "0.17.14"
+version = "0.17.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7"
+checksum = "c17fa4cb658e3583423e915b9f3acc01cceaee1860e33d59ebae66adc3a2dc0d"
 dependencies = [
 "cc",
 "cfg-if",
 "getrandom 0.2.15",
 "libc",
+ "spin",
 "untrusted",
 "windows-sys 0.52.0",
 ]
@@ -11232,7 +11142,6 @@ dependencies = [
 "common-base",
 "common-catalog",
 "common-config",
- "common-datasource",
 "common-error",
 "common-frontend",
 "common-grpc",
@@ -11275,23 +11184,16 @@ dependencies = [
 "local-ip-address",
 "log-query",
 "loki-proto",
- "metric-engine",
 "mime_guess",
- "mito-codec",
- "mito2",
 "mysql_async",
 "notify",
 "object-pool",
- "object-store",
 "once_cell",
 "openmetrics-parser",
 "opensrv-mysql",
 "opentelemetry-proto 0.27.0",
- "operator",
 "otel-arrow-rust",
 "parking_lot 0.12.3",
- "parquet",
- "partition",
 "permutation",
 "pgwire",
 "pin-project",
@@ -11305,7 +11207,7 @@ dependencies = [
 "quoted-string",
 "rand 0.9.0",
 "regex",
- "reqwest 0.12.9",
+ "reqwest",
 "rust-embed",
 "rustls",
 "rustls-pemfile",
@@ -11749,7 +11651,7 @@ dependencies = [
 "local-ip-address",
 "mysql",
 "num_cpus",
- "reqwest 0.12.9",
+ "reqwest",
 "serde",
 "serde_json",
 "sha2",
@@ -12399,27 +12301,6 @@ dependencies = [
 "nom",
 ]

-[[package]]
-name = "system-configuration"
-version = "0.5.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ba3a3adc5c275d719af8cb4272ea1c4a6d668a777f37e115f6d11ddbc1c8e0e7"
-dependencies = [
- "bitflags 1.3.2",
- "core-foundation",
- "system-configuration-sys",
-]
-
-[[package]]
-name = "system-configuration-sys"
-version = "0.5.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a75fb188eb626b924683e3b95e3a48e63551fcfb51949de2f06a9d91dbee93c9"
-dependencies = [
- "core-foundation-sys",
- "libc",
-]
-
 [[package]]
 name = "table"
 version = "0.15.0"
@@ -12710,7 +12591,7 @@ dependencies = [
 "paste",
 "rand 0.9.0",
 "rand_chacha 0.9.0",
- "reqwest 0.12.9",
+ "reqwest",
 "schemars",
 "serde",
 "serde_json",
@@ -13866,13 +13747,12 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"

 [[package]]
 name = "uuid"
-version = "1.17.0"
+version = "1.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3cf4199d1e5d15ddd86a694e4d0dffa9c323ce759fea589f00fef9d81cc1931d"
+checksum = "81dfa00651efa65069b0b6b651f4aaa31ba9e3c3ce0137aaad053604ee7e0314"
 dependencies = [
- "getrandom 0.3.2",
- "js-sys",
- "rand 0.9.0",
+ "getrandom 0.2.15",
+ "rand 0.8.5",
 "serde",
 "wasm-bindgen",
 ]
@@ -14594,16 +14474,6 @@ dependencies = [
 "memchr",
 ]

-[[package]]
-name = "winreg"
-version = "0.50.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "524e57b2c537c0f9b1e69f1965311ec12182b4122e45035b1508cd24d2adadb1"
-dependencies = [
- "cfg-if",
- "windows-sys 0.48.0",
-]
-
 [[package]]
 name = "wit-bindgen-rt"
 version = "0.39.0"
--- 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",
@@ -121,7 +120,6 @@ datafusion = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "
 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-functions-aggregate-common = { 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" }
@@ -135,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 = "464226cf8a4a22696503536a123d0b9e318582f4" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "5f6119ac7952878d39dcde0343c4bf828d18ffc8" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"
@@ -276,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/README.md
+++ b/README.md
@@ -189,8 +189,7 @@ We invite you to engage and contribute!
 - [Official Website](https://greptime.com/)
 - [Blog](https://greptime.com/blogs/)
 - [LinkedIn](https://www.linkedin.com/company/greptime/)
- [X (Twitter)](https://X.com/greptime)
- [YouTube](https://www.youtube.com/@greptime)
+- [Twitter](https://twitter.com/greptime)

 ## License

--- a/config/config.md
+++ b/config/config.md
@@ -123,7 +123,6 @@
 | `storage.http_client.connect_timeout` | String | `30s` | The timeout for only the connect phase of a http client. |
 | `storage.http_client.timeout` | String | `30s` | The total request timeout, applied from when the request starts connecting until the response body has finished.<br/>Also considered a total deadline. |
 | `storage.http_client.pool_idle_timeout` | String | `90s` | The timeout for idle sockets being kept-alive. |
-| `storage.http_client.skip_ssl_validation` | Bool | `false` | To skip the ssl verification<br/>**Security Notice**: Setting `skip_ssl_validation = true` disables certificate verification, making connections vulnerable to man-in-the-middle attacks. Only use this in development or trusted private networks. |
 | `[[region_engine]]` | -- | -- | The region engine options. You can configure multiple region engines. |
 | `region_engine.mito` | -- | -- | The Mito engine options. |
 | `region_engine.mito.num_workers` | Integer | `8` | Number of region workers. |
@@ -185,11 +184,10 @@
 | `logging.dir` | String | `./greptimedb_data/logs` | The directory to store the log files. If set to empty, logs will not be written to files. |
 | `logging.level` | String | Unset | The log level. Can be `info`/`debug`/`warn`/`error`. |
 | `logging.enable_otlp_tracing` | Bool | `false` | Enable OTLP tracing. |
-| `logging.otlp_endpoint` | String | `http://localhost:4318` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4317` | The OTLP tracing endpoint. |
 | `logging.append_stdout` | Bool | `true` | Whether to append logs to stdout. |
 | `logging.log_format` | String | `text` | The log format. Can be `text`/`json`. |
 | `logging.max_log_files` | Integer | `720` | The maximum amount of log files. |
-| `logging.otlp_export_protocol` | String | `http` | The OTLP tracing export protocol. Can be `grpc`/`http`. |
 | `logging.tracing_sample_ratio` | -- | -- | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
 | `logging.tracing_sample_ratio.default_ratio` | Float | `1.0` | -- |
 | `slow_query` | -- | -- | The slow query log options. |
@@ -289,11 +287,10 @@
 | `logging.dir` | String | `./greptimedb_data/logs` | The directory to store the log files. If set to empty, logs will not be written to files. |
 | `logging.level` | String | Unset | The log level. Can be `info`/`debug`/`warn`/`error`. |
 | `logging.enable_otlp_tracing` | Bool | `false` | Enable OTLP tracing. |
-| `logging.otlp_endpoint` | String | `http://localhost:4318` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4317` | The OTLP tracing endpoint. |
 | `logging.append_stdout` | Bool | `true` | Whether to append logs to stdout. |
 | `logging.log_format` | String | `text` | The log format. Can be `text`/`json`. |
 | `logging.max_log_files` | Integer | `720` | The maximum amount of log files. |
-| `logging.otlp_export_protocol` | String | `http` | The OTLP tracing export protocol. Can be `grpc`/`http`. |
 | `logging.tracing_sample_ratio` | -- | -- | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
 | `logging.tracing_sample_ratio.default_ratio` | Float | `1.0` | -- |
 | `slow_query` | -- | -- | The slow query log options. |
@@ -325,7 +322,6 @@
 | `selector` | String | `round_robin` | Datanode selector type.<br/>- `round_robin` (default value)<br/>- `lease_based`<br/>- `load_based`<br/>For details, please see "https://docs.greptime.com/developer-guide/metasrv/selector". |
 | `use_memory_store` | Bool | `false` | Store data in memory. |
 | `enable_region_failover` | Bool | `false` | Whether to enable region failover.<br/>This feature is only available on GreptimeDB running on cluster mode and<br/>- Using Remote WAL<br/>- Using shared storage (e.g., s3). |
-| `region_failure_detector_initialization_delay` | String | `10m` | Delay before initializing region failure detectors.<br/>This delay helps prevent premature initialization of region failure detectors in cases where<br/>cluster maintenance mode is enabled right after metasrv starts, especially when the cluster<br/>is not deployed via the recommended GreptimeDB Operator. Without this delay, early detector registration<br/>may trigger unnecessary region failovers during datanode startup. |
 | `allow_region_failover_on_local_wal` | Bool | `false` | Whether to allow region failover on local WAL.<br/>**This option is not recommended to be set to true, because it may lead to data loss during failover.** |
 | `node_max_idle_time` | String | `24hours` | Max allowed idle time before removing node info from metasrv memory. |
 | `enable_telemetry` | Bool | `true` | Whether to enable greptimedb telemetry. Enabled by default. |
@@ -373,11 +369,10 @@
 | `logging.dir` | String | `./greptimedb_data/logs` | The directory to store the log files. If set to empty, logs will not be written to files. |
 | `logging.level` | String | Unset | The log level. Can be `info`/`debug`/`warn`/`error`. |
 | `logging.enable_otlp_tracing` | Bool | `false` | Enable OTLP tracing. |
-| `logging.otlp_endpoint` | String | `http://localhost:4318` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4317` | The OTLP tracing endpoint. |
 | `logging.append_stdout` | Bool | `true` | Whether to append logs to stdout. |
 | `logging.log_format` | String | `text` | The log format. Can be `text`/`json`. |
 | `logging.max_log_files` | Integer | `720` | The maximum amount of log files. |
-| `logging.otlp_export_protocol` | String | `http` | The OTLP tracing export protocol. Can be `grpc`/`http`. |
 | `logging.tracing_sample_ratio` | -- | -- | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
 | `logging.tracing_sample_ratio.default_ratio` | Float | `1.0` | -- |
 | `export_metrics` | -- | -- | The metasrv can export its metrics and send to Prometheus compatible service (e.g. `greptimedb` itself) from remote-write API.<br/>This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape. |
@@ -476,7 +471,6 @@
 | `storage.http_client.connect_timeout` | String | `30s` | The timeout for only the connect phase of a http client. |
 | `storage.http_client.timeout` | String | `30s` | The total request timeout, applied from when the request starts connecting until the response body has finished.<br/>Also considered a total deadline. |
 | `storage.http_client.pool_idle_timeout` | String | `90s` | The timeout for idle sockets being kept-alive. |
-| `storage.http_client.skip_ssl_validation` | Bool | `false` | To skip the ssl verification<br/>**Security Notice**: Setting `skip_ssl_validation = true` disables certificate verification, making connections vulnerable to man-in-the-middle attacks. Only use this in development or trusted private networks. |
 | `[[region_engine]]` | -- | -- | The region engine options. You can configure multiple region engines. |
 | `region_engine.mito` | -- | -- | The Mito engine options. |
 | `region_engine.mito.num_workers` | Integer | `8` | Number of region workers. |
@@ -538,11 +532,10 @@
 | `logging.dir` | String | `./greptimedb_data/logs` | The directory to store the log files. If set to empty, logs will not be written to files. |
 | `logging.level` | String | Unset | The log level. Can be `info`/`debug`/`warn`/`error`. |
 | `logging.enable_otlp_tracing` | Bool | `false` | Enable OTLP tracing. |
-| `logging.otlp_endpoint` | String | `http://localhost:4318` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4317` | The OTLP tracing endpoint. |
 | `logging.append_stdout` | Bool | `true` | Whether to append logs to stdout. |
 | `logging.log_format` | String | `text` | The log format. Can be `text`/`json`. |
 | `logging.max_log_files` | Integer | `720` | The maximum amount of log files. |
-| `logging.otlp_export_protocol` | String | `http` | The OTLP tracing export protocol. Can be `grpc`/`http`. |
 | `logging.tracing_sample_ratio` | -- | -- | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
 | `logging.tracing_sample_ratio.default_ratio` | Float | `1.0` | -- |
 | `export_metrics` | -- | -- | The datanode can export its metrics and send to Prometheus compatible service (e.g. `greptimedb` itself) from remote-write API.<br/>This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape. |
@@ -589,11 +582,10 @@
 | `logging.dir` | String | `./greptimedb_data/logs` | The directory to store the log files. If set to empty, logs will not be written to files. |
 | `logging.level` | String | Unset | The log level. Can be `info`/`debug`/`warn`/`error`. |
 | `logging.enable_otlp_tracing` | Bool | `false` | Enable OTLP tracing. |
-| `logging.otlp_endpoint` | String | `http://localhost:4318` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4317` | The OTLP tracing endpoint. |
 | `logging.append_stdout` | Bool | `true` | Whether to append logs to stdout. |
 | `logging.log_format` | String | `text` | The log format. Can be `text`/`json`. |
 | `logging.max_log_files` | Integer | `720` | The maximum amount of log files. |
-| `logging.otlp_export_protocol` | String | `http` | The OTLP tracing export protocol. Can be `grpc`/`http`. |
 | `logging.tracing_sample_ratio` | -- | -- | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
 | `logging.tracing_sample_ratio.default_ratio` | Float | `1.0` | -- |
 | `tracing` | -- | -- | The tracing options. Only effect when compiled with `tokio-console` feature. |
--- a/config/datanode.example.toml
+++ b/config/datanode.example.toml
@@ -367,10 +367,6 @@ timeout = "30s"
 ## The timeout for idle sockets being kept-alive.
 pool_idle_timeout = "90s"

-## To skip the ssl verification
-## **Security Notice**: Setting `skip_ssl_validation = true` disables certificate verification, making connections vulnerable to man-in-the-middle attacks. Only use this in development or trusted private networks.
-skip_ssl_validation = false
-
 # Custom storage options
 # [[storage.providers]]
 # name = "S3"
@@ -629,7 +625,7 @@ level = "info"
 enable_otlp_tracing = false

 ## The OTLP tracing endpoint.
-otlp_endpoint = "http://localhost:4318"
+otlp_endpoint = "http://localhost:4317"

 ## Whether to append logs to stdout.
 append_stdout = true
@@ -640,9 +636,6 @@ log_format = "text"
 ## The maximum amount of log files.
 max_log_files = 720

-## The OTLP tracing export protocol. Can be `grpc`/`http`.
-otlp_export_protocol = "http"
-
 ## The percentage of tracing will be sampled and exported.
 ## Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.
 ## ratio > 1 are treated as 1. Fractions < 0 are treated as 0
--- a/config/flownode.example.toml
+++ b/config/flownode.example.toml
@@ -83,7 +83,7 @@ level = "info"
 enable_otlp_tracing = false

 ## The OTLP tracing endpoint.
-otlp_endpoint = "http://localhost:4318"
+otlp_endpoint = "http://localhost:4317"

 ## Whether to append logs to stdout.
 append_stdout = true
@@ -94,9 +94,6 @@ log_format = "text"
 ## The maximum amount of log files.
 max_log_files = 720

-## The OTLP tracing export protocol. Can be `grpc`/`http`.
-otlp_export_protocol = "http"
-
 ## The percentage of tracing will be sampled and exported.
 ## Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.
 ## ratio > 1 are treated as 1. Fractions < 0 are treated as 0
--- a/config/frontend.example.toml
+++ b/config/frontend.example.toml
@@ -218,7 +218,7 @@ level = "info"
 enable_otlp_tracing = false

 ## The OTLP tracing endpoint.
-otlp_endpoint = "http://localhost:4318"
+otlp_endpoint = "http://localhost:4317"

 ## Whether to append logs to stdout.
 append_stdout = true
@@ -229,9 +229,6 @@ log_format = "text"
 ## The maximum amount of log files.
 max_log_files = 720

-## The OTLP tracing export protocol. Can be `grpc`/`http`.
-otlp_export_protocol = "http"
-
 ## The percentage of tracing will be sampled and exported.
 ## Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.
 ## ratio > 1 are treated as 1. Fractions < 0 are treated as 0
--- a/config/metasrv.example.toml
+++ b/config/metasrv.example.toml
@@ -43,13 +43,6 @@ use_memory_store = false
 ## - Using shared storage (e.g., s3).
 enable_region_failover = false

-## Delay before initializing region failure detectors.
-## This delay helps prevent premature initialization of region failure detectors in cases where
-## cluster maintenance mode is enabled right after metasrv starts, especially when the cluster
-## is not deployed via the recommended GreptimeDB Operator. Without this delay, early detector registration
-## may trigger unnecessary region failovers during datanode startup.
-region_failure_detector_initialization_delay = '10m'
-
 ## Whether to allow region failover on local WAL.
 ## **This option is not recommended to be set to true, because it may lead to data loss during failover.**
 allow_region_failover_on_local_wal = false
@@ -227,7 +220,7 @@ level = "info"
 enable_otlp_tracing = false

 ## The OTLP tracing endpoint.
-otlp_endpoint = "http://localhost:4318"
+otlp_endpoint = "http://localhost:4317"

 ## Whether to append logs to stdout.
 append_stdout = true
@@ -238,9 +231,6 @@ log_format = "text"
 ## The maximum amount of log files.
 max_log_files = 720

-## The OTLP tracing export protocol. Can be `grpc`/`http`.
-otlp_export_protocol = "http"
-
 ## The percentage of tracing will be sampled and exported.
 ## Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.
 ## ratio > 1 are treated as 1. Fractions < 0 are treated as 0
--- a/config/standalone.example.toml
+++ b/config/standalone.example.toml
@@ -458,10 +458,6 @@ timeout = "30s"
 ## The timeout for idle sockets being kept-alive.
 pool_idle_timeout = "90s"

-## To skip the ssl verification
-## **Security Notice**: Setting `skip_ssl_validation = true` disables certificate verification, making connections vulnerable to man-in-the-middle attacks. Only use this in development or trusted private networks.
-skip_ssl_validation = false
-
 # Custom storage options
 # [[storage.providers]]
 # name = "S3"
@@ -720,7 +716,7 @@ level = "info"
 enable_otlp_tracing = false

 ## The OTLP tracing endpoint.
-otlp_endpoint = "http://localhost:4318"
+otlp_endpoint = "http://localhost:4317"

 ## Whether to append logs to stdout.
 append_stdout = true
@@ -731,9 +727,6 @@ log_format = "text"
 ## The maximum amount of log files.
 max_log_files = 720

-## The OTLP tracing export protocol. Can be `grpc`/`http`.
-otlp_export_protocol = "http"
-
 ## The percentage of tracing will be sampled and exported.
 ## Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.
 ## ratio > 1 are treated as 1. Fractions < 0 are treated as 0
--- a/grafana/dashboards/metrics/cluster/dashboard.json
+++ b/grafana/dashboards/metrics/cluster/dashboard.json
--- a/grafana/dashboards/metrics/cluster/dashboard.md
+++ b/grafana/dashboards/metrics/cluster/dashboard.md
@@ -70,7 +70,6 @@
 | Inflight Flush | `greptime_mito_inflight_flush_count` | `timeseries` | Ongoing flush task count | `prometheus` | `none` | `[{{instance}}]-[{{pod}}]` |
 | Compaction Input/Output Bytes | `sum by(instance, pod) (greptime_mito_compaction_input_bytes)`<br/>`sum by(instance, pod) (greptime_mito_compaction_output_bytes)` | `timeseries` | Compaction oinput output bytes | `prometheus` | `bytes` | `[{{instance}}]-[{{pod}}]-input` |
 | Region Worker Handle Bulk Insert Requests | `histogram_quantile(0.95, sum by(le,instance, stage, pod) (rate(greptime_region_worker_handle_write_bucket[$__rate_interval])))`<br/>`sum by(instance, stage, pod) (rate(greptime_region_worker_handle_write_sum[$__rate_interval]))/sum by(instance, stage, pod) (rate(greptime_region_worker_handle_write_count[$__rate_interval]))` | `timeseries` | Per-stage elapsed time for region worker to handle bulk insert region requests. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-P95` |
-| Active Series and Field Builders Count | `sum by(instance, pod) (greptime_mito_memtable_active_series_count)`<br/>`sum by(instance, pod) (greptime_mito_memtable_field_builder_count)` | `timeseries` | Compaction oinput output bytes | `prometheus` | `none` | `[{{instance}}]-[{{pod}}]-series` |
 | Region Worker Convert Requests | `histogram_quantile(0.95, sum by(le, instance, stage, pod) (rate(greptime_datanode_convert_region_request_bucket[$__rate_interval])))`<br/>`sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_sum[$__rate_interval]))/sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_count[$__rate_interval]))` | `timeseries` | Per-stage elapsed time for region worker to decode requests. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-P95` |
 # OpenDAL
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
--- a/grafana/dashboards/metrics/cluster/dashboard.yaml
+++ b/grafana/dashboards/metrics/cluster/dashboard.yaml
@@ -612,21 +612,6 @@ groups:
                type: prometheus
                uid: ${metrics}
              legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-AVG'
-        - title: Active Series and Field Builders Count
-          type: timeseries
-          description: Compaction oinput output bytes
-          unit: none
-          queries:
-            - expr: sum by(instance, pod) (greptime_mito_memtable_active_series_count)
-              datasource:
-                type: prometheus
-                uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-series'
-            - expr: sum by(instance, pod) (greptime_mito_memtable_field_builder_count)
-              datasource:
-                type: prometheus
-                uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-field_builders'
        - title: Region Worker Convert Requests
          type: timeseries
          description: Per-stage elapsed time for region worker to decode requests.
--- a/grafana/dashboards/metrics/standalone/dashboard.json
+++ b/grafana/dashboards/metrics/standalone/dashboard.json
--- a/grafana/dashboards/metrics/standalone/dashboard.md
+++ b/grafana/dashboards/metrics/standalone/dashboard.md
@@ -70,7 +70,6 @@
 | Inflight Flush | `greptime_mito_inflight_flush_count` | `timeseries` | Ongoing flush task count | `prometheus` | `none` | `[{{instance}}]-[{{pod}}]` |
 | Compaction Input/Output Bytes | `sum by(instance, pod) (greptime_mito_compaction_input_bytes)`<br/>`sum by(instance, pod) (greptime_mito_compaction_output_bytes)` | `timeseries` | Compaction oinput output bytes | `prometheus` | `bytes` | `[{{instance}}]-[{{pod}}]-input` |
 | Region Worker Handle Bulk Insert Requests | `histogram_quantile(0.95, sum by(le,instance, stage, pod) (rate(greptime_region_worker_handle_write_bucket[$__rate_interval])))`<br/>`sum by(instance, stage, pod) (rate(greptime_region_worker_handle_write_sum[$__rate_interval]))/sum by(instance, stage, pod) (rate(greptime_region_worker_handle_write_count[$__rate_interval]))` | `timeseries` | Per-stage elapsed time for region worker to handle bulk insert region requests. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-P95` |
-| Active Series and Field Builders Count | `sum by(instance, pod) (greptime_mito_memtable_active_series_count)`<br/>`sum by(instance, pod) (greptime_mito_memtable_field_builder_count)` | `timeseries` | Compaction oinput output bytes | `prometheus` | `none` | `[{{instance}}]-[{{pod}}]-series` |
 | Region Worker Convert Requests | `histogram_quantile(0.95, sum by(le, instance, stage, pod) (rate(greptime_datanode_convert_region_request_bucket[$__rate_interval])))`<br/>`sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_sum[$__rate_interval]))/sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_count[$__rate_interval]))` | `timeseries` | Per-stage elapsed time for region worker to decode requests. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-P95` |
 # OpenDAL
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
--- a/grafana/dashboards/metrics/standalone/dashboard.yaml
+++ b/grafana/dashboards/metrics/standalone/dashboard.yaml
@@ -612,21 +612,6 @@ groups:
                type: prometheus
                uid: ${metrics}
              legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-AVG'
-        - title: Active Series and Field Builders Count
-          type: timeseries
-          description: Compaction oinput output bytes
-          unit: none
-          queries:
-            - expr: sum by(instance, pod) (greptime_mito_memtable_active_series_count)
-              datasource:
-                type: prometheus
-                uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-series'
-            - expr: sum by(instance, pod) (greptime_mito_memtable_field_builder_count)
-              datasource:
-                type: prometheus
-                uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-field_builders'
        - title: Region Worker Convert Requests
          type: timeseries
          description: Per-stage elapsed time for region worker to decode requests.
--- a/licenserc.toml
+++ b/licenserc.toml
@@ -31,7 +31,6 @@ excludes = [
    "src/operator/src/expr_helper/trigger.rs",
    "src/sql/src/statements/create/trigger.rs",
    "src/sql/src/statements/show/trigger.rs",
-    "src/sql/src/statements/drop/trigger.rs",
    "src/sql/src/parsers/create_parser/trigger.rs",
    "src/sql/src/parsers/show_parser/trigger.rs",
 ]
--- a/src/api/src/region.rs
+++ b/src/api/src/region.rs
@@ -22,7 +22,6 @@ use greptime_proto::v1::region::RegionResponse as RegionResponseV1;
 pub struct RegionResponse {
    pub affected_rows: AffectedRows,
    pub extensions: HashMap<String, Vec<u8>>,
-    pub metadata: Vec<u8>,
 }

 impl RegionResponse {
@@ -30,7 +29,6 @@ impl RegionResponse {
        Self {
            affected_rows: region_response.affected_rows as _,
            extensions: region_response.extensions,
-            metadata: region_response.metadata,
        }
    }

@@ -39,16 +37,6 @@ impl RegionResponse {
        Self {
            affected_rows,
            extensions: Default::default(),
-            metadata: Vec::new(),
-        }
-    }
-
-    /// Creates one response with metadata.
-    pub fn from_metadata(metadata: Vec<u8>) -> Self {
-        Self {
-            affected_rows: 0,
-            extensions: Default::default(),
-            metadata,
        }
    }
 }
--- a/src/catalog/Cargo.toml
+++ b/src/catalog/Cargo.toml
@@ -17,7 +17,6 @@ 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
--- a/src/catalog/src/error.rs
+++ b/src/catalog/src/error.rs
@@ -278,25 +278,12 @@ pub enum Error {
        location: Location,
    },

-    #[snafu(display("Failed to invoke frontend services"))]
-    InvokeFrontend {
+    #[snafu(display("Failed to list frontend nodes"))]
+    ListProcess {
        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 +352,7 @@ 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
-            }
+            Error::ListProcess { source, .. } => source.status_code(),
        }
    }

--- a/src/catalog/src/kvbackend/manager.rs
+++ b/src/catalog/src/kvbackend/manager.rs
@@ -22,9 +22,7 @@ use common_catalog::consts::{
    PG_CATALOG_NAME,
 };
 use common_error::ext::BoxedError;
-use common_meta::cache::{
-    LayeredCacheRegistryRef, TableRoute, TableRouteCacheRef, ViewInfoCacheRef,
-};
+use common_meta::cache::{LayeredCacheRegistryRef, ViewInfoCacheRef};
 use common_meta::key::catalog_name::CatalogNameKey;
 use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::schema_name::SchemaNameKey;
@@ -268,68 +266,16 @@ impl CatalogManager for KvBackendCatalogManager {
        let table_cache: TableCacheRef = self.cache_registry.get().context(CacheNotFoundSnafu {
            name: "table_cache",
        })?;
-        let table_route_cache: TableRouteCacheRef =
-            self.cache_registry.get().context(CacheNotFoundSnafu {
-                name: "table_route_cache",
-            })?;
-        let table = table_cache
+        if let Some(table) = table_cache
            .get_by_ref(&TableName {
                catalog_name: catalog_name.to_string(),
                schema_name: schema_name.to_string(),
                table_name: table_name.to_string(),
            })
            .await
-            .context(GetTableCacheSnafu)?;
-
-        // Override logical table's partition key indices with physical table's.
-        if let Some(table) = &table
-            && let Some(table_route_value) = table_route_cache
-                .get(table.table_info().table_id())
-                .await
-                .context(TableMetadataManagerSnafu)?
-            && let TableRoute::Logical(logical_route) = &*table_route_value
-            && let Some(physical_table_info_value) = self
-                .table_metadata_manager
-                .table_info_manager()
-                .get(logical_route.physical_table_id())
-                .await
-                .context(TableMetadataManagerSnafu)?
+            .context(GetTableCacheSnafu)?
        {
-            let mut new_table_info = (*table.table_info()).clone();
-            // Gather all column names from the logical table
-            let logical_column_names: std::collections::HashSet<_> = new_table_info
-                .meta
-                .schema
-                .column_schemas()
-                .iter()
-                .map(|col| &col.name)
-                .collect();
-
-            // Only preserve partition key indices where the corresponding columns exist in logical table
-            new_table_info.meta.partition_key_indices = physical_table_info_value
-                .table_info
-                .meta
-                .partition_key_indices
-                .iter()
-                .filter(|&&index| {
-                    if let Some(physical_column) = physical_table_info_value
-                        .table_info
-                        .meta
-                        .schema
-                        .column_schemas
-                        .get(index)
-                    {
-                        logical_column_names.contains(&physical_column.name)
-                    } else {
-                        false
-                    }
-                })
-                .cloned()
-                .collect();
-
-            let new_table = DistTable::table(Arc::new(new_table_info));
-
-            return Ok(Some(new_table));
+            return Ok(Some(table));
        }

        if channel == Channel::Postgres {
@@ -342,7 +288,7 @@ impl CatalogManager for KvBackendCatalogManager {
            }
        }

-        Ok(table)
+        return Ok(None);
    }

    async fn tables_by_ids(
--- a/src/catalog/src/lib.rs
+++ b/src/catalog/src/lib.rs
@@ -14,7 +14,6 @@

 #![feature(assert_matches)]
 #![feature(try_blocks)]
-#![feature(let_chains)]

 use std::any::Any;
 use std::fmt::{Debug, Formatter};
--- 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
@@ -14,33 +14,24 @@

 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
-use std::fmt::{Debug, Formatter};
-use std::sync::atomic::{AtomicU32, Ordering};
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, RwLock};

-use api::v1::frontend::{KillProcessRequest, ListProcessRequest, ProcessInfo};
-use common_base::cancellation::CancellationHandle;
+use api::v1::frontend::{ListProcessRequest, ProcessInfo};
 use common_frontend::selector::{FrontendSelector, MetaClientSelector};
 use common_telemetry::{debug, info};
 use common_time::util::current_time_millis;
 use meta_client::MetaClientRef;
-use snafu::{ensure, OptionExt, ResultExt};
+use snafu::ResultExt;

 use crate::error;
-use crate::metrics::{PROCESS_KILL_COUNT, PROCESS_LIST_COUNT};

-pub type ProcessId = u32;
 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: AtomicU32,
-    /// Running process per catalog.
-    catalogs: RwLock<HashMap<String, HashMap<ProcessId, CancellableProcess>>>,
-    /// Frontend selector to locate frontend nodes.
+    next_id: AtomicU64,
+    catalogs: RwLock<HashMap<String, HashMap<u64, ProcessInfo>>>,
    frontend_selector: Option<MetaClientSelector>,
 }

@@ -59,16 +50,15 @@ impl ProcessManager {

 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,
-        query_id: Option<ProcessId>,
+        id: Option<u64>,
    ) -> Ticket {
-        let id = query_id.unwrap_or_else(|| self.next_id.fetch_add(1, Ordering::Relaxed));
+        let id = id.unwrap_or_else(|| self.next_id.fetch_add(1, Ordering::Relaxed));
        let process = ProcessInfo {
            id,
            catalog: catalog.clone(),
@@ -78,53 +68,53 @@ impl ProcessManager {
            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);
-
+            .insert(id, process);
        Ticket {
            catalog,
            manager: self.clone(),
            id,
-            cancellation_handle,
        }
    }

    /// Generates the next process id.
-    pub fn next_id(&self) -> u32 {
+    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: ProcessId) {
+    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() {
+            if o.get_mut().is_empty() {
                o.remove();
            }
        }
    }

+    pub fn deregister_all_queries(&self) {
+        self.catalogs.write().unwrap().clear();
+        info!("All queries on {} has been deregistered", self.server_addr);
+    }
+
    /// List local running processes in given catalog.
    pub fn local_processes(&self, catalog: Option<&str>) -> error::Result<Vec<ProcessInfo>> {
        let catalogs = self.catalogs.read().unwrap();
        let result = if let Some(catalog) = catalog {
            if let Some(catalogs) = catalogs.get(catalog) {
-                catalogs.values().map(|p| p.process.clone()).collect()
+                catalogs.values().cloned().collect()
            } else {
                vec![]
            }
        } else {
            catalogs
                .values()
-                .flat_map(|v| v.values().map(|p| p.process.clone()))
+                .flat_map(|v| v.values().cloned())
                .collect()
        };
        Ok(result)
@@ -139,14 +129,14 @@ impl ProcessManager {
            let frontends = remote_frontend_selector
                .select(|node| node.peer.addr != self.server_addr)
                .await
-                .context(error::InvokeFrontendSnafu)?;
+                .context(error::ListProcessSnafu)?;
            for mut f in frontends {
                processes.extend(
                    f.list_process(ListProcessRequest {
                        catalog: catalog.unwrap_or_default().to_string(),
                    })
                    .await
-                    .context(error::InvokeFrontendSnafu)?
+                    .context(error::ListProcessSnafu)?
                    .processes,
                );
            }
@@ -154,69 +144,12 @@ impl ProcessManager {
        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: ProcessId,
-    ) -> error::Result<bool> {
-        if server_addr == self.server_addr {
-            self.kill_local_process(catalog, id).await
-        } 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)
-        }
-    }
-
-    /// Kills local query with provided catalog and id.
-    pub async fn kill_local_process(&self, catalog: String, id: ProcessId) -> error::Result<bool> {
-        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)
-        }
-    }
 }

 pub struct Ticket {
    pub(crate) catalog: String,
    pub(crate) manager: ProcessManagerRef,
-    pub(crate) id: ProcessId,
-    pub cancellation_handle: Arc<CancellationHandle>,
+    pub(crate) id: u64,
 }

 impl Drop for Ticket {
@@ -226,37 +159,6 @@ impl Drop for Ticket {
    }
 }

-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;
@@ -283,212 +185,4 @@ mod tests {
        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<u32> = 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,7 @@ mod information_memory_table;
 pub mod key_column_usage;
 mod partitions;
 mod procedure_info;
-pub mod process_list;
+mod process_list;
 pub mod region_peers;
 mod region_statistics;
 mod runtime_metrics;
--- a/src/catalog/src/system_schema/information_schema/process_list.rs
+++ b/src/catalog/src/system_schema/information_schema/process_list.rs
@@ -39,14 +39,14 @@ use crate::process_manager::ProcessManagerRef;
 use crate::system_schema::information_schema::InformationTable;

 /// Column names of `information_schema.process_list`
-pub const ID: &str = "id";
-pub const CATALOG: &str = "catalog";
-pub const SCHEMAS: &str = "schemas";
-pub const QUERY: &str = "query";
-pub const CLIENT: &str = "client";
-pub const FRONTEND: &str = "frontend";
-pub const START_TIMESTAMP: &str = "start_timestamp";
-pub const ELAPSED_TIME: &str = "elapsed_time";
+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.
--- a/src/cli/Cargo.toml
+++ b/src/cli/Cargo.toml
@@ -16,7 +16,6 @@ mysql_kvbackend = ["common-meta/mysql_kvbackend", "meta-srv/mysql_kvbackend"]
 workspace = true

 [dependencies]
-async-stream.workspace = true
 async-trait.workspace = true
 auth.workspace = true
 base64.workspace = true
@@ -51,7 +50,6 @@ meta-client.workspace = true
 meta-srv.workspace = true
 nu-ansi-term = "0.46"
 object-store.workspace = true
-operator.workspace = true
 query.workspace = true
 rand.workspace = true
 reqwest.workspace = true
@@ -67,7 +65,6 @@ tokio.workspace = true
 tracing-appender.workspace = true

 [dev-dependencies]
-common-meta = { workspace = true, features = ["testing"] }
 common-version.workspace = true
 serde.workspace = true
 tempfile.workspace = true
--- a/src/cli/src/error.rs
+++ b/src/cli/src/error.rs
@@ -17,10 +17,8 @@ use std::any::Any;
 use common_error::ext::{BoxedError, ErrorExt};
 use common_error::status_code::StatusCode;
 use common_macro::stack_trace_debug;
-use common_meta::peer::Peer;
 use object_store::Error as ObjectStoreError;
 use snafu::{Location, Snafu};
-use store_api::storage::TableId;

 #[derive(Snafu)]
 #[snafu(visibility(pub))]
@@ -75,20 +73,6 @@ pub enum Error {
        source: common_meta::error::Error,
    },

-    #[snafu(display("Failed to get table metadata"))]
-    TableMetadata {
-        #[snafu(implicit)]
-        location: Location,
-        source: common_meta::error::Error,
-    },
-
-    #[snafu(display("Unexpected error: {}", msg))]
-    Unexpected {
-        msg: String,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
    #[snafu(display("Missing config, msg: {}", msg))]
    MissingConfig {
        msg: String,
@@ -238,13 +222,6 @@ pub enum Error {
        location: Location,
    },

-    #[snafu(display("Table not found: {table_id}"))]
-    TableNotFound {
-        table_id: TableId,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
    #[snafu(display("OpenDAL operator failed"))]
    OpenDal {
        #[snafu(implicit)]
@@ -290,29 +267,6 @@ pub enum Error {
        #[snafu(implicit)]
        location: Location,
    },
-
-    #[snafu(display("Failed to init backend"))]
-    InitBackend {
-        #[snafu(implicit)]
-        location: Location,
-        #[snafu(source)]
-        error: ObjectStoreError,
-    },
-
-    #[snafu(display("Covert column schemas to defs failed"))]
-    CovertColumnSchemasToDefs {
-        #[snafu(implicit)]
-        location: Location,
-        source: operator::error::Error,
-    },
-
-    #[snafu(display("Failed to send request to datanode: {}", peer))]
-    SendRequestToDatanode {
-        peer: Peer,
-        #[snafu(implicit)]
-        location: Location,
-        source: common_meta::error::Error,
-    },
 }

 pub type Result<T> = std::result::Result<T, Error>;
@@ -320,9 +274,9 @@ pub type Result<T> = std::result::Result<T, Error>;
 impl ErrorExt for Error {
    fn status_code(&self) -> StatusCode {
        match self {
-            Error::InitMetadata { source, .. }
-            | Error::InitDdlManager { source, .. }
-            | Error::TableMetadata { source, .. } => source.status_code(),
+            Error::InitMetadata { source, .. } | Error::InitDdlManager { source, .. } => {
+                source.status_code()
+            }

            Error::MissingConfig { .. }
            | Error::LoadLayeredConfig { .. }
@@ -336,9 +290,6 @@ impl ErrorExt for Error {
            | Error::InvalidArguments { .. }
            | Error::ParseProxyOpts { .. } => StatusCode::InvalidArguments,

-            Error::CovertColumnSchemasToDefs { source, .. } => source.status_code(),
-            Error::SendRequestToDatanode { source, .. } => source.status_code(),
-
            Error::StartProcedureManager { source, .. }
            | Error::StopProcedureManager { source, .. } => source.status_code(),
            Error::StartWalOptionsAllocator { source, .. } => source.status_code(),
@@ -346,7 +297,6 @@ impl ErrorExt for Error {
            Error::ParseSql { source, .. } | Error::PlanStatement { source, .. } => {
                source.status_code()
            }
-            Error::Unexpected { .. } => StatusCode::Unexpected,

            Error::SerdeJson { .. }
            | Error::FileIo { .. }
@@ -355,7 +305,7 @@ impl ErrorExt for Error {
            | Error::BuildClient { .. } => StatusCode::Unexpected,

            Error::Other { source, .. } => source.status_code(),
-            Error::OpenDal { .. } | Error::InitBackend { .. } => StatusCode::Internal,
+            Error::OpenDal { .. } => StatusCode::Internal,
            Error::S3ConfigNotSet { .. }
            | Error::OutputDirNotSet { .. }
            | Error::EmptyStoreAddrs { .. } => StatusCode::InvalidArguments,
@@ -364,7 +314,6 @@ impl ErrorExt for Error {

            Error::CacheRequired { .. } | Error::BuildCacheRegistry { .. } => StatusCode::Internal,
            Error::MetaClientInit { source, .. } => source.status_code(),
-            Error::TableNotFound { .. } => StatusCode::TableNotFound,
            Error::SchemaNotFound { .. } => StatusCode::DatabaseNotFound,
        }
    }
--- a/src/cli/src/metadata.rs
+++ b/src/cli/src/metadata.rs
@@ -14,39 +14,29 @@

 mod common;
 mod control;
-mod repair;
 mod snapshot;
-mod utils;

 use clap::Subcommand;
 use common_error::ext::BoxedError;

-use crate::metadata::control::{DelCommand, GetCommand};
-use crate::metadata::repair::RepairLogicalTablesCommand;
+use crate::metadata::control::ControlCommand;
 use crate::metadata::snapshot::SnapshotCommand;
 use crate::Tool;

-/// Command for managing metadata operations,
-/// including saving and restoring metadata snapshots,
-/// controlling metadata operations, and diagnosing and repairing metadata.
+/// 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)]
-    Get(GetCommand),
-    #[clap(subcommand)]
-    Del(DelCommand),
-    RepairLogicalTables(RepairLogicalTablesCommand),
+    Control(ControlCommand),
 }

 impl MetadataCommand {
    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
        match self {
            MetadataCommand::Snapshot(cmd) => cmd.build().await,
-            MetadataCommand::RepairLogicalTables(cmd) => cmd.build().await,
-            MetadataCommand::Get(cmd) => cmd.build().await,
-            MetadataCommand::Del(cmd) => cmd.build().await,
+            MetadataCommand::Control(cmd) => cmd.build().await,
        }
    }
 }
--- a/src/cli/src/metadata/control.rs
+++ b/src/cli/src/metadata/control.rs
@@ -12,11 +12,27 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-mod del;
 mod get;
-#[cfg(test)]
-mod test_utils;
 mod utils;

-pub(crate) use del::DelCommand;
-pub(crate) use get::GetCommand;
+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/del.rs
+++ b/src/cli/src/metadata/control/del.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 key;
-mod table;
-
-use clap::Subcommand;
-use common_error::ext::BoxedError;
-
-use crate::metadata::control::del::key::DelKeyCommand;
-use crate::metadata::control::del::table::DelTableCommand;
-use crate::Tool;
-
-/// The prefix of the tombstone keys.
-pub(crate) const CLI_TOMBSTONE_PREFIX: &str = "__cli_tombstone/";
-
-/// Subcommand for deleting metadata from the metadata store.
-#[derive(Subcommand)]
-pub enum DelCommand {
-    Key(DelKeyCommand),
-    Table(DelTableCommand),
-}
-
-impl DelCommand {
-    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
-        match self {
-            DelCommand::Key(cmd) => cmd.build().await,
-            DelCommand::Table(cmd) => cmd.build().await,
-        }
-    }
-}
--- a/src/cli/src/metadata/control/del/key.rs
+++ b/src/cli/src/metadata/control/del/key.rs
@@ -1,132 +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 async_trait::async_trait;
-use clap::Parser;
-use common_error::ext::BoxedError;
-use common_meta::key::tombstone::TombstoneManager;
-use common_meta::kv_backend::KvBackendRef;
-use common_meta::rpc::store::RangeRequest;
-
-use crate::metadata::common::StoreConfig;
-use crate::metadata::control::del::CLI_TOMBSTONE_PREFIX;
-use crate::Tool;
-
-/// Delete key-value pairs logically from the metadata store.
-#[derive(Debug, Default, Parser)]
-pub struct DelKeyCommand {
-    /// The key to delete from the metadata store.
-    key: String,
-
-    /// Delete key-value pairs with the given prefix.
-    #[clap(long)]
-    prefix: bool,
-
-    #[clap(flatten)]
-    store: StoreConfig,
-}
-
-impl DelKeyCommand {
-    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
-        let kv_backend = self.store.build().await?;
-        Ok(Box::new(DelKeyTool {
-            key: self.key.to_string(),
-            prefix: self.prefix,
-            key_deleter: KeyDeleter::new(kv_backend),
-        }))
-    }
-}
-
-struct KeyDeleter {
-    kv_backend: KvBackendRef,
-    tombstone_manager: TombstoneManager,
-}
-
-impl KeyDeleter {
-    fn new(kv_backend: KvBackendRef) -> Self {
-        Self {
-            kv_backend: kv_backend.clone(),
-            tombstone_manager: TombstoneManager::new_with_prefix(kv_backend, CLI_TOMBSTONE_PREFIX),
-        }
-    }
-
-    async fn delete(&self, key: &str, prefix: bool) -> Result<usize, BoxedError> {
-        let mut req = RangeRequest::default().with_keys_only();
-        if prefix {
-            req = req.with_prefix(key.as_bytes());
-        } else {
-            req = req.with_key(key.as_bytes());
-        }
-        let resp = self.kv_backend.range(req).await.map_err(BoxedError::new)?;
-        let keys = resp.kvs.iter().map(|kv| kv.key.clone()).collect::<Vec<_>>();
-        self.tombstone_manager
-            .create(keys)
-            .await
-            .map_err(BoxedError::new)
-    }
-}
-
-struct DelKeyTool {
-    key: String,
-    prefix: bool,
-    key_deleter: KeyDeleter,
-}
-
-#[async_trait]
-impl Tool for DelKeyTool {
-    async fn do_work(&self) -> Result<(), BoxedError> {
-        let deleted = self.key_deleter.delete(&self.key, self.prefix).await?;
-        // Print the number of deleted keys.
-        println!("{}", deleted);
-        Ok(())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::sync::Arc;
-
-    use common_meta::kv_backend::chroot::ChrootKvBackend;
-    use common_meta::kv_backend::memory::MemoryKvBackend;
-    use common_meta::kv_backend::{KvBackend, KvBackendRef};
-    use common_meta::rpc::store::RangeRequest;
-
-    use crate::metadata::control::del::key::KeyDeleter;
-    use crate::metadata::control::del::CLI_TOMBSTONE_PREFIX;
-    use crate::metadata::control::test_utils::put_key;
-
-    #[tokio::test]
-    async fn test_delete_keys() {
-        let kv_backend = Arc::new(MemoryKvBackend::new()) as KvBackendRef;
-        let key_deleter = KeyDeleter::new(kv_backend.clone());
-        put_key(&kv_backend, "foo", "bar").await;
-        put_key(&kv_backend, "foo/bar", "baz").await;
-        put_key(&kv_backend, "foo/baz", "qux").await;
-        let deleted = key_deleter.delete("foo", true).await.unwrap();
-        assert_eq!(deleted, 3);
-        let deleted = key_deleter.delete("foo/bar", false).await.unwrap();
-        assert_eq!(deleted, 0);
-
-        let chroot = ChrootKvBackend::new(CLI_TOMBSTONE_PREFIX.as_bytes().to_vec(), kv_backend);
-        let req = RangeRequest::default().with_prefix(b"foo");
-        let resp = chroot.range(req).await.unwrap();
-        assert_eq!(resp.kvs.len(), 3);
-        assert_eq!(resp.kvs[0].key, b"foo");
-        assert_eq!(resp.kvs[0].value, b"bar");
-        assert_eq!(resp.kvs[1].key, b"foo/bar");
-        assert_eq!(resp.kvs[1].value, b"baz");
-        assert_eq!(resp.kvs[2].key, b"foo/baz");
-        assert_eq!(resp.kvs[2].value, b"qux");
-    }
-}
--- a/src/cli/src/metadata/control/del/table.rs
+++ b/src/cli/src/metadata/control/del/table.rs
@@ -1,235 +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 async_trait::async_trait;
-use clap::Parser;
-use client::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
-use common_catalog::format_full_table_name;
-use common_error::ext::BoxedError;
-use common_meta::ddl::utils::get_region_wal_options;
-use common_meta::key::table_name::TableNameManager;
-use common_meta::key::TableMetadataManager;
-use common_meta::kv_backend::KvBackendRef;
-use store_api::storage::TableId;
-
-use crate::error::{InvalidArgumentsSnafu, TableNotFoundSnafu};
-use crate::metadata::common::StoreConfig;
-use crate::metadata::control::del::CLI_TOMBSTONE_PREFIX;
-use crate::metadata::control::utils::get_table_id_by_name;
-use crate::Tool;
-
-/// Delete table metadata logically from the metadata store.
-#[derive(Debug, Default, Parser)]
-pub struct DelTableCommand {
-    /// The table id to delete from the metadata store.
-    #[clap(long)]
-    table_id: Option<u32>,
-
-    /// The table name to delete from the metadata store.
-    #[clap(long)]
-    table_name: Option<String>,
-
-    /// The schema name of the table.
-    #[clap(long, default_value = DEFAULT_SCHEMA_NAME)]
-    schema_name: String,
-
-    /// The catalog name of the table.
-    #[clap(long, default_value = DEFAULT_CATALOG_NAME)]
-    catalog_name: String,
-
-    #[clap(flatten)]
-    store: StoreConfig,
-}
-
-impl DelTableCommand {
-    fn validate(&self) -> Result<(), BoxedError> {
-        if matches!(
-            (&self.table_id, &self.table_name),
-            (Some(_), Some(_)) | (None, None)
-        ) {
-            return Err(BoxedError::new(
-                InvalidArgumentsSnafu {
-                    msg: "You must specify either --table-id or --table-name.",
-                }
-                .build(),
-            ));
-        }
-        Ok(())
-    }
-}
-
-impl DelTableCommand {
-    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
-        self.validate()?;
-        let kv_backend = self.store.build().await?;
-        Ok(Box::new(DelTableTool {
-            table_id: self.table_id,
-            table_name: self.table_name.clone(),
-            schema_name: self.schema_name.clone(),
-            catalog_name: self.catalog_name.clone(),
-            table_name_manager: TableNameManager::new(kv_backend.clone()),
-            table_metadata_deleter: TableMetadataDeleter::new(kv_backend),
-        }))
-    }
-}
-
-struct DelTableTool {
-    table_id: Option<u32>,
-    table_name: Option<String>,
-    schema_name: String,
-    catalog_name: String,
-    table_name_manager: TableNameManager,
-    table_metadata_deleter: TableMetadataDeleter,
-}
-
-#[async_trait]
-impl Tool for DelTableTool {
-    async fn do_work(&self) -> Result<(), BoxedError> {
-        let table_id = if let Some(table_name) = &self.table_name {
-            let catalog_name = &self.catalog_name;
-            let schema_name = &self.schema_name;
-
-            let Some(table_id) = get_table_id_by_name(
-                &self.table_name_manager,
-                catalog_name,
-                schema_name,
-                table_name,
-            )
-            .await?
-            else {
-                println!(
-                    "Table({}) not found",
-                    format_full_table_name(catalog_name, schema_name, table_name)
-                );
-                return Ok(());
-            };
-            table_id
-        } else {
-            // Safety: we have validated that table_id or table_name is not None
-            self.table_id.unwrap()
-        };
-        self.table_metadata_deleter.delete(table_id).await?;
-        println!("Table({}) deleted", table_id);
-
-        Ok(())
-    }
-}
-
-struct TableMetadataDeleter {
-    table_metadata_manager: TableMetadataManager,
-}
-
-impl TableMetadataDeleter {
-    fn new(kv_backend: KvBackendRef) -> Self {
-        Self {
-            table_metadata_manager: TableMetadataManager::new_with_custom_tombstone_prefix(
-                kv_backend,
-                CLI_TOMBSTONE_PREFIX,
-            ),
-        }
-    }
-
-    async fn delete(&self, table_id: TableId) -> Result<(), BoxedError> {
-        let (table_info, table_route) = self
-            .table_metadata_manager
-            .get_full_table_info(table_id)
-            .await
-            .map_err(BoxedError::new)?;
-        let Some(table_info) = table_info else {
-            return Err(BoxedError::new(TableNotFoundSnafu { table_id }.build()));
-        };
-        let Some(table_route) = table_route else {
-            return Err(BoxedError::new(TableNotFoundSnafu { table_id }.build()));
-        };
-        let physical_table_id = self
-            .table_metadata_manager
-            .table_route_manager()
-            .get_physical_table_id(table_id)
-            .await
-            .map_err(BoxedError::new)?;
-
-        let table_name = table_info.table_name();
-        let region_wal_options = get_region_wal_options(
-            &self.table_metadata_manager,
-            &table_route,
-            physical_table_id,
-        )
-        .await
-        .map_err(BoxedError::new)?;
-
-        self.table_metadata_manager
-            .delete_table_metadata(table_id, &table_name, &table_route, &region_wal_options)
-            .await
-            .map_err(BoxedError::new)?;
-        Ok(())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::collections::HashMap;
-    use std::sync::Arc;
-
-    use common_error::ext::ErrorExt;
-    use common_error::status_code::StatusCode;
-    use common_meta::key::table_route::TableRouteValue;
-    use common_meta::key::TableMetadataManager;
-    use common_meta::kv_backend::chroot::ChrootKvBackend;
-    use common_meta::kv_backend::memory::MemoryKvBackend;
-    use common_meta::kv_backend::{KvBackend, KvBackendRef};
-    use common_meta::rpc::store::RangeRequest;
-
-    use crate::metadata::control::del::table::TableMetadataDeleter;
-    use crate::metadata::control::del::CLI_TOMBSTONE_PREFIX;
-    use crate::metadata::control::test_utils::prepare_physical_table_metadata;
-
-    #[tokio::test]
-    async fn test_delete_table_not_found() {
-        let kv_backend = Arc::new(MemoryKvBackend::new()) as KvBackendRef;
-
-        let table_metadata_deleter = TableMetadataDeleter::new(kv_backend);
-        let table_id = 1;
-        let err = table_metadata_deleter.delete(table_id).await.unwrap_err();
-        assert_eq!(err.status_code(), StatusCode::TableNotFound);
-    }
-
-    #[tokio::test]
-    async fn test_delete_table_metadata() {
-        let kv_backend = Arc::new(MemoryKvBackend::new());
-        let table_metadata_manager = TableMetadataManager::new(kv_backend.clone());
-        let table_id = 1024;
-        let (table_info, table_route) = prepare_physical_table_metadata("my_table", table_id).await;
-        table_metadata_manager
-            .create_table_metadata(
-                table_info,
-                TableRouteValue::Physical(table_route),
-                HashMap::new(),
-            )
-            .await
-            .unwrap();
-
-        let total_keys = kv_backend.len();
-        assert!(total_keys > 0);
-
-        let table_metadata_deleter = TableMetadataDeleter::new(kv_backend.clone());
-        table_metadata_deleter.delete(table_id).await.unwrap();
-
-        // Check the tombstone keys are deleted
-        let chroot =
-            ChrootKvBackend::new(CLI_TOMBSTONE_PREFIX.as_bytes().to_vec(), kv_backend.clone());
-        let req = RangeRequest::default().with_range(vec![0], vec![0]);
-        let resp = chroot.range(req).await.unwrap();
-        assert_eq!(resp.kvs.len(), total_keys);
-    }
-}
--- a/src/cli/src/metadata/control/get.rs
+++ b/src/cli/src/metadata/control/get.rs
@@ -20,6 +20,7 @@ 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;
@@ -29,10 +30,10 @@ use futures::TryStreamExt;

 use crate::error::InvalidArgumentsSnafu;
 use crate::metadata::common::StoreConfig;
-use crate::metadata::control::utils::{decode_key_value, get_table_id_by_name, json_fromatter};
+use crate::metadata::control::utils::{decode_key_value, json_fromatter};
 use crate::Tool;

-/// Getting metadata from metadata store.
+/// Subcommand for get command.
 #[derive(Subcommand)]
 pub enum GetCommand {
    Key(GetKeyCommand),
@@ -51,7 +52,7 @@ impl GetCommand {
 /// Get key-value pairs from the metadata store.
 #[derive(Debug, Default, Parser)]
 pub struct GetKeyCommand {
-    /// The key to get from the metadata store.
+    /// The key to get from the metadata store. If empty, returns all key-value pairs.
    #[clap(default_value = "")]
    key: String,

@@ -129,12 +130,8 @@ pub struct GetTableCommand {
    table_name: Option<String>,

    /// The schema name of the table.
-    #[clap(long, default_value = DEFAULT_SCHEMA_NAME)]
-    schema_name: String,
-
-    /// The catalog name of the table.
-    #[clap(long, default_value = DEFAULT_CATALOG_NAME)]
-    catalog_name: String,
+    #[clap(long)]
+    schema_name: Option<String>,

    /// Pretty print the output.
    #[clap(long, default_value = "false")]
@@ -146,10 +143,7 @@ pub struct GetTableCommand {

 impl GetTableCommand {
    pub fn validate(&self) -> Result<(), BoxedError> {
-        if matches!(
-            (&self.table_id, &self.table_name),
-            (Some(_), Some(_)) | (None, None)
-        ) {
+        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.",
@@ -165,8 +159,7 @@ struct GetTableTool {
    kvbackend: KvBackendRef,
    table_id: Option<u32>,
    table_name: Option<String>,
-    schema_name: String,
-    catalog_name: String,
+    schema_name: Option<String>,
    pretty: bool,
 }

@@ -179,20 +172,23 @@ impl Tool for GetTableTool {
        let table_route_manager = table_metadata_manager.table_route_manager();

        let table_id = if let Some(table_name) = &self.table_name {
-            let catalog_name = &self.catalog_name;
-            let schema_name = &self.schema_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_id) =
-                get_table_id_by_name(table_name_manager, catalog_name, schema_name, table_name)
-                    .await?
+            let Some(table_name) = table_name_manager.get(key).await.map_err(BoxedError::new)?
            else {
                println!(
                    "Table({}) not found",
-                    format_full_table_name(catalog_name, schema_name, table_name)
+                    format_full_table_name(&catalog, &schema_name, table_name)
                );
                return Ok(());
            };
-            table_id
+
+            table_name.table_id()
        } else {
            // Safety: we have validated that table_id or table_name is not None
            self.table_id.unwrap()
@@ -240,7 +236,6 @@ impl GetTableCommand {
            table_id: self.table_id,
            table_name: self.table_name.clone(),
            schema_name: self.schema_name.clone(),
-            catalog_name: self.catalog_name.clone(),
            pretty: self.pretty,
        }))
    }
--- a/src/cli/src/metadata/control/test_utils.rs
+++ b/src/cli/src/metadata/control/test_utils.rs
@@ -1,51 +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::ddl::test_util::test_create_physical_table_task;
-use common_meta::key::table_route::PhysicalTableRouteValue;
-use common_meta::kv_backend::KvBackendRef;
-use common_meta::peer::Peer;
-use common_meta::rpc::router::{Region, RegionRoute};
-use common_meta::rpc::store::PutRequest;
-use store_api::storage::{RegionId, TableId};
-use table::metadata::RawTableInfo;
-
-/// Puts a key-value pair into the kv backend.
-pub async fn put_key(kv_backend: &KvBackendRef, key: &str, value: &str) {
-    let put_req = PutRequest::new()
-        .with_key(key.as_bytes())
-        .with_value(value.as_bytes());
-    kv_backend.put(put_req).await.unwrap();
-}
-
-/// Prepares the physical table metadata for testing.
-///
-/// Returns the table info and the table route.
-pub async fn prepare_physical_table_metadata(
-    table_name: &str,
-    table_id: TableId,
-) -> (RawTableInfo, PhysicalTableRouteValue) {
-    let mut create_physical_table_task = test_create_physical_table_task(table_name);
-    let table_route = PhysicalTableRouteValue::new(vec![RegionRoute {
-        region: Region {
-            id: RegionId::new(table_id, 1),
-            ..Default::default()
-        },
-        leader_peer: Some(Peer::empty(1)),
-        ..Default::default()
-    }]);
-    create_physical_table_task.set_table_id(table_id);
-
-    (create_physical_table_task.table_info, table_route)
-}
--- a/src/cli/src/metadata/control/utils.rs
+++ b/src/cli/src/metadata/control/utils.rs
@@ -12,12 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use common_error::ext::BoxedError;
 use common_meta::error::Result as CommonMetaResult;
-use common_meta::key::table_name::{TableNameKey, TableNameManager};
 use common_meta::rpc::KeyValue;
 use serde::Serialize;
-use store_api::storage::TableId;

 /// Decodes a key-value pair into a string.
 pub fn decode_key_value(kv: KeyValue) -> CommonMetaResult<(String, String)> {
@@ -37,21 +34,3 @@ where
        serde_json::to_string(value).unwrap()
    }
 }
-
-/// Gets the table id by table name.
-pub async fn get_table_id_by_name(
-    table_name_manager: &TableNameManager,
-    catalog_name: &str,
-    schema_name: &str,
-    table_name: &str,
-) -> Result<Option<TableId>, BoxedError> {
-    let table_name_key = TableNameKey::new(catalog_name, schema_name, table_name);
-    let Some(table_name_value) = table_name_manager
-        .get(table_name_key)
-        .await
-        .map_err(BoxedError::new)?
-    else {
-        return Ok(None);
-    };
-    Ok(Some(table_name_value.table_id()))
-}
--- a/src/cli/src/metadata/repair.rs
+++ b/src/cli/src/metadata/repair.rs
@@ -1,369 +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 alter_table;
-mod create_table;
-
-use std::sync::Arc;
-use std::time::Duration;
-
-use async_trait::async_trait;
-use clap::Parser;
-use client::api::v1::CreateTableExpr;
-use client::client_manager::NodeClients;
-use client::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
-use common_error::ext::{BoxedError, ErrorExt};
-use common_error::status_code::StatusCode;
-use common_grpc::channel_manager::ChannelConfig;
-use common_meta::error::Error as CommonMetaError;
-use common_meta::key::TableMetadataManager;
-use common_meta::kv_backend::KvBackendRef;
-use common_meta::node_manager::NodeManagerRef;
-use common_meta::peer::Peer;
-use common_meta::rpc::router::{find_leaders, RegionRoute};
-use common_telemetry::{error, info, warn};
-use futures::TryStreamExt;
-use snafu::{ensure, ResultExt};
-use store_api::storage::TableId;
-
-use crate::error::{
-    InvalidArgumentsSnafu, Result, SendRequestToDatanodeSnafu, TableMetadataSnafu, UnexpectedSnafu,
-};
-use crate::metadata::common::StoreConfig;
-use crate::metadata::utils::{FullTableMetadata, IteratorInput, TableMetadataIterator};
-use crate::Tool;
-
-/// Repair metadata of logical tables.
-#[derive(Debug, Default, Parser)]
-pub struct RepairLogicalTablesCommand {
-    /// The names of the tables to repair.
-    #[clap(long, value_delimiter = ',', alias = "table-name")]
-    table_names: Vec<String>,
-
-    /// The id of the table to repair.
-    #[clap(long, value_delimiter = ',', alias = "table-id")]
-    table_ids: Vec<TableId>,
-
-    /// The schema of the tables to repair.
-    #[clap(long, default_value = DEFAULT_SCHEMA_NAME)]
-    schema_name: String,
-
-    /// The catalog of the tables to repair.
-    #[clap(long, default_value = DEFAULT_CATALOG_NAME)]
-    catalog_name: String,
-
-    /// Whether to fail fast if any repair operation fails.
-    #[clap(long)]
-    fail_fast: bool,
-
-    #[clap(flatten)]
-    store: StoreConfig,
-
-    /// The timeout for the client to operate the datanode.
-    #[clap(long, default_value_t = 30)]
-    client_timeout_secs: u64,
-
-    /// The timeout for the client to connect to the datanode.
-    #[clap(long, default_value_t = 3)]
-    client_connect_timeout_secs: u64,
-}
-
-impl RepairLogicalTablesCommand {
-    fn validate(&self) -> Result<()> {
-        ensure!(
-            !self.table_names.is_empty() || !self.table_ids.is_empty(),
-            InvalidArgumentsSnafu {
-                msg: "You must specify --table-names or --table-ids.",
-            }
-        );
-        Ok(())
-    }
-}
-
-impl RepairLogicalTablesCommand {
-    pub async fn build(&self) -> std::result::Result<Box<dyn Tool>, BoxedError> {
-        self.validate().map_err(BoxedError::new)?;
-        let kv_backend = self.store.build().await?;
-        let node_client_channel_config = ChannelConfig::new()
-            .timeout(Duration::from_secs(self.client_timeout_secs))
-            .connect_timeout(Duration::from_secs(self.client_connect_timeout_secs));
-        let node_manager = Arc::new(NodeClients::new(node_client_channel_config));
-
-        Ok(Box::new(RepairTool {
-            table_names: self.table_names.clone(),
-            table_ids: self.table_ids.clone(),
-            schema_name: self.schema_name.clone(),
-            catalog_name: self.catalog_name.clone(),
-            fail_fast: self.fail_fast,
-            kv_backend,
-            node_manager,
-        }))
-    }
-}
-
-struct RepairTool {
-    table_names: Vec<String>,
-    table_ids: Vec<TableId>,
-    schema_name: String,
-    catalog_name: String,
-    fail_fast: bool,
-    kv_backend: KvBackendRef,
-    node_manager: NodeManagerRef,
-}
-
-#[async_trait]
-impl Tool for RepairTool {
-    async fn do_work(&self) -> std::result::Result<(), BoxedError> {
-        self.repair_tables().await.map_err(BoxedError::new)
-    }
-}
-
-impl RepairTool {
-    fn generate_iterator_input(&self) -> Result<IteratorInput> {
-        if !self.table_names.is_empty() {
-            let table_names = &self.table_names;
-            let catalog = &self.catalog_name;
-            let schema_name = &self.schema_name;
-
-            let table_names = table_names
-                .iter()
-                .map(|table_name| {
-                    (
-                        catalog.to_string(),
-                        schema_name.to_string(),
-                        table_name.to_string(),
-                    )
-                })
-                .collect::<Vec<_>>();
-            return Ok(IteratorInput::new_table_names(table_names));
-        } else if !self.table_ids.is_empty() {
-            return Ok(IteratorInput::new_table_ids(self.table_ids.clone()));
-        };
-
-        InvalidArgumentsSnafu {
-            msg: "You must specify --table-names or --table-id.",
-        }
-        .fail()
-    }
-
-    async fn repair_tables(&self) -> Result<()> {
-        let input = self.generate_iterator_input()?;
-        let mut table_metadata_iterator =
-            Box::pin(TableMetadataIterator::new(self.kv_backend.clone(), input).into_stream());
-        let table_metadata_manager = TableMetadataManager::new(self.kv_backend.clone());
-
-        let mut skipped_table = 0;
-        let mut success_table = 0;
-        while let Some(full_table_metadata) = table_metadata_iterator.try_next().await? {
-            let full_table_name = full_table_metadata.full_table_name();
-            if !full_table_metadata.is_metric_engine() {
-                warn!(
-                    "Skipping repair for non-metric engine table: {}",
-                    full_table_name
-                );
-                skipped_table += 1;
-                continue;
-            }
-
-            if full_table_metadata.is_physical_table() {
-                warn!("Skipping repair for physical table: {}", full_table_name);
-                skipped_table += 1;
-                continue;
-            }
-
-            let (physical_table_id, physical_table_route) = table_metadata_manager
-                .table_route_manager()
-                .get_physical_table_route(full_table_metadata.table_id)
-                .await
-                .context(TableMetadataSnafu)?;
-
-            if let Err(err) = self
-                .repair_table(
-                    &full_table_metadata,
-                    physical_table_id,
-                    &physical_table_route.region_routes,
-                )
-                .await
-            {
-                error!(
-                    err;
-                    "Failed to repair table: {}, skipped table: {}",
-                    full_table_name,
-                    skipped_table,
-                );
-
-                if self.fail_fast {
-                    return Err(err);
-                }
-            } else {
-                success_table += 1;
-            }
-        }
-
-        info!(
-            "Repair logical tables result: {} tables repaired, {} tables skipped",
-            success_table, skipped_table
-        );
-
-        Ok(())
-    }
-
-    async fn alter_table_on_datanodes(
-        &self,
-        full_table_metadata: &FullTableMetadata,
-        physical_region_routes: &[RegionRoute],
-    ) -> Result<Vec<(Peer, CommonMetaError)>> {
-        let logical_table_id = full_table_metadata.table_id;
-        let alter_table_expr = alter_table::generate_alter_table_expr_for_all_columns(
-            &full_table_metadata.table_info,
-        )?;
-        let node_manager = self.node_manager.clone();
-
-        let mut failed_peers = Vec::new();
-        info!(
-            "Sending alter table requests to all datanodes for table: {}, number of regions:{}.",
-            full_table_metadata.full_table_name(),
-            physical_region_routes.len()
-        );
-        let leaders = find_leaders(physical_region_routes);
-        for peer in &leaders {
-            let alter_table_request = alter_table::make_alter_region_request_for_peer(
-                logical_table_id,
-                &alter_table_expr,
-                full_table_metadata.table_info.ident.version,
-                peer,
-                physical_region_routes,
-            )?;
-            let datanode = node_manager.datanode(peer).await;
-            if let Err(err) = datanode.handle(alter_table_request).await {
-                failed_peers.push((peer.clone(), err));
-            }
-        }
-
-        Ok(failed_peers)
-    }
-
-    async fn create_table_on_datanode(
-        &self,
-        create_table_expr: &CreateTableExpr,
-        logical_table_id: TableId,
-        physical_table_id: TableId,
-        peer: &Peer,
-        physical_region_routes: &[RegionRoute],
-    ) -> Result<()> {
-        let node_manager = self.node_manager.clone();
-        let datanode = node_manager.datanode(peer).await;
-        let create_table_request = create_table::make_create_region_request_for_peer(
-            logical_table_id,
-            physical_table_id,
-            create_table_expr,
-            peer,
-            physical_region_routes,
-        )?;
-
-        datanode
-            .handle(create_table_request)
-            .await
-            .with_context(|_| SendRequestToDatanodeSnafu { peer: peer.clone() })?;
-
-        Ok(())
-    }
-
-    async fn repair_table(
-        &self,
-        full_table_metadata: &FullTableMetadata,
-        physical_table_id: TableId,
-        physical_region_routes: &[RegionRoute],
-    ) -> Result<()> {
-        let full_table_name = full_table_metadata.full_table_name();
-        // First we sends alter table requests to all datanodes with all columns.
-        let failed_peers = self
-            .alter_table_on_datanodes(full_table_metadata, physical_region_routes)
-            .await?;
-
-        if failed_peers.is_empty() {
-            info!(
-                "All alter table requests sent successfully for table: {}",
-                full_table_name
-            );
-            return Ok(());
-        }
-        warn!(
-            "Sending alter table requests to datanodes for table: {} failed for the datanodes: {:?}",
-            full_table_name,
-            failed_peers.iter().map(|(peer, _)| peer.id).collect::<Vec<_>>()
-        );
-
-        let create_table_expr =
-            create_table::generate_create_table_expr(&full_table_metadata.table_info)?;
-
-        let mut errors = Vec::new();
-        for (peer, err) in failed_peers {
-            if err.status_code() != StatusCode::RegionNotFound {
-                error!(
-                    err;
-                    "Sending alter table requests to datanode: {} for table: {} failed",
-                    peer.id,
-                    full_table_name,
-                );
-                continue;
-            }
-            info!(
-                "Region not found for table: {}, datanode: {}, trying to create the logical table on that datanode",
-                full_table_name,
-                peer.id
-            );
-
-            // If the alter table request fails for any datanode, we attempt to create the table on that datanode
-            // as a fallback mechanism to ensure table consistency across the cluster.
-            if let Err(err) = self
-                .create_table_on_datanode(
-                    &create_table_expr,
-                    full_table_metadata.table_id,
-                    physical_table_id,
-                    &peer,
-                    physical_region_routes,
-                )
-                .await
-            {
-                error!(
-                    err;
-                    "Failed to create table on datanode: {} for table: {}",
-                    peer.id, full_table_name
-                );
-                errors.push(err);
-                if self.fail_fast {
-                    break;
-                }
-            } else {
-                info!(
-                    "Created table on datanode: {} for table: {}",
-                    peer.id, full_table_name
-                );
-            }
-        }
-
-        if !errors.is_empty() {
-            return UnexpectedSnafu {
-                msg: format!(
-                    "Failed to create table on datanodes for table: {}",
-                    full_table_name,
-                ),
-            }
-            .fail();
-        }
-
-        Ok(())
-    }
-}
--- a/src/cli/src/metadata/repair/alter_table.rs
+++ b/src/cli/src/metadata/repair/alter_table.rs
@@ -1,85 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use client::api::v1::alter_table_expr::Kind;
-use client::api::v1::region::{region_request, AlterRequests, RegionRequest, RegionRequestHeader};
-use client::api::v1::{AddColumn, AddColumns, AlterTableExpr};
-use common_meta::ddl::alter_logical_tables::make_alter_region_request;
-use common_meta::peer::Peer;
-use common_meta::rpc::router::{find_leader_regions, RegionRoute};
-use operator::expr_helper::column_schemas_to_defs;
-use snafu::ResultExt;
-use store_api::storage::{RegionId, TableId};
-use table::metadata::RawTableInfo;
-
-use crate::error::{CovertColumnSchemasToDefsSnafu, Result};
-
-/// Generates alter table expression for all columns.
-pub fn generate_alter_table_expr_for_all_columns(
-    table_info: &RawTableInfo,
-) -> Result<AlterTableExpr> {
-    let schema = &table_info.meta.schema;
-
-    let mut alter_table_expr = AlterTableExpr {
-        catalog_name: table_info.catalog_name.to_string(),
-        schema_name: table_info.schema_name.to_string(),
-        table_name: table_info.name.to_string(),
-        ..Default::default()
-    };
-
-    let primary_keys = table_info
-        .meta
-        .primary_key_indices
-        .iter()
-        .map(|i| schema.column_schemas[*i].name.clone())
-        .collect::<Vec<_>>();
-
-    let add_columns = column_schemas_to_defs(schema.column_schemas.clone(), &primary_keys)
-        .context(CovertColumnSchemasToDefsSnafu)?;
-
-    alter_table_expr.kind = Some(Kind::AddColumns(AddColumns {
-        add_columns: add_columns
-            .into_iter()
-            .map(|col| AddColumn {
-                column_def: Some(col),
-                location: None,
-                add_if_not_exists: true,
-            })
-            .collect(),
-    }));
-
-    Ok(alter_table_expr)
-}
-
-/// Makes an alter region request for a peer.
-pub fn make_alter_region_request_for_peer(
-    logical_table_id: TableId,
-    alter_table_expr: &AlterTableExpr,
-    schema_version: u64,
-    peer: &Peer,
-    region_routes: &[RegionRoute],
-) -> Result<RegionRequest> {
-    let regions_on_this_peer = find_leader_regions(region_routes, peer);
-    let mut requests = Vec::with_capacity(regions_on_this_peer.len());
-    for region_number in &regions_on_this_peer {
-        let region_id = RegionId::new(logical_table_id, *region_number);
-        let request = make_alter_region_request(region_id, alter_table_expr, schema_version);
-        requests.push(request);
-    }
-
-    Ok(RegionRequest {
-        header: Some(RegionRequestHeader::default()),
-        body: Some(region_request::Body::Alters(AlterRequests { requests })),
-    })
-}
--- a/src/cli/src/metadata/repair/create_table.rs
+++ b/src/cli/src/metadata/repair/create_table.rs
@@ -1,89 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::collections::HashMap;
-
-use client::api::v1::region::{region_request, CreateRequests, RegionRequest, RegionRequestHeader};
-use client::api::v1::CreateTableExpr;
-use common_meta::ddl::create_logical_tables::create_region_request_builder;
-use common_meta::ddl::utils::region_storage_path;
-use common_meta::peer::Peer;
-use common_meta::rpc::router::{find_leader_regions, RegionRoute};
-use operator::expr_helper::column_schemas_to_defs;
-use snafu::ResultExt;
-use store_api::storage::{RegionId, TableId};
-use table::metadata::RawTableInfo;
-
-use crate::error::{CovertColumnSchemasToDefsSnafu, Result};
-
-/// Generates a `CreateTableExpr` from a `RawTableInfo`.
-pub fn generate_create_table_expr(table_info: &RawTableInfo) -> Result<CreateTableExpr> {
-    let schema = &table_info.meta.schema;
-    let primary_keys = table_info
-        .meta
-        .primary_key_indices
-        .iter()
-        .map(|i| schema.column_schemas[*i].name.clone())
-        .collect::<Vec<_>>();
-
-    let timestamp_index = schema.timestamp_index.as_ref().unwrap();
-    let time_index = schema.column_schemas[*timestamp_index].name.clone();
-    let column_defs = column_schemas_to_defs(schema.column_schemas.clone(), &primary_keys)
-        .context(CovertColumnSchemasToDefsSnafu)?;
-    let table_options = HashMap::from(&table_info.meta.options);
-
-    Ok(CreateTableExpr {
-        catalog_name: table_info.catalog_name.to_string(),
-        schema_name: table_info.schema_name.to_string(),
-        table_name: table_info.name.to_string(),
-        desc: String::default(),
-        column_defs,
-        time_index,
-        primary_keys,
-        create_if_not_exists: true,
-        table_options,
-        table_id: None,
-        engine: table_info.meta.engine.to_string(),
-    })
-}
-
-/// Makes a create region request for a peer.
-pub fn make_create_region_request_for_peer(
-    logical_table_id: TableId,
-    physical_table_id: TableId,
-    create_table_expr: &CreateTableExpr,
-    peer: &Peer,
-    region_routes: &[RegionRoute],
-) -> Result<RegionRequest> {
-    let regions_on_this_peer = find_leader_regions(region_routes, peer);
-    let mut requests = Vec::with_capacity(regions_on_this_peer.len());
-    let request_builder =
-        create_region_request_builder(create_table_expr, physical_table_id).unwrap();
-
-    let catalog = &create_table_expr.catalog_name;
-    let schema = &create_table_expr.schema_name;
-    let storage_path = region_storage_path(catalog, schema);
-
-    for region_number in &regions_on_this_peer {
-        let region_id = RegionId::new(logical_table_id, *region_number);
-        let region_request =
-            request_builder.build_one(region_id, storage_path.clone(), &HashMap::new());
-        requests.push(region_request);
-    }
-
-    Ok(RegionRequest {
-        header: Some(RegionRequestHeader::default()),
-        body: Some(region_request::Body::Creates(CreateRequests { requests })),
-    })
-}
--- a/src/cli/src/metadata/utils.rs
+++ b/src/cli/src/metadata/utils.rs
@@ -1,178 +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::VecDeque;
-
-use async_stream::try_stream;
-use common_catalog::consts::METRIC_ENGINE;
-use common_catalog::format_full_table_name;
-use common_meta::key::table_name::TableNameKey;
-use common_meta::key::table_route::TableRouteValue;
-use common_meta::key::TableMetadataManager;
-use common_meta::kv_backend::KvBackendRef;
-use futures::Stream;
-use snafu::{OptionExt, ResultExt};
-use store_api::storage::TableId;
-use table::metadata::RawTableInfo;
-
-use crate::error::{Result, TableMetadataSnafu, UnexpectedSnafu};
-
-/// The input for the iterator.
-pub enum IteratorInput {
-    TableIds(VecDeque<TableId>),
-    TableNames(VecDeque<(String, String, String)>),
-}
-
-impl IteratorInput {
-    /// Creates a new iterator input from a list of table ids.
-    pub fn new_table_ids(table_ids: Vec<TableId>) -> Self {
-        Self::TableIds(table_ids.into())
-    }
-
-    /// Creates a new iterator input from a list of table names.
-    pub fn new_table_names(table_names: Vec<(String, String, String)>) -> Self {
-        Self::TableNames(table_names.into())
-    }
-}
-
-/// An iterator for retrieving table metadata from the metadata store.
-///
-/// This struct provides functionality to iterate over table metadata based on
-/// either [`TableId`] and their associated regions or fully qualified table names.
-pub struct TableMetadataIterator {
-    input: IteratorInput,
-    table_metadata_manager: TableMetadataManager,
-}
-
-/// The full table metadata.
-pub struct FullTableMetadata {
-    pub table_id: TableId,
-    pub table_info: RawTableInfo,
-    pub table_route: TableRouteValue,
-}
-
-impl FullTableMetadata {
-    /// Returns true if it's [TableRouteValue::Physical].
-    pub fn is_physical_table(&self) -> bool {
-        self.table_route.is_physical()
-    }
-
-    /// Returns true if it's a metric engine table.
-    pub fn is_metric_engine(&self) -> bool {
-        self.table_info.meta.engine == METRIC_ENGINE
-    }
-
-    /// Returns the full table name.
-    pub fn full_table_name(&self) -> String {
-        format_full_table_name(
-            &self.table_info.catalog_name,
-            &self.table_info.schema_name,
-            &self.table_info.name,
-        )
-    }
-}
-
-impl TableMetadataIterator {
-    pub fn new(kvbackend: KvBackendRef, input: IteratorInput) -> Self {
-        let table_metadata_manager = TableMetadataManager::new(kvbackend);
-        Self {
-            input,
-            table_metadata_manager,
-        }
-    }
-
-    /// Returns the next table metadata.
-    ///
-    /// This method handles two types of inputs:
-    /// - TableIds: Returns metadata for a specific [`TableId`].
-    /// - TableNames: Returns metadata for a table identified by its full name (catalog.schema.table).
-    ///
-    /// Returns `None` when there are no more tables to process.
-    pub async fn next(&mut self) -> Result<Option<FullTableMetadata>> {
-        match &mut self.input {
-            IteratorInput::TableIds(table_ids) => {
-                if let Some(table_id) = table_ids.pop_front() {
-                    let full_table_metadata = self.get_table_metadata(table_id).await?;
-                    return Ok(Some(full_table_metadata));
-                }
-            }
-
-            IteratorInput::TableNames(table_names) => {
-                if let Some(full_table_name) = table_names.pop_front() {
-                    let table_id = self.get_table_id_by_name(full_table_name).await?;
-                    let full_table_metadata = self.get_table_metadata(table_id).await?;
-                    return Ok(Some(full_table_metadata));
-                }
-            }
-        }
-
-        Ok(None)
-    }
-
-    /// Converts the iterator into a stream of table metadata.
-    pub fn into_stream(mut self) -> impl Stream<Item = Result<FullTableMetadata>> {
-        try_stream!({
-            while let Some(full_table_metadata) = self.next().await? {
-                yield full_table_metadata;
-            }
-        })
-    }
-
-    async fn get_table_id_by_name(
-        &mut self,
-        (catalog_name, schema_name, table_name): (String, String, String),
-    ) -> Result<TableId> {
-        let key = TableNameKey::new(&catalog_name, &schema_name, &table_name);
-        let table_id = self
-            .table_metadata_manager
-            .table_name_manager()
-            .get(key)
-            .await
-            .context(TableMetadataSnafu)?
-            .with_context(|| UnexpectedSnafu {
-                msg: format!(
-                    "Table not found: {}",
-                    format_full_table_name(&catalog_name, &schema_name, &table_name)
-                ),
-            })?
-            .table_id();
-        Ok(table_id)
-    }
-
-    async fn get_table_metadata(&mut self, table_id: TableId) -> Result<FullTableMetadata> {
-        let (table_info, table_route) = self
-            .table_metadata_manager
-            .get_full_table_info(table_id)
-            .await
-            .context(TableMetadataSnafu)?;
-
-        let table_info = table_info
-            .with_context(|| UnexpectedSnafu {
-                msg: format!("Table info not found for table id: {table_id}"),
-            })?
-            .into_inner()
-            .table_info;
-        let table_route = table_route
-            .with_context(|| UnexpectedSnafu {
-                msg: format!("Table route not found for table id: {table_id}"),
-            })?
-            .into_inner();
-
-        Ok(FullTableMetadata {
-            table_id,
-            table_info,
-            table_route,
-        })
-    }
-}
--- 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/cmd/Cargo.toml
+++ b/src/cmd/Cargo.toml
@@ -67,7 +67,6 @@ metric-engine.workspace = true
 mito2.workspace = true
 moka.workspace = true
 nu-ansi-term = "0.46"
-object-store.workspace = true
 plugins.workspace = true
 prometheus.workspace = true
 prost.workspace = true
--- a/src/cmd/src/datanode.rs
+++ b/src/cmd/src/datanode.rs
@@ -280,7 +280,7 @@ mod tests {

    use common_config::ENV_VAR_SEP;
    use common_test_util::temp_dir::create_named_temp_file;
-    use object_store::config::{FileConfig, GcsConfig, ObjectStoreConfig, S3Config};
+    use datanode::config::{FileConfig, GcsConfig, ObjectStoreConfig, S3Config};
    use servers::heartbeat_options::HeartbeatOptions;

    use super::*;
--- a/src/cmd/src/datanode/builder.rs
+++ b/src/cmd/src/datanode/builder.rs
@@ -93,7 +93,6 @@ impl InstanceBuilder {
            MetaClientType::Datanode { member_id },
            meta_client_options,
            Some(&plugins),
-            None,
        )
        .await
        .context(MetaClientInitSnafu)?;
--- a/src/cmd/src/flownode.rs
+++ b/src/cmd/src/flownode.rs
@@ -55,32 +55,14 @@ type FlownodeOptions = GreptimeOptions<flow::FlownodeOptions>;
 pub struct Instance {
    flownode: FlownodeInstance,

-    // The components of flownode, which make it easier to expand based
-    // on the components.
-    #[cfg(feature = "enterprise")]
-    components: Components,
-
    // Keep the logging guard to prevent the worker from being dropped.
    _guard: Vec<WorkerGuard>,
 }

-#[cfg(feature = "enterprise")]
-pub struct Components {
-    pub catalog_manager: catalog::CatalogManagerRef,
-    pub fe_client: Arc<FrontendClient>,
-    pub kv_backend: common_meta::kv_backend::KvBackendRef,
-}
-
 impl Instance {
-    pub fn new(
-        flownode: FlownodeInstance,
-        #[cfg(feature = "enterprise")] components: Components,
-        guard: Vec<WorkerGuard>,
-    ) -> Self {
+    pub fn new(flownode: FlownodeInstance, guard: Vec<WorkerGuard>) -> Self {
        Self {
            flownode,
-            #[cfg(feature = "enterprise")]
-            components,
            _guard: guard,
        }
    }
@@ -93,11 +75,6 @@ impl Instance {
    pub fn flownode_mut(&mut self) -> &mut FlownodeInstance {
        &mut self.flownode
    }
-
-    #[cfg(feature = "enterprise")]
-    pub fn components(&self) -> &Components {
-        &self.components
-    }
 }

 #[async_trait::async_trait]
@@ -306,7 +283,6 @@ impl StartCommand {
            MetaClientType::Flownode { member_id },
            meta_config,
            None,
-            None,
        )
        .await
        .context(MetaClientInitSnafu)?;
@@ -373,20 +349,19 @@ impl StartCommand {
        let flow_auth_header = get_flow_auth_options(&opts).context(StartFlownodeSnafu)?;
        let frontend_client =
            FrontendClient::from_meta_client(meta_client.clone(), flow_auth_header);
-        let frontend_client = Arc::new(frontend_client);
        let flownode_builder = FlownodeBuilder::new(
            opts.clone(),
            plugins,
            table_metadata_manager,
            catalog_manager.clone(),
            flow_metadata_manager,
-            frontend_client.clone(),
+            Arc::new(frontend_client),
        )
        .with_heartbeat_task(heartbeat_task);

        let mut flownode = flownode_builder.build().await.context(StartFlownodeSnafu)?;
        let services = FlownodeServiceBuilder::new(&opts)
-            .with_default_grpc_server(flownode.flownode_server())
+            .with_grpc_server(flownode.flownode_server().clone())
            .enable_http_service()
            .build()
            .context(StartFlownodeSnafu)?;
@@ -418,16 +393,6 @@ impl StartCommand {
            .set_frontend_invoker(invoker)
            .await;

-        #[cfg(feature = "enterprise")]
-        let components = Components {
-            catalog_manager: catalog_manager.clone(),
-            fe_client: frontend_client,
-            kv_backend: cached_meta_backend,
-        };
-
-        #[cfg(not(feature = "enterprise"))]
-        return Ok(Instance::new(flownode, guard));
-        #[cfg(feature = "enterprise")]
-        Ok(Instance::new(flownode, components, guard))
+        Ok(Instance::new(flownode, guard))
    }
 }
--- a/src/cmd/src/frontend.rs
+++ b/src/cmd/src/frontend.rs
@@ -313,7 +313,6 @@ impl StartCommand {
            MetaClientType::Frontend,
            meta_client_options,
            Some(&plugins),
-            None,
        )
        .await
        .context(error::MetaClientInitSnafu)?;
--- a/src/cmd/src/standalone.rs
+++ b/src/cmd/src/standalone.rs
@@ -30,16 +30,20 @@ use common_catalog::consts::{MIN_USER_FLOW_ID, MIN_USER_TABLE_ID};
 use common_config::{metadata_store_dir, Configurable, KvBackendConfig};
 use common_error::ext::BoxedError;
 use common_meta::cache::LayeredCacheRegistryBuilder;
+use common_meta::cache_invalidator::CacheInvalidatorRef;
 use common_meta::cluster::{NodeInfo, NodeStatus};
 use common_meta::datanode::RegionStat;
-use common_meta::ddl::flow_meta::FlowMetadataAllocator;
-use common_meta::ddl::table_meta::TableMetadataAllocator;
+use common_meta::ddl::flow_meta::{FlowMetadataAllocator, FlowMetadataAllocatorRef};
+use common_meta::ddl::table_meta::{TableMetadataAllocator, TableMetadataAllocatorRef};
 use common_meta::ddl::{DdlContext, NoopRegionFailureDetectorControl, ProcedureExecutorRef};
 use common_meta::ddl_manager::DdlManager;
+#[cfg(feature = "enterprise")]
+use common_meta::ddl_manager::TriggerDdlManagerRef;
 use common_meta::key::flow::flow_state::FlowStat;
-use common_meta::key::flow::FlowMetadataManager;
+use common_meta::key::flow::{FlowMetadataManager, FlowMetadataManagerRef};
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
+use common_meta::node_manager::NodeManagerRef;
 use common_meta::peer::Peer;
 use common_meta::region_keeper::MemoryRegionKeeper;
 use common_meta::region_registry::LeaderRegionRegistry;
@@ -257,34 +261,15 @@ pub struct Instance {
    flownode: FlownodeInstance,
    procedure_manager: ProcedureManagerRef,
    wal_options_allocator: WalOptionsAllocatorRef,
-
-    // The components of standalone, which make it easier to expand based
-    // on the components.
-    #[cfg(feature = "enterprise")]
-    components: Components,
-
    // Keep the logging guard to prevent the worker from being dropped.
    _guard: Vec<WorkerGuard>,
 }

-#[cfg(feature = "enterprise")]
-pub struct Components {
-    pub plugins: Plugins,
-    pub kv_backend: KvBackendRef,
-    pub frontend_client: Arc<FrontendClient>,
-    pub catalog_manager: catalog::CatalogManagerRef,
-}
-
 impl Instance {
    /// Find the socket addr of a server by its `name`.
    pub fn server_addr(&self, name: &str) -> Option<SocketAddr> {
        self.frontend.server_handlers().addr(name)
    }
-
-    #[cfg(feature = "enterprise")]
-    pub fn components(&self) -> &Components {
-        &self.components
-    }
 }

 #[async_trait]
@@ -565,14 +550,13 @@ impl StartCommand {
        // actually make a connection
        let (frontend_client, frontend_instance_handler) =
            FrontendClient::from_empty_grpc_handler();
-        let frontend_client = Arc::new(frontend_client);
        let flow_builder = FlownodeBuilder::new(
            flownode_options,
            plugins.clone(),
            table_metadata_manager.clone(),
            catalog_manager.clone(),
            flow_metadata_manager.clone(),
-            frontend_client.clone(),
+            Arc::new(frontend_client.clone()),
        );
        let flownode = flow_builder
            .build()
@@ -610,36 +594,28 @@ impl StartCommand {
            .await
            .context(error::BuildWalOptionsAllocatorSnafu)?;
        let wal_options_allocator = Arc::new(wal_options_allocator);
-        let table_metadata_allocator = Arc::new(TableMetadataAllocator::new(
+        let table_meta_allocator = Arc::new(TableMetadataAllocator::new(
            table_id_sequence,
            wal_options_allocator.clone(),
        ));
-        let flow_metadata_allocator = Arc::new(FlowMetadataAllocator::with_noop_peer_allocator(
+        let flow_meta_allocator = Arc::new(FlowMetadataAllocator::with_noop_peer_allocator(
            flow_id_sequence,
        ));

-        let ddl_context = DdlContext {
-            node_manager: node_manager.clone(),
-            cache_invalidator: layered_cache_registry.clone(),
-            memory_region_keeper: Arc::new(MemoryRegionKeeper::default()),
-            leader_region_registry: Arc::new(LeaderRegionRegistry::default()),
-            table_metadata_manager: table_metadata_manager.clone(),
-            table_metadata_allocator: table_metadata_allocator.clone(),
-            flow_metadata_manager: flow_metadata_manager.clone(),
-            flow_metadata_allocator: flow_metadata_allocator.clone(),
-            region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl),
-        };
-        let procedure_manager_c = procedure_manager.clone();
-
-        let ddl_manager = DdlManager::try_new(ddl_context, procedure_manager_c, true)
-            .context(error::InitDdlManagerSnafu)?;
        #[cfg(feature = "enterprise")]
-        let ddl_manager = {
-            let trigger_ddl_manager: Option<common_meta::ddl_manager::TriggerDdlManagerRef> =
-                plugins.get();
-            ddl_manager.with_trigger_ddl_manager(trigger_ddl_manager)
-        };
-        let ddl_task_executor: ProcedureExecutorRef = Arc::new(ddl_manager);
+        let trigger_ddl_manager: Option<TriggerDdlManagerRef> = plugins.get();
+        let ddl_task_executor = Self::create_ddl_task_executor(
+            procedure_manager.clone(),
+            node_manager.clone(),
+            layered_cache_registry.clone(),
+            table_metadata_manager,
+            table_meta_allocator,
+            flow_metadata_manager,
+            flow_meta_allocator,
+            #[cfg(feature = "enterprise")]
+            trigger_ddl_manager,
+        )
+        .await?;

        let fe_instance = FrontendBuilder::new(
            fe_opts.clone(),
@@ -682,7 +658,7 @@ impl StartCommand {
        let export_metrics_task = ExportMetricsTask::try_new(&opts.export_metrics, Some(&plugins))
            .context(error::ServersSnafu)?;

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

@@ -693,26 +669,51 @@ impl StartCommand {
            export_metrics_task,
        };

-        #[cfg(feature = "enterprise")]
-        let components = Components {
-            plugins,
-            kv_backend,
-            frontend_client,
-            catalog_manager,
-        };
-
        Ok(Instance {
            datanode,
            frontend,
            flownode,
            procedure_manager,
            wal_options_allocator,
-            #[cfg(feature = "enterprise")]
-            components,
            _guard: guard,
        })
    }

+    #[allow(clippy::too_many_arguments)]
+    pub async fn create_ddl_task_executor(
+        procedure_manager: ProcedureManagerRef,
+        node_manager: NodeManagerRef,
+        cache_invalidator: CacheInvalidatorRef,
+        table_metadata_manager: TableMetadataManagerRef,
+        table_metadata_allocator: TableMetadataAllocatorRef,
+        flow_metadata_manager: FlowMetadataManagerRef,
+        flow_metadata_allocator: FlowMetadataAllocatorRef,
+        #[cfg(feature = "enterprise")] trigger_ddl_manager: Option<TriggerDdlManagerRef>,
+    ) -> Result<ProcedureExecutorRef> {
+        let procedure_executor: ProcedureExecutorRef = Arc::new(
+            DdlManager::try_new(
+                DdlContext {
+                    node_manager,
+                    cache_invalidator,
+                    memory_region_keeper: Arc::new(MemoryRegionKeeper::default()),
+                    leader_region_registry: Arc::new(LeaderRegionRegistry::default()),
+                    table_metadata_manager,
+                    table_metadata_allocator,
+                    flow_metadata_manager,
+                    flow_metadata_allocator,
+                    region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl),
+                },
+                procedure_manager,
+                true,
+                #[cfg(feature = "enterprise")]
+                trigger_ddl_manager,
+            )
+            .context(error::InitDdlManagerSnafu)?,
+        );
+
+        Ok(procedure_executor)
+    }
+
    pub async fn create_table_metadata_manager(
        kv_backend: KvBackendRef,
    ) -> Result<TableMetadataManagerRef> {
@@ -848,7 +849,7 @@ mod tests {
    use common_config::ENV_VAR_SEP;
    use common_test_util::temp_dir::create_named_temp_file;
    use common_wal::config::DatanodeWalConfig;
-    use object_store::config::{FileConfig, GcsConfig};
+    use datanode::config::{FileConfig, GcsConfig};

    use super::*;
    use crate::options::GlobalOptions;
@@ -967,15 +968,15 @@ mod tests {

        assert!(matches!(
            &dn_opts.storage.store,
-            object_store::config::ObjectStoreConfig::File(FileConfig { .. })
+            datanode::config::ObjectStoreConfig::File(FileConfig { .. })
        ));
        assert_eq!(dn_opts.storage.providers.len(), 2);
        assert!(matches!(
            dn_opts.storage.providers[0],
-            object_store::config::ObjectStoreConfig::Gcs(GcsConfig { .. })
+            datanode::config::ObjectStoreConfig::Gcs(GcsConfig { .. })
        ));
        match &dn_opts.storage.providers[1] {
-            object_store::config::ObjectStoreConfig::S3(s3_config) => {
+            datanode::config::ObjectStoreConfig::S3(s3_config) => {
                assert_eq!(
                    "SecretBox<alloc::string::String>([REDACTED])".to_string(),
                    format!("{:?}", s3_config.access_key_id)
--- a/src/cmd/tests/load_config_test.rs
+++ b/src/cmd/tests/load_config_test.rs
@@ -12,13 +12,14 @@
 // 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;
 use cmd::standalone::StandaloneOptions;
 use common_config::{Configurable, DEFAULT_DATA_HOME};
 use common_options::datanode::{ClientOptions, DatanodeClientOptions};
-use common_telemetry::logging::{LoggingOptions, DEFAULT_LOGGING_DIR, DEFAULT_OTLP_HTTP_ENDPOINT};
+use common_telemetry::logging::{LoggingOptions, DEFAULT_LOGGING_DIR, DEFAULT_OTLP_ENDPOINT};
 use common_wal::config::raft_engine::RaftEngineConfig;
 use common_wal::config::DatanodeWalConfig;
 use datanode::config::{DatanodeOptions, RegionEngineConfig, StorageConfig};
@@ -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,8 +86,11 @@ fn test_load_datanode_example_config() {
            ],
            logging: LoggingOptions {
                level: Some("info".to_string()),
-                dir: format!("{}/{}", DEFAULT_DATA_HOME, DEFAULT_LOGGING_DIR),
-                otlp_endpoint: Some(DEFAULT_OTLP_HTTP_ENDPOINT.to_string()),
+                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,8 +132,11 @@ fn test_load_frontend_example_config() {
            }),
            logging: LoggingOptions {
                level: Some("info".to_string()),
-                dir: format!("{}/{}", DEFAULT_DATA_HOME, DEFAULT_LOGGING_DIR),
-                otlp_endpoint: Some(DEFAULT_OTLP_HTTP_ENDPOINT.to_string()),
+                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,9 +182,12 @@ 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_HTTP_ENDPOINT.to_string()),
+                otlp_endpoint: Some(DEFAULT_OTLP_ENDPOINT.to_string()),
                tracing_sample_ratio: Some(Default::default()),
                ..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,8 +248,11 @@ fn test_load_standalone_example_config() {
            },
            logging: LoggingOptions {
                level: Some("info".to_string()),
-                dir: format!("{}/{}", DEFAULT_DATA_HOME, DEFAULT_LOGGING_DIR),
-                otlp_endpoint: Some(DEFAULT_OTLP_HTTP_ENDPOINT.to_string()),
+                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/config/Cargo.toml
+++ b/src/common/config/Cargo.toml
@@ -14,7 +14,6 @@ common-macro.workspace = true
 config.workspace = true
 humantime-serde.workspace = true
 num_cpus.workspace = true
-object-store.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 serde_with.workspace = true
--- a/src/common/config/src/config.rs
+++ b/src/common/config/src/config.rs
@@ -106,7 +106,7 @@ mod tests {
    use common_telemetry::logging::LoggingOptions;
    use common_test_util::temp_dir::create_named_temp_file;
    use common_wal::config::DatanodeWalConfig;
-    use datanode::config::StorageConfig;
+    use datanode::config::{ObjectStoreConfig, StorageConfig};
    use meta_client::MetaClientOptions;
    use serde::{Deserialize, Serialize};

@@ -212,7 +212,7 @@ mod tests {

                // Check the configs from environment variables.
                match &opts.storage.store {
-                    object_store::config::ObjectStoreConfig::S3(s3_config) => {
+                    ObjectStoreConfig::S3(s3_config) => {
                        assert_eq!(s3_config.bucket, "mybucket".to_string());
                    }
                    _ => panic!("unexpected store type"),
--- a/src/common/datasource/src/lib.rs
+++ b/src/common/datasource/src/lib.rs
@@ -21,7 +21,6 @@ pub mod error;
 pub mod file_format;
 pub mod lister;
 pub mod object_store;
-pub mod parquet_writer;
 pub mod share_buffer;
 #[cfg(test)]
 pub mod test_util;
--- a/src/common/datasource/src/parquet_writer.rs
+++ b/src/common/datasource/src/parquet_writer.rs
@@ -1,52 +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 bytes::Bytes;
-use futures::future::BoxFuture;
-use object_store::Writer;
-use parquet::arrow::async_writer::AsyncFileWriter;
-use parquet::errors::ParquetError;
-
-/// Bridges opendal [Writer] with parquet [AsyncFileWriter].
-pub struct AsyncWriter {
-    inner: Writer,
-}
-
-impl AsyncWriter {
-    /// Create a [`AsyncWriter`] by given [`Writer`].
-    pub fn new(writer: Writer) -> Self {
-        Self { inner: writer }
-    }
-}
-
-impl AsyncFileWriter for AsyncWriter {
-    fn write(&mut self, bs: Bytes) -> BoxFuture<'_, parquet::errors::Result<()>> {
-        Box::pin(async move {
-            self.inner
-                .write(bs)
-                .await
-                .map_err(|err| ParquetError::External(Box::new(err)))
-        })
-    }
-
-    fn complete(&mut self) -> BoxFuture<'_, parquet::errors::Result<()>> {
-        Box::pin(async move {
-            self.inner
-                .close()
-                .await
-                .map(|_| ())
-                .map_err(|err| ParquetError::External(Box::new(err)))
-        })
-    }
-}
--- a/src/common/frontend/src/error.rs
+++ b/src/common/frontend/src/error.rs
@@ -42,8 +42,8 @@ pub enum Error {
        location: Location,
    },

-    #[snafu(display("Failed to invoke frontend service"))]
-    InvokeFrontend {
+    #[snafu(display("Failed to invoke list process service"))]
+    ListProcess {
        #[snafu(source)]
        error: tonic::Status,
        #[snafu(implicit)]
@@ -67,7 +67,7 @@ impl ErrorExt for Error {
            External { source, .. } => source.status_code(),
            Meta { source, .. } => source.status_code(),
            ParseProcessId { .. } => StatusCode::InvalidArguments,
-            InvokeFrontend { .. } => StatusCode::Unexpected,
+            ListProcess { .. } => StatusCode::External,
            CreateChannel { source, .. } => source.status_code(),
        }
    }
--- a/src/common/frontend/src/lib.rs
+++ b/src/common/frontend/src/lib.rs
@@ -23,7 +23,7 @@ pub mod selector;
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub struct DisplayProcessId {
    pub server_addr: String,
-    pub id: u32,
+    pub id: u64,
 }

 impl Display for DisplayProcessId {
@@ -44,7 +44,7 @@ impl TryFrom<&str> for DisplayProcessId {
        let id = split
            .next()
            .context(error::ParseProcessIdSnafu { s: value })?;
-        let id = u32::from_str(id)
+        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
@@ -16,13 +16,9 @@ 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 greptime_proto::v1::frontend::{frontend_client, ListProcessRequest, ListProcessResponse};
 use meta_client::MetaClientRef;
 use snafu::ResultExt;
-use tonic::Response;

 use crate::error;
 use crate::error::{MetaSnafu, Result};
@@ -32,28 +28,18 @@ 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,
-        )
+        let response: 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)
+        .context(error::ListProcessSnafu)?
+        .into_inner();
+        Ok(response)
    }
 }

--- a/src/common/function/Cargo.toml
+++ b/src/common/function/Cargo.toml
@@ -33,7 +33,6 @@ common-version.workspace = true
 datafusion.workspace = true
 datafusion-common.workspace = true
 datafusion-expr.workspace = true
-datafusion-functions-aggregate-common.workspace = true
 datatypes.workspace = true
 derive_more = { version = "1", default-features = false, features = ["display"] }
 geo = { version = "0.29", optional = true }
--- a/src/common/function/src/aggrs.rs
+++ b/src/common/function/src/aggrs.rs
@@ -13,7 +13,6 @@
 // limitations under the License.

 pub mod approximate;
-pub mod count_hash;
 #[cfg(feature = "geo")]
 pub mod geo;
 pub mod vector;
--- a/src/common/function/src/aggrs/approximate.rs
+++ b/src/common/function/src/aggrs/approximate.rs
@@ -14,8 +14,8 @@

 use crate::function_registry::FunctionRegistry;

-pub mod hll;
-pub mod uddsketch;
+pub(crate) mod hll;
+mod uddsketch;

 pub(crate) struct ApproximateFunction;

--- a/src/common/function/src/aggrs/count_hash.rs
+++ b/src/common/function/src/aggrs/count_hash.rs
@@ -1,647 +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.
-
-//! `CountHash` / `count_hash` is a hash-based approximate distinct count function.
-//!
-//! It is a variant of `CountDistinct` that uses a hash function to approximate the
-//! distinct count.
-//! It is designed to be more efficient than `CountDistinct` for large datasets,
-//! but it is not as accurate, as the hash value may be collision.
-
-use std::collections::HashSet;
-use std::fmt::Debug;
-use std::sync::Arc;
-
-use ahash::RandomState;
-use datafusion_common::cast::as_list_array;
-use datafusion_common::error::Result;
-use datafusion_common::hash_utils::create_hashes;
-use datafusion_common::utils::SingleRowListArrayBuilder;
-use datafusion_common::{internal_err, not_impl_err, ScalarValue};
-use datafusion_expr::function::{AccumulatorArgs, StateFieldsArgs};
-use datafusion_expr::utils::{format_state_name, AggregateOrderSensitivity};
-use datafusion_expr::{
-    Accumulator, AggregateUDF, AggregateUDFImpl, EmitTo, GroupsAccumulator, ReversedUDAF,
-    SetMonotonicity, Signature, TypeSignature, Volatility,
-};
-use datafusion_functions_aggregate_common::aggregate::groups_accumulator::nulls::filtered_null_mask;
-use datatypes::arrow;
-use datatypes::arrow::array::{
-    Array, ArrayRef, AsArray, BooleanArray, Int64Array, ListArray, UInt64Array,
-};
-use datatypes::arrow::buffer::{OffsetBuffer, ScalarBuffer};
-use datatypes::arrow::datatypes::{DataType, Field};
-
-use crate::function_registry::FunctionRegistry;
-
-type HashValueType = u64;
-
-// read from /dev/urandom 4047821dc6144e4b2abddf23ad4171126a52eeecd26eff2191cf673b965a7875
-const RANDOM_SEED_0: u64 = 0x4047821dc6144e4b;
-const RANDOM_SEED_1: u64 = 0x2abddf23ad417112;
-const RANDOM_SEED_2: u64 = 0x6a52eeecd26eff21;
-const RANDOM_SEED_3: u64 = 0x91cf673b965a7875;
-
-impl CountHash {
-    pub fn register(registry: &FunctionRegistry) {
-        registry.register_aggr(CountHash::udf_impl());
-    }
-
-    pub fn udf_impl() -> AggregateUDF {
-        AggregateUDF::new_from_impl(CountHash {
-            signature: Signature::one_of(
-                vec![TypeSignature::VariadicAny, TypeSignature::Nullary],
-                Volatility::Immutable,
-            ),
-        })
-    }
-}
-
-#[derive(Debug, Clone)]
-pub struct CountHash {
-    signature: Signature,
-}
-
-impl AggregateUDFImpl for CountHash {
-    fn as_any(&self) -> &dyn std::any::Any {
-        self
-    }
-
-    fn name(&self) -> &str {
-        "count_hash"
-    }
-
-    fn signature(&self) -> &Signature {
-        &self.signature
-    }
-
-    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
-        Ok(DataType::Int64)
-    }
-
-    fn is_nullable(&self) -> bool {
-        false
-    }
-
-    fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<Field>> {
-        Ok(vec![Field::new_list(
-            format_state_name(args.name, "count_hash"),
-            Field::new_list_field(DataType::UInt64, true),
-            // For count_hash accumulator, null list item stands for an
-            // empty value set (i.e., all NULL value so far for that group).
-            true,
-        )])
-    }
-
-    fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
-        if acc_args.exprs.len() > 1 {
-            return not_impl_err!("count_hash with multiple arguments");
-        }
-
-        Ok(Box::new(CountHashAccumulator {
-            values: HashSet::default(),
-            random_state: RandomState::with_seeds(
-                RANDOM_SEED_0,
-                RANDOM_SEED_1,
-                RANDOM_SEED_2,
-                RANDOM_SEED_3,
-            ),
-            batch_hashes: vec![],
-        }))
-    }
-
-    fn aliases(&self) -> &[String] {
-        &[]
-    }
-
-    fn groups_accumulator_supported(&self, _args: AccumulatorArgs) -> bool {
-        true
-    }
-
-    fn create_groups_accumulator(
-        &self,
-        args: AccumulatorArgs,
-    ) -> Result<Box<dyn GroupsAccumulator>> {
-        if args.exprs.len() > 1 {
-            return not_impl_err!("count_hash with multiple arguments");
-        }
-
-        Ok(Box::new(CountHashGroupAccumulator::new()))
-    }
-
-    fn reverse_expr(&self) -> ReversedUDAF {
-        ReversedUDAF::Identical
-    }
-
-    fn order_sensitivity(&self) -> AggregateOrderSensitivity {
-        AggregateOrderSensitivity::Insensitive
-    }
-
-    fn default_value(&self, _data_type: &DataType) -> Result<ScalarValue> {
-        Ok(ScalarValue::Int64(Some(0)))
-    }
-
-    fn set_monotonicity(&self, _data_type: &DataType) -> SetMonotonicity {
-        SetMonotonicity::Increasing
-    }
-}
-
-/// GroupsAccumulator for `count_hash` aggregate function
-#[derive(Debug)]
-pub struct CountHashGroupAccumulator {
-    /// One HashSet per group to track distinct values
-    distinct_sets: Vec<HashSet<HashValueType, RandomState>>,
-    random_state: RandomState,
-    batch_hashes: Vec<HashValueType>,
-}
-
-impl Default for CountHashGroupAccumulator {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl CountHashGroupAccumulator {
-    pub fn new() -> Self {
-        Self {
-            distinct_sets: vec![],
-            random_state: RandomState::with_seeds(
-                RANDOM_SEED_0,
-                RANDOM_SEED_1,
-                RANDOM_SEED_2,
-                RANDOM_SEED_3,
-            ),
-            batch_hashes: vec![],
-        }
-    }
-
-    fn ensure_sets(&mut self, total_num_groups: usize) {
-        if self.distinct_sets.len() < total_num_groups {
-            self.distinct_sets
-                .resize_with(total_num_groups, HashSet::default);
-        }
-    }
-}
-
-impl GroupsAccumulator for CountHashGroupAccumulator {
-    fn update_batch(
-        &mut self,
-        values: &[ArrayRef],
-        group_indices: &[usize],
-        opt_filter: Option<&BooleanArray>,
-        total_num_groups: usize,
-    ) -> Result<()> {
-        assert_eq!(values.len(), 1, "count_hash expects a single argument");
-        self.ensure_sets(total_num_groups);
-
-        let array = &values[0];
-        self.batch_hashes.clear();
-        self.batch_hashes.resize(array.len(), 0);
-        let hashes = create_hashes(
-            &[ArrayRef::clone(array)],
-            &self.random_state,
-            &mut self.batch_hashes,
-        )?;
-
-        // Use a pattern similar to accumulate_indices to process rows
-        // that are not null and pass the filter
-        let nulls = array.logical_nulls();
-
-        match (nulls.as_ref(), opt_filter) {
-            (None, None) => {
-                // No nulls, no filter - process all rows
-                for (row_idx, &group_idx) in group_indices.iter().enumerate() {
-                    self.distinct_sets[group_idx].insert(hashes[row_idx]);
-                }
-            }
-            (Some(nulls), None) => {
-                // Has nulls, no filter
-                for (row_idx, (&group_idx, is_valid)) in
-                    group_indices.iter().zip(nulls.iter()).enumerate()
-                {
-                    if is_valid {
-                        self.distinct_sets[group_idx].insert(hashes[row_idx]);
-                    }
-                }
-            }
-            (None, Some(filter)) => {
-                // No nulls, has filter
-                for (row_idx, (&group_idx, filter_value)) in
-                    group_indices.iter().zip(filter.iter()).enumerate()
-                {
-                    if let Some(true) = filter_value {
-                        self.distinct_sets[group_idx].insert(hashes[row_idx]);
-                    }
-                }
-            }
-            (Some(nulls), Some(filter)) => {
-                // Has nulls and filter
-                let iter = filter
-                    .iter()
-                    .zip(group_indices.iter())
-                    .zip(nulls.iter())
-                    .enumerate();
-
-                for (row_idx, ((filter_value, &group_idx), is_valid)) in iter {
-                    if is_valid && filter_value == Some(true) {
-                        self.distinct_sets[group_idx].insert(hashes[row_idx]);
-                    }
-                }
-            }
-        }
-
-        Ok(())
-    }
-
-    fn evaluate(&mut self, emit_to: EmitTo) -> Result<ArrayRef> {
-        let distinct_sets: Vec<HashSet<u64, RandomState>> =
-            emit_to.take_needed(&mut self.distinct_sets);
-
-        let counts = distinct_sets
-            .iter()
-            .map(|set| set.len() as i64)
-            .collect::<Vec<_>>();
-        Ok(Arc::new(Int64Array::from(counts)))
-    }
-
-    fn merge_batch(
-        &mut self,
-        values: &[ArrayRef],
-        group_indices: &[usize],
-        _opt_filter: Option<&BooleanArray>,
-        total_num_groups: usize,
-    ) -> Result<()> {
-        assert_eq!(
-            values.len(),
-            1,
-            "count_hash merge expects a single state array"
-        );
-        self.ensure_sets(total_num_groups);
-
-        let list_array = as_list_array(&values[0])?;
-
-        // For each group in the incoming batch
-        for (i, &group_idx) in group_indices.iter().enumerate() {
-            if i < list_array.len() {
-                let inner_array = list_array.value(i);
-                let inner_array = inner_array.as_any().downcast_ref::<UInt64Array>().unwrap();
-                // Add each value to our set for this group
-                for j in 0..inner_array.len() {
-                    if !inner_array.is_null(j) {
-                        self.distinct_sets[group_idx].insert(inner_array.value(j));
-                    }
-                }
-            }
-        }
-
-        Ok(())
-    }
-
-    fn state(&mut self, emit_to: EmitTo) -> Result<Vec<ArrayRef>> {
-        let distinct_sets: Vec<HashSet<u64, RandomState>> =
-            emit_to.take_needed(&mut self.distinct_sets);
-
-        let mut offsets = Vec::with_capacity(distinct_sets.len() + 1);
-        offsets.push(0);
-        let mut curr_len = 0i32;
-
-        let mut value_iter = distinct_sets
-            .into_iter()
-            .flat_map(|set| {
-                // build offset
-                curr_len += set.len() as i32;
-                offsets.push(curr_len);
-                // convert into iter
-                set.into_iter()
-            })
-            .peekable();
-        let data_array: ArrayRef = if value_iter.peek().is_none() {
-            arrow::array::new_empty_array(&DataType::UInt64) as _
-        } else {
-            Arc::new(UInt64Array::from_iter_values(value_iter))
-        };
-        let offset_buffer = OffsetBuffer::new(ScalarBuffer::from(offsets));
-
-        let list_array = ListArray::new(
-            Arc::new(Field::new_list_field(DataType::UInt64, true)),
-            offset_buffer,
-            data_array,
-            None,
-        );
-
-        Ok(vec![Arc::new(list_array) as _])
-    }
-
-    fn convert_to_state(
-        &self,
-        values: &[ArrayRef],
-        opt_filter: Option<&BooleanArray>,
-    ) -> Result<Vec<ArrayRef>> {
-        // For a single hash value per row, create a list array with that value
-        assert_eq!(values.len(), 1, "count_hash expects a single argument");
-        let values = ArrayRef::clone(&values[0]);
-
-        let offsets = OffsetBuffer::new(ScalarBuffer::from_iter(0..values.len() as i32 + 1));
-        let nulls = filtered_null_mask(opt_filter, &values);
-        let list_array = ListArray::new(
-            Arc::new(Field::new_list_field(DataType::UInt64, true)),
-            offsets,
-            values,
-            nulls,
-        );
-
-        Ok(vec![Arc::new(list_array)])
-    }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
-    fn size(&self) -> usize {
-        // Base size of the struct
-        let mut size = size_of::<Self>();
-
-        // Size of the vector holding the HashSets
-        size += size_of::<Vec<HashSet<HashValueType, RandomState>>>()
-            + self.distinct_sets.capacity() * size_of::<HashSet<HashValueType, RandomState>>();
-
-        // Estimate HashSet contents size more efficiently
-        // Instead of iterating through all values which is expensive, use an approximation
-        for set in &self.distinct_sets {
-            // Base size of the HashSet
-            size += set.capacity() * size_of::<HashValueType>();
-        }
-
-        size
-    }
-}
-
-#[derive(Debug)]
-struct CountHashAccumulator {
-    values: HashSet<HashValueType, RandomState>,
-    random_state: RandomState,
-    batch_hashes: Vec<HashValueType>,
-}
-
-impl CountHashAccumulator {
-    // calculating the size for fixed length values, taking first batch size *
-    // number of batches.
-    fn fixed_size(&self) -> usize {
-        size_of_val(self) + (size_of::<HashValueType>() * self.values.capacity())
-    }
-}
-
-impl Accumulator for CountHashAccumulator {
-    /// Returns the distinct values seen so far as (one element) ListArray.
-    fn state(&mut self) -> Result<Vec<ScalarValue>> {
-        let values = self.values.iter().cloned().collect::<Vec<_>>();
-        let arr = Arc::new(UInt64Array::from(values)) as _;
-        let list_scalar = SingleRowListArrayBuilder::new(arr).build_list_scalar();
-        Ok(vec![list_scalar])
-    }
-
-    fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
-        if values.is_empty() {
-            return Ok(());
-        }
-
-        let arr = &values[0];
-        if arr.data_type() == &DataType::Null {
-            return Ok(());
-        }
-
-        self.batch_hashes.clear();
-        self.batch_hashes.resize(arr.len(), 0);
-        let hashes = create_hashes(
-            &[ArrayRef::clone(arr)],
-            &self.random_state,
-            &mut self.batch_hashes,
-        )?;
-        for hash in hashes.as_slice() {
-            self.values.insert(*hash);
-        }
-        Ok(())
-    }
-
-    /// Merges multiple sets of distinct values into the current set.
-    ///
-    /// The input to this function is a `ListArray` with **multiple** rows,
-    /// where each row contains the values from a partial aggregate's phase (e.g.
-    /// the result of calling `Self::state` on multiple accumulators).
-    fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
-        if states.is_empty() {
-            return Ok(());
-        }
-        assert_eq!(states.len(), 1, "array_agg states must be singleton!");
-        let array = &states[0];
-        let list_array = array.as_list::<i32>();
-        for inner_array in list_array.iter() {
-            let Some(inner_array) = inner_array else {
-                return internal_err!(
-                    "Intermediate results of count_hash should always be non null"
-                );
-            };
-            let hash_array = inner_array.as_any().downcast_ref::<UInt64Array>().unwrap();
-            for i in 0..hash_array.len() {
-                self.values.insert(hash_array.value(i));
-            }
-        }
-        Ok(())
-    }
-
-    fn evaluate(&mut self) -> Result<ScalarValue> {
-        Ok(ScalarValue::Int64(Some(self.values.len() as i64)))
-    }
-
-    fn size(&self) -> usize {
-        self.fixed_size()
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use datatypes::arrow::array::{Array, BooleanArray, Int32Array, Int64Array};
-
-    use super::*;
-
-    fn create_test_accumulator() -> CountHashAccumulator {
-        CountHashAccumulator {
-            values: HashSet::default(),
-            random_state: RandomState::with_seeds(
-                RANDOM_SEED_0,
-                RANDOM_SEED_1,
-                RANDOM_SEED_2,
-                RANDOM_SEED_3,
-            ),
-            batch_hashes: vec![],
-        }
-    }
-
-    #[test]
-    fn test_count_hash_accumulator() -> Result<()> {
-        let mut acc = create_test_accumulator();
-
-        // Test with some data
-        let array = Arc::new(Int32Array::from(vec![
-            Some(1),
-            Some(2),
-            Some(3),
-            Some(1),
-            Some(2),
-            None,
-        ])) as ArrayRef;
-        acc.update_batch(&[array])?;
-        let result = acc.evaluate()?;
-        assert_eq!(result, ScalarValue::Int64(Some(4)));
-
-        // Test with empty data
-        let mut acc = create_test_accumulator();
-        let array = Arc::new(Int32Array::from(vec![] as Vec<Option<i32>>)) as ArrayRef;
-        acc.update_batch(&[array])?;
-        let result = acc.evaluate()?;
-        assert_eq!(result, ScalarValue::Int64(Some(0)));
-
-        // Test with only nulls
-        let mut acc = create_test_accumulator();
-        let array = Arc::new(Int32Array::from(vec![None, None, None])) as ArrayRef;
-        acc.update_batch(&[array])?;
-        let result = acc.evaluate()?;
-        assert_eq!(result, ScalarValue::Int64(Some(1)));
-
-        Ok(())
-    }
-
-    #[test]
-    fn test_count_hash_accumulator_merge() -> Result<()> {
-        // Accumulator 1
-        let mut acc1 = create_test_accumulator();
-        let array1 = Arc::new(Int32Array::from(vec![Some(1), Some(2), Some(3)])) as ArrayRef;
-        acc1.update_batch(&[array1])?;
-        let state1 = acc1.state()?;
-
-        // Accumulator 2
-        let mut acc2 = create_test_accumulator();
-        let array2 = Arc::new(Int32Array::from(vec![Some(3), Some(4), Some(5)])) as ArrayRef;
-        acc2.update_batch(&[array2])?;
-        let state2 = acc2.state()?;
-
-        // Merge state1 and state2 into a new accumulator
-        let mut acc_merged = create_test_accumulator();
-        let state_array1 = state1[0].to_array()?;
-        let state_array2 = state2[0].to_array()?;
-
-        acc_merged.merge_batch(&[state_array1])?;
-        acc_merged.merge_batch(&[state_array2])?;
-
-        let result = acc_merged.evaluate()?;
-        // Distinct values are {1, 2, 3, 4, 5}, so count is 5
-        assert_eq!(result, ScalarValue::Int64(Some(5)));
-
-        Ok(())
-    }
-
-    fn create_test_group_accumulator() -> CountHashGroupAccumulator {
-        CountHashGroupAccumulator::new()
-    }
-
-    #[test]
-    fn test_count_hash_group_accumulator() -> Result<()> {
-        let mut acc = create_test_group_accumulator();
-        let values = Arc::new(Int32Array::from(vec![1, 2, 1, 3, 2, 4, 5])) as ArrayRef;
-        let group_indices = vec![0, 1, 0, 0, 1, 2, 0];
-        let total_num_groups = 3;
-
-        acc.update_batch(&[values], &group_indices, None, total_num_groups)?;
-
-        let result_array = acc.evaluate(EmitTo::All)?;
-        let result = result_array.as_any().downcast_ref::<Int64Array>().unwrap();
-
-        // Group 0: {1, 3, 5} -> 3
-        // Group 1: {2} -> 1
-        // Group 2: {4} -> 1
-        assert_eq!(result.value(0), 3);
-        assert_eq!(result.value(1), 1);
-        assert_eq!(result.value(2), 1);
-
-        Ok(())
-    }
-
-    #[test]
-    fn test_count_hash_group_accumulator_with_filter() -> Result<()> {
-        let mut acc = create_test_group_accumulator();
-        let values = Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5, 6])) as ArrayRef;
-        let group_indices = vec![0, 0, 1, 1, 2, 2];
-        let filter = BooleanArray::from(vec![true, false, true, true, false, true]);
-        let total_num_groups = 3;
-
-        acc.update_batch(&[values], &group_indices, Some(&filter), total_num_groups)?;
-
-        let result_array = acc.evaluate(EmitTo::All)?;
-        let result = result_array.as_any().downcast_ref::<Int64Array>().unwrap();
-
-        // Group 0: {1} (2 is filtered out) -> 1
-        // Group 1: {3, 4} -> 2
-        // Group 2: {6} (5 is filtered out) -> 1
-        assert_eq!(result.value(0), 1);
-        assert_eq!(result.value(1), 2);
-        assert_eq!(result.value(2), 1);
-
-        Ok(())
-    }
-
-    #[test]
-    fn test_count_hash_group_accumulator_merge() -> Result<()> {
-        // Accumulator 1
-        let mut acc1 = create_test_group_accumulator();
-        let values1 = Arc::new(Int32Array::from(vec![1, 2, 3, 4])) as ArrayRef;
-        let group_indices1 = vec![0, 0, 1, 1];
-        acc1.update_batch(&[values1], &group_indices1, None, 2)?;
-        // acc1 state: group 0 -> {1, 2}, group 1 -> {3, 4}
-        let state1 = acc1.state(EmitTo::All)?;
-
-        // Accumulator 2
-        let mut acc2 = create_test_group_accumulator();
-        let values2 = Arc::new(Int32Array::from(vec![5, 6, 1, 3])) as ArrayRef;
-        // Merge into different group indices
-        let group_indices2 = vec![2, 2, 0, 1];
-        acc2.update_batch(&[values2], &group_indices2, None, 3)?;
-        // acc2 state: group 0 -> {1}, group 1 -> {3}, group 2 -> {5, 6}
-
-        // Merge state from acc1 into acc2
-        // We will merge acc1's group 0 into acc2's group 0
-        // and acc1's group 1 into acc2's group 2
-        let merge_group_indices = vec![0, 2];
-        acc2.merge_batch(&state1, &merge_group_indices, None, 3)?;
-
-        let result_array = acc2.evaluate(EmitTo::All)?;
-        let result = result_array.as_any().downcast_ref::<Int64Array>().unwrap();
-
-        // Final state of acc2:
-        // Group 0: {1} U {1, 2} -> {1, 2}, count = 2
-        // Group 1: {3}, count = 1
-        // Group 2: {5, 6} U {3, 4} -> {3, 4, 5, 6}, count = 4
-        assert_eq!(result.value(0), 2);
-        assert_eq!(result.value(1), 1);
-        assert_eq!(result.value(2), 4);
-
-        Ok(())
-    }
-
-    #[test]
-    fn test_size() {
-        let acc = create_test_group_accumulator();
-        // Just test it doesn't crash and returns a value.
-        assert!(acc.size() > 0);
-    }
-}
--- a/src/common/function/src/function_registry.rs
+++ b/src/common/function/src/function_registry.rs
@@ -21,7 +21,6 @@ use once_cell::sync::Lazy;

 use crate::admin::AdminFunction;
 use crate::aggrs::approximate::ApproximateFunction;
-use crate::aggrs::count_hash::CountHash;
 use crate::aggrs::vector::VectorFunction as VectorAggrFunction;
 use crate::function::{AsyncFunctionRef, Function, FunctionRef};
 use crate::function_factory::ScalarFunctionFactory;
@@ -145,9 +144,6 @@ pub static FUNCTION_REGISTRY: Lazy<Arc<FunctionRegistry>> = Lazy::new(|| {
    // Approximate functions
    ApproximateFunction::register(&function_registry);

-    // CountHash function
-    CountHash::register(&function_registry);
-
    Arc::new(function_registry)
 });

--- 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, UInt32Vector, 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(UInt32Vector::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(UInt32Vector::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-expr/src/alter.rs
+++ b/src/common/grpc-expr/src/alter.rs
@@ -180,22 +180,6 @@ pub fn alter_expr_to_request(table_id: TableId, expr: AlterTableExpr) -> Result<
            },
            None => return MissingAlterIndexOptionSnafu.fail(),
        },
-        Kind::DropDefaults(o) => {
-            let names = o
-                .drop_defaults
-                .into_iter()
-                .map(|col| {
-                    ensure!(
-                        !col.column_name.is_empty(),
-                        MissingFieldSnafu {
-                            field: "column_name"
-                        }
-                    );
-                    Ok(col.column_name)
-                })
-                .collect::<Result<Vec<_>>>()?;
-            AlterKind::DropDefaults { names }
-        }
    };

    let request = AlterTableRequest {
--- a/src/common/grpc/src/channel_manager.rs
+++ b/src/common/grpc/src/channel_manager.rs
@@ -201,8 +201,8 @@ impl ChannelManager {
            "http"
        };

-        let mut endpoint = Endpoint::new(format!("{http_prefix}://{addr}"))
-            .context(CreateChannelSnafu { addr })?;
+        let mut endpoint =
+            Endpoint::new(format!("{http_prefix}://{addr}")).context(CreateChannelSnafu)?;

        if let Some(dur) = self.config().timeout {
            endpoint = endpoint.timeout(dur);
@@ -237,7 +237,7 @@ impl ChannelManager {
        if let Some(tls_config) = &self.inner.client_tls_config {
            endpoint = endpoint
                .tls_config(tls_config.clone())
-                .context(CreateChannelSnafu { addr })?;
+                .context(CreateChannelSnafu)?;
        }

        endpoint = endpoint
--- a/src/common/grpc/src/error.rs
+++ b/src/common/grpc/src/error.rs
@@ -52,9 +52,8 @@ pub enum Error {
        location: Location,
    },

-    #[snafu(display("Failed to create gRPC channel from '{addr}'"))]
+    #[snafu(display("Failed to create gRPC channel"))]
    CreateChannel {
-        addr: String,
        #[snafu(source)]
        error: tonic::transport::Error,
        #[snafu(implicit)]
--- a/src/common/meta/Cargo.toml
+++ b/src/common/meta/Cargo.toml
@@ -17,7 +17,7 @@ workspace = true
 anymap2 = "0.13.0"
 api.workspace = true
 async-recursion = "1.0"
-async-stream.workspace = true
+async-stream = "0.3"
 async-trait.workspace = true
 backon = { workspace = true, optional = true }
 base64.workspace = true
--- a/src/common/meta/src/ddl/alter_logical_tables.rs
+++ b/src/common/meta/src/ddl/alter_logical_tables.rs
@@ -25,7 +25,6 @@ use common_procedure::error::{FromJsonSnafu, Result as ProcedureResult, ToJsonSn
 use common_procedure::{Context, LockKey, Procedure, Status};
 use common_telemetry::{error, info, warn};
 use futures_util::future;
-pub use region_request::make_alter_region_request;
 use serde::{Deserialize, Serialize};
 use snafu::{ensure, ResultExt};
 use store_api::metadata::ColumnMetadata;
--- a/src/common/meta/src/ddl/alter_logical_tables/region_request.rs
+++ b/src/common/meta/src/ddl/alter_logical_tables/region_request.rs
@@ -12,18 +12,20 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use api::v1;
 use api::v1::alter_table_expr::Kind;
 use api::v1::region::{
    alter_request, region_request, AddColumn, AddColumns, AlterRequest, AlterRequests,
    RegionColumnDef, RegionRequest, RegionRequestHeader,
 };
-use api::v1::{self, AlterTableExpr};
 use common_telemetry::tracing_context::TracingContext;
 use store_api::storage::RegionId;

 use crate::ddl::alter_logical_tables::AlterLogicalTablesProcedure;
 use crate::error::Result;
+use crate::key::table_info::TableInfoValue;
 use crate::peer::Peer;
+use crate::rpc::ddl::AlterTableTask;
 use crate::rpc::router::{find_leader_regions, RegionRoute};

 impl AlterLogicalTablesProcedure {
@@ -60,37 +62,34 @@ impl AlterLogicalTablesProcedure {
        {
            for region_number in &regions_on_this_peer {
                let region_id = RegionId::new(table.table_info.ident.table_id, *region_number);
-                let request = make_alter_region_request(
-                    region_id,
-                    &task.alter_table,
-                    table.table_info.ident.version,
-                );
+                let request = self.make_alter_region_request(region_id, task, table)?;
                requests.push(request);
            }
        }

        Ok(AlterRequests { requests })
    }
-}

-/// Makes an alter region request.
-pub fn make_alter_region_request(
-    region_id: RegionId,
-    alter_table_expr: &AlterTableExpr,
-    schema_version: u64,
-) -> AlterRequest {
-    let region_id = region_id.as_u64();
-    let kind = match &alter_table_expr.kind {
-        Some(Kind::AddColumns(add_columns)) => Some(alter_request::Kind::AddColumns(
-            to_region_add_columns(add_columns),
-        )),
-        _ => unreachable!(), // Safety: we have checked the kind in check_input_tasks
-    };
+    fn make_alter_region_request(
+        &self,
+        region_id: RegionId,
+        task: &AlterTableTask,
+        table: &TableInfoValue,
+    ) -> Result<AlterRequest> {
+        let region_id = region_id.as_u64();
+        let schema_version = table.table_info.ident.version;
+        let kind = match &task.alter_table.kind {
+            Some(Kind::AddColumns(add_columns)) => Some(alter_request::Kind::AddColumns(
+                to_region_add_columns(add_columns),
+            )),
+            _ => unreachable!(), // Safety: we have checked the kind in check_input_tasks
+        };

-    AlterRequest {
-        region_id,
-        schema_version,
-        kind,
+        Ok(AlterRequest {
+            region_id,
+            schema_version,
+            kind,
+        })
    }
 }

--- a/src/common/meta/src/ddl/alter_table/region_request.rs
+++ b/src/common/meta/src/ddl/alter_table/region_request.rs
@@ -135,7 +135,6 @@ fn create_proto_alter_kind(
        Kind::UnsetTableOptions(v) => Ok(Some(alter_request::Kind::UnsetTableOptions(v.clone()))),
        Kind::SetIndex(v) => Ok(Some(alter_request::Kind::SetIndex(v.clone()))),
        Kind::UnsetIndex(v) => Ok(Some(alter_request::Kind::UnsetIndex(v.clone()))),
-        Kind::DropDefaults(v) => Ok(Some(alter_request::Kind::DropDefaults(v.clone()))),
    }
 }

--- a/src/common/meta/src/ddl/alter_table/update_metadata.rs
+++ b/src/common/meta/src/ddl/alter_table/update_metadata.rs
@@ -61,8 +61,7 @@ impl AlterTableProcedure {
            | AlterKind::SetTableOptions { .. }
            | AlterKind::UnsetTableOptions { .. }
            | AlterKind::SetIndex { .. }
-            | AlterKind::UnsetIndex { .. }
-            | AlterKind::DropDefaults { .. } => {}
+            | AlterKind::UnsetIndex { .. } => {}
        }

        Ok(new_info)
--- a/src/common/meta/src/ddl/create_logical_tables.rs
+++ b/src/common/meta/src/ddl/create_logical_tables.rs
@@ -25,7 +25,6 @@ use common_procedure::error::{FromJsonSnafu, Result as ProcedureResult, ToJsonSn
 use common_procedure::{Context as ProcedureContext, LockKey, Procedure, Status};
 use common_telemetry::{debug, error, warn};
 use futures::future;
-pub use region_request::create_region_request_builder;
 use serde::{Deserialize, Serialize};
 use snafu::{ensure, ResultExt};
 use store_api::metadata::ColumnMetadata;
--- a/src/common/meta/src/ddl/create_logical_tables/region_request.rs
+++ b/src/common/meta/src/ddl/create_logical_tables/region_request.rs
@@ -15,16 +15,16 @@
 use std::collections::HashMap;

 use api::v1::region::{region_request, CreateRequests, RegionRequest, RegionRequestHeader};
-use api::v1::CreateTableExpr;
 use common_telemetry::debug;
 use common_telemetry::tracing_context::TracingContext;
-use store_api::storage::{RegionId, TableId};
+use store_api::storage::RegionId;

 use crate::ddl::create_logical_tables::CreateLogicalTablesProcedure;
 use crate::ddl::create_table_template::{build_template, CreateRequestBuilder};
 use crate::ddl::utils::region_storage_path;
 use crate::error::Result;
 use crate::peer::Peer;
+use crate::rpc::ddl::CreateTableTask;
 use crate::rpc::router::{find_leader_regions, RegionRoute};

 impl CreateLogicalTablesProcedure {
@@ -45,15 +45,13 @@ impl CreateLogicalTablesProcedure {
            let catalog = &create_table_expr.catalog_name;
            let schema = &create_table_expr.schema_name;
            let logical_table_id = task.table_info.ident.table_id;
-            let physical_table_id = self.data.physical_table_id;
            let storage_path = region_storage_path(catalog, schema);
-            let request_builder =
-                create_region_request_builder(&task.create_table, physical_table_id)?;
+            let request_builder = self.create_region_request_builder(task)?;

            for region_number in &regions_on_this_peer {
                let region_id = RegionId::new(logical_table_id, *region_number);
                let one_region_request =
-                    request_builder.build_one(region_id, storage_path.clone(), &HashMap::new());
+                    request_builder.build_one(region_id, storage_path.clone(), &HashMap::new())?;
                requests.push(one_region_request);
            }
        }
@@ -71,13 +69,16 @@ impl CreateLogicalTablesProcedure {
            body: Some(region_request::Body::Creates(CreateRequests { requests })),
        }))
    }
-}

-/// Creates a region request builder.
-pub fn create_region_request_builder(
-    create_table_expr: &CreateTableExpr,
-    physical_table_id: TableId,
-) -> Result<CreateRequestBuilder> {
-    let template = build_template(create_table_expr)?;
-    Ok(CreateRequestBuilder::new(template, Some(physical_table_id)))
+    fn create_region_request_builder(
+        &self,
+        task: &CreateTableTask,
+    ) -> Result<CreateRequestBuilder> {
+        let create_expr = &task.create_table;
+        let template = build_template(create_expr)?;
+        Ok(CreateRequestBuilder::new(
+            template,
+            Some(self.data.physical_table_id),
+        ))
+    }
 }
--- a/src/common/meta/src/ddl/create_table.rs
+++ b/src/common/meta/src/ddl/create_table.rs
@@ -218,8 +218,11 @@ impl CreateTableProcedure {
            let mut requests = Vec::with_capacity(regions.len());
            for region_number in regions {
                let region_id = RegionId::new(self.table_id(), region_number);
-                let create_region_request =
-                    request_builder.build_one(region_id, storage_path.clone(), region_wal_options);
+                let create_region_request = request_builder.build_one(
+                    region_id,
+                    storage_path.clone(),
+                    region_wal_options,
+                )?;
                requests.push(PbRegionRequest::Create(create_region_request));
            }

--- a/src/common/meta/src/ddl/create_table_template.rs
+++ b/src/common/meta/src/ddl/create_table_template.rs
@@ -105,12 +105,12 @@ impl CreateRequestBuilder {
        &self.template
    }

-    pub fn build_one(
+    pub(crate) fn build_one(
        &self,
        region_id: RegionId,
        storage_path: String,
        region_wal_options: &HashMap<RegionNumber, String>,
-    ) -> CreateRequest {
+    ) -> Result<CreateRequest> {
        let mut request = self.template.clone();

        request.region_id = region_id.as_u64();
@@ -130,6 +130,6 @@ impl CreateRequestBuilder {
            );
        }

-        request
+        Ok(request)
    }
 }
--- a/src/common/meta/src/ddl/drop_database/executor.rs
+++ b/src/common/meta/src/ddl/drop_database/executor.rs
@@ -13,6 +13,7 @@
 // limitations under the License.

 use std::any::Any;
+use std::collections::HashMap;

 use common_procedure::Status;
 use common_telemetry::info;
@@ -24,7 +25,7 @@ use table::table_name::TableName;
 use crate::ddl::drop_database::cursor::DropDatabaseCursor;
 use crate::ddl::drop_database::{DropDatabaseContext, DropTableTarget, State};
 use crate::ddl::drop_table::executor::DropTableExecutor;
-use crate::ddl::utils::get_region_wal_options;
+use crate::ddl::utils::extract_region_wal_options;
 use crate::ddl::DdlContext;
 use crate::error::{self, Result};
 use crate::key::table_route::TableRouteValue;
@@ -108,12 +109,17 @@ impl State for DropDatabaseExecutor {
        );

        // Deletes topic-region mapping if dropping physical table
-        let region_wal_options = get_region_wal_options(
-            &ddl_ctx.table_metadata_manager,
-            &table_route_value,
-            self.physical_table_id,
-        )
-        .await?;
+        let region_wal_options =
+            if let TableRouteValue::Physical(table_route_value) = &table_route_value {
+                let datanode_table_values = ddl_ctx
+                    .table_metadata_manager
+                    .datanode_table_manager()
+                    .regions(self.physical_table_id, table_route_value)
+                    .await?;
+                extract_region_wal_options(&datanode_table_values)?
+            } else {
+                HashMap::new()
+            };

        executor
            .on_destroy_metadata(ddl_ctx, &table_route_value, &region_wal_options)
--- a/src/common/meta/src/ddl/test_util/datanode_handler.rs
+++ b/src/common/meta/src/ddl/test_util/datanode_handler.rs
@@ -32,7 +32,6 @@ impl MockDatanodeHandler for () {
        Ok(RegionResponse {
            affected_rows: 0,
            extensions: Default::default(),
-            metadata: Vec::new(),
        })
    }

--- a/src/common/meta/src/ddl/utils.rs
+++ b/src/common/meta/src/ddl/utils.rs
@@ -42,8 +42,7 @@ use crate::error::{
 };
 use crate::key::datanode_table::DatanodeTableValue;
 use crate::key::table_name::TableNameKey;
-use crate::key::table_route::TableRouteValue;
-use crate::key::{TableMetadataManager, TableMetadataManagerRef};
+use crate::key::TableMetadataManagerRef;
 use crate::peer::Peer;
 use crate::rpc::ddl::CreateTableTask;
 use crate::rpc::router::{find_follower_regions, find_followers, RegionRoute};
@@ -188,25 +187,6 @@ pub fn parse_region_wal_options(
    Ok(region_wal_options)
 }

-/// Gets the wal options for a table.
-pub async fn get_region_wal_options(
-    table_metadata_manager: &TableMetadataManager,
-    table_route_value: &TableRouteValue,
-    physical_table_id: TableId,
-) -> Result<HashMap<RegionNumber, WalOptions>> {
-    let region_wal_options =
-        if let TableRouteValue::Physical(table_route_value) = &table_route_value {
-            let datanode_table_values = table_metadata_manager
-                .datanode_table_manager()
-                .regions(physical_table_id, table_route_value)
-                .await?;
-            extract_region_wal_options(&datanode_table_values)?
-        } else {
-            HashMap::new()
-        };
-    Ok(region_wal_options)
-}
-
 /// Extracts region wal options from [DatanodeTableValue]s.
 pub fn extract_region_wal_options(
    datanode_table_values: &Vec<DatanodeTableValue>,
--- a/src/common/meta/src/ddl_manager.rs
+++ b/src/common/meta/src/ddl_manager.rs
@@ -50,11 +50,7 @@ use crate::key::{DeserializedValueWithBytes, TableMetadataManagerRef};
 #[cfg(feature = "enterprise")]
 use crate::rpc::ddl::trigger::CreateTriggerTask;
 #[cfg(feature = "enterprise")]
-use crate::rpc::ddl::trigger::DropTriggerTask;
-#[cfg(feature = "enterprise")]
 use crate::rpc::ddl::DdlTask::CreateTrigger;
-#[cfg(feature = "enterprise")]
-use crate::rpc::ddl::DdlTask::DropTrigger;
 use crate::rpc::ddl::DdlTask::{
    AlterDatabase, AlterLogicalTables, AlterTable, CreateDatabase, CreateFlow, CreateLogicalTables,
    CreateTable, CreateView, DropDatabase, DropFlow, DropLogicalTables, DropTable, DropView,
@@ -95,14 +91,6 @@ pub trait TriggerDdlManager: Send + Sync {
        query_context: QueryContext,
    ) -> Result<SubmitDdlTaskResponse>;

-    async fn drop_trigger(
-        &self,
-        drop_trigger_task: DropTriggerTask,
-        procedure_manager: ProcedureManagerRef,
-        ddl_context: DdlContext,
-        query_context: QueryContext,
-    ) -> Result<SubmitDdlTaskResponse>;
-
    fn as_any(&self) -> &dyn std::any::Any;
 }

@@ -137,12 +125,13 @@ impl DdlManager {
        ddl_context: DdlContext,
        procedure_manager: ProcedureManagerRef,
        register_loaders: bool,
+        #[cfg(feature = "enterprise")] trigger_ddl_manager: Option<TriggerDdlManagerRef>,
    ) -> Result<Self> {
        let manager = Self {
            ddl_context,
            procedure_manager,
            #[cfg(feature = "enterprise")]
-            trigger_ddl_manager: None,
+            trigger_ddl_manager,
        };
        if register_loaders {
            manager.register_loaders()?;
@@ -150,15 +139,6 @@ impl DdlManager {
        Ok(manager)
    }

-    #[cfg(feature = "enterprise")]
-    pub fn with_trigger_ddl_manager(
-        mut self,
-        trigger_ddl_manager: Option<TriggerDdlManagerRef>,
-    ) -> Self {
-        self.trigger_ddl_manager = trigger_ddl_manager;
-        self
-    }
-
    /// Returns the [TableMetadataManagerRef].
    pub fn table_metadata_manager(&self) -> &TableMetadataManagerRef {
        &self.ddl_context.table_metadata_manager
@@ -660,28 +640,6 @@ async fn handle_drop_flow_task(
    })
 }

-#[cfg(feature = "enterprise")]
-async fn handle_drop_trigger_task(
-    ddl_manager: &DdlManager,
-    drop_trigger_task: DropTriggerTask,
-    query_context: QueryContext,
-) -> Result<SubmitDdlTaskResponse> {
-    let Some(m) = ddl_manager.trigger_ddl_manager.as_ref() else {
-        return UnsupportedSnafu {
-            operation: "drop trigger",
-        }
-        .fail();
-    };
-
-    m.drop_trigger(
-        drop_trigger_task,
-        ddl_manager.procedure_manager.clone(),
-        ddl_manager.ddl_context.clone(),
-        query_context,
-    )
-    .await
-}
-
 async fn handle_drop_view_task(
    ddl_manager: &DdlManager,
    drop_view_task: DropViewTask,
@@ -869,11 +827,6 @@ impl ProcedureExecutor for DdlManager {
                    handle_create_flow_task(self, create_flow_task, request.query_context.into())
                        .await
                }
-                DropFlow(drop_flow_task) => handle_drop_flow_task(self, drop_flow_task).await,
-                CreateView(create_view_task) => {
-                    handle_create_view_task(self, create_view_task).await
-                }
-                DropView(drop_view_task) => handle_drop_view_task(self, drop_view_task).await,
                #[cfg(feature = "enterprise")]
                CreateTrigger(create_trigger_task) => {
                    handle_create_trigger_task(
@@ -883,11 +836,11 @@ impl ProcedureExecutor for DdlManager {
                    )
                    .await
                }
-                #[cfg(feature = "enterprise")]
-                DropTrigger(drop_trigger_task) => {
-                    handle_drop_trigger_task(self, drop_trigger_task, request.query_context.into())
-                        .await
+                DropFlow(drop_flow_task) => handle_drop_flow_task(self, drop_flow_task).await,
+                CreateView(create_view_task) => {
+                    handle_create_view_task(self, create_view_task).await
                }
+                DropView(drop_view_task) => handle_drop_view_task(self, drop_view_task).await,
            }
        }
        .trace(span)
@@ -1011,6 +964,8 @@ mod tests {
            },
            procedure_manager.clone(),
            true,
+            #[cfg(feature = "enterprise")]
+            None,
        );

        let expected_loaders = vec![
--- a/src/common/meta/src/key.rs
+++ b/src/common/meta/src/key.rs
@@ -109,7 +109,7 @@ pub mod table_name;
 pub mod table_route;
 #[cfg(any(test, feature = "testing"))]
 pub mod test_utils;
-pub mod tombstone;
+mod tombstone;
 pub mod topic_name;
 pub mod topic_region;
 pub mod txn_helper;
@@ -535,29 +535,6 @@ impl TableMetadataManager {
        }
    }

-    /// Creates a new `TableMetadataManager` with a custom tombstone prefix.
-    pub fn new_with_custom_tombstone_prefix(
-        kv_backend: KvBackendRef,
-        tombstone_prefix: &str,
-    ) -> Self {
-        Self {
-            table_name_manager: TableNameManager::new(kv_backend.clone()),
-            table_info_manager: TableInfoManager::new(kv_backend.clone()),
-            view_info_manager: ViewInfoManager::new(kv_backend.clone()),
-            datanode_table_manager: DatanodeTableManager::new(kv_backend.clone()),
-            catalog_manager: CatalogManager::new(kv_backend.clone()),
-            schema_manager: SchemaManager::new(kv_backend.clone()),
-            table_route_manager: TableRouteManager::new(kv_backend.clone()),
-            tombstone_manager: TombstoneManager::new_with_prefix(
-                kv_backend.clone(),
-                tombstone_prefix,
-            ),
-            topic_name_manager: TopicNameManager::new(kv_backend.clone()),
-            topic_region_manager: TopicRegionManager::new(kv_backend.clone()),
-            kv_backend,
-        }
-    }
-
    pub async fn init(&self) -> Result<()> {
        let catalog_name = CatalogNameKey::new(DEFAULT_CATALOG_NAME);

@@ -948,7 +925,7 @@ impl TableMetadataManager {
    ) -> Result<()> {
        let keys =
            self.table_metadata_keys(table_id, table_name, table_route_value, region_wal_options)?;
-        self.tombstone_manager.create(keys).await.map(|_| ())
+        self.tombstone_manager.create(keys).await
    }

    /// Deletes metadata tombstone for table **permanently**.
@@ -962,10 +939,7 @@ impl TableMetadataManager {
    ) -> Result<()> {
        let table_metadata_keys =
            self.table_metadata_keys(table_id, table_name, table_route_value, region_wal_options)?;
-        self.tombstone_manager
-            .delete(table_metadata_keys)
-            .await
-            .map(|_| ())
+        self.tombstone_manager.delete(table_metadata_keys).await
    }

    /// Restores metadata for table.
@@ -979,7 +953,7 @@ impl TableMetadataManager {
    ) -> Result<()> {
        let keys =
            self.table_metadata_keys(table_id, table_name, table_route_value, region_wal_options)?;
-        self.tombstone_manager.restore(keys).await.map(|_| ())
+        self.tombstone_manager.restore(keys).await
    }

    /// Deletes metadata for table **permanently**.
--- a/src/common/meta/src/key/table_route.rs
+++ b/src/common/meta/src/key/table_route.rs
@@ -48,11 +48,6 @@ impl TableRouteKey {
    pub fn new(table_id: TableId) -> Self {
        Self { table_id }
    }
-
-    /// Returns the range prefix of the table route key.
-    pub fn range_prefix() -> Vec<u8> {
-        format!("{}/", TABLE_ROUTE_PREFIX).into_bytes()
-    }
 }

 #[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]
--- a/src/common/meta/src/key/tombstone.rs
+++ b/src/common/meta/src/key/tombstone.rs
@@ -25,32 +25,20 @@ use crate::rpc::store::BatchGetRequest;
 /// [TombstoneManager] provides the ability to:
 /// - logically delete values
 /// - restore the deleted values
-pub struct TombstoneManager {
+pub(crate) struct TombstoneManager {
    kv_backend: KvBackendRef,
-    tombstone_prefix: String,
 }

 const TOMBSTONE_PREFIX: &str = "__tombstone/";

+fn to_tombstone(key: &[u8]) -> Vec<u8> {
+    [TOMBSTONE_PREFIX.as_bytes(), key].concat()
+}
+
 impl TombstoneManager {
    /// Returns [TombstoneManager].
    pub fn new(kv_backend: KvBackendRef) -> Self {
-        Self {
-            kv_backend,
-            tombstone_prefix: TOMBSTONE_PREFIX.to_string(),
-        }
-    }
-
-    /// Returns [TombstoneManager] with a custom tombstone prefix.
-    pub fn new_with_prefix(kv_backend: KvBackendRef, prefix: &str) -> Self {
-        Self {
-            kv_backend,
-            tombstone_prefix: prefix.to_string(),
-        }
-    }
-
-    pub fn to_tombstone(&self, key: &[u8]) -> Vec<u8> {
-        [self.tombstone_prefix.as_bytes(), key].concat()
+        Self { kv_backend }
    }

    /// Moves value to `dest_key`.
@@ -79,7 +67,7 @@ impl TombstoneManager {
        (txn, TxnOpGetResponseSet::filter(src_key))
    }

-    async fn move_values_inner(&self, keys: &[Vec<u8>], dest_keys: &[Vec<u8>]) -> Result<usize> {
+    async fn move_values_inner(&self, keys: &[Vec<u8>], dest_keys: &[Vec<u8>]) -> Result<()> {
        ensure!(
            keys.len() == dest_keys.len(),
            error::UnexpectedSnafu {
@@ -114,7 +102,7 @@ impl TombstoneManager {
                .unzip();
            let mut resp = self.kv_backend.txn(Txn::merge_all(txns)).await?;
            if resp.succeeded {
-                return Ok(keys.len());
+                return Ok(());
            }
            let mut set = TxnOpGetResponseSet::from(&mut resp.responses);
            // Updates results.
@@ -137,9 +125,7 @@ impl TombstoneManager {
    }

    /// Moves values to `dest_key`.
-    ///
-    /// Returns the number of keys that were moved.
-    async fn move_values(&self, keys: Vec<Vec<u8>>, dest_keys: Vec<Vec<u8>>) -> Result<usize> {
+    async fn move_values(&self, keys: Vec<Vec<u8>>, dest_keys: Vec<Vec<u8>>) -> Result<()> {
        let chunk_size = self.kv_backend.max_txn_ops() / 2;
        if keys.len() > chunk_size {
            let keys_chunks = keys.chunks(chunk_size).collect::<Vec<_>>();
@@ -148,7 +134,7 @@ impl TombstoneManager {
                self.move_values_inner(keys, dest_keys).await?;
            }

-            Ok(keys.len())
+            Ok(())
        } else {
            self.move_values_inner(&keys, &dest_keys).await
        }
@@ -159,13 +145,11 @@ impl TombstoneManager {
    /// Preforms to:
    /// - deletes origin values.
    /// - stores tombstone values.
-    ///
-    /// Returns the number of keys that were moved.
-    pub async fn create(&self, keys: Vec<Vec<u8>>) -> Result<usize> {
+    pub(crate) async fn create(&self, keys: Vec<Vec<u8>>) -> Result<()> {
        let (keys, dest_keys): (Vec<_>, Vec<_>) = keys
            .into_iter()
            .map(|key| {
-                let tombstone_key = self.to_tombstone(&key);
+                let tombstone_key = to_tombstone(&key);
                (key, tombstone_key)
            })
            .unzip();
@@ -178,13 +162,11 @@ impl TombstoneManager {
    /// Preforms to:
    /// - restore origin value.
    /// - deletes tombstone values.
-    ///
-    /// Returns the number of keys that were restored.
-    pub async fn restore(&self, keys: Vec<Vec<u8>>) -> Result<usize> {
+    pub(crate) async fn restore(&self, keys: Vec<Vec<u8>>) -> Result<()> {
        let (keys, dest_keys): (Vec<_>, Vec<_>) = keys
            .into_iter()
            .map(|key| {
-                let tombstone_key = self.to_tombstone(&key);
+                let tombstone_key = to_tombstone(&key);
                (tombstone_key, key)
            })
            .unzip();
@@ -193,18 +175,16 @@ impl TombstoneManager {
    }

    /// Deletes tombstones values for the specified `keys`.
-    ///
-    /// Returns the number of keys that were deleted.
-    pub async fn delete(&self, keys: Vec<Vec<u8>>) -> Result<usize> {
+    pub(crate) async fn delete(&self, keys: Vec<Vec<u8>>) -> Result<()> {
        let operations = keys
            .iter()
-            .map(|key| TxnOp::Delete(self.to_tombstone(key)))
+            .map(|key| TxnOp::Delete(to_tombstone(key)))
            .collect::<Vec<_>>();

        let txn = Txn::new().and_then(operations);
        // Always success.
        let _ = self.kv_backend.txn(txn).await?;
-        Ok(keys.len())
+        Ok(())
    }
 }

@@ -214,6 +194,7 @@ mod tests {
    use std::collections::HashMap;
    use std::sync::Arc;

+    use super::to_tombstone;
    use crate::error::Error;
    use crate::key::tombstone::TombstoneManager;
    use crate::kv_backend::memory::MemoryKvBackend;
@@ -265,7 +246,7 @@ mod tests {
        assert!(!kv_backend.exists(b"foo").await.unwrap());
        assert_eq!(
            kv_backend
-                .get(&tombstone_manager.to_tombstone(b"bar"))
+                .get(&to_tombstone(b"bar"))
                .await
                .unwrap()
                .unwrap()
@@ -274,7 +255,7 @@ mod tests {
        );
        assert_eq!(
            kv_backend
-                .get(&tombstone_manager.to_tombstone(b"foo"))
+                .get(&to_tombstone(b"foo"))
                .await
                .unwrap()
                .unwrap()
@@ -306,7 +287,7 @@ mod tests {
            kv_backend.clone(),
            &[MoveValue {
                key: b"bar".to_vec(),
-                dest_key: tombstone_manager.to_tombstone(b"bar"),
+                dest_key: to_tombstone(b"bar"),
                value: b"baz".to_vec(),
            }],
        )
@@ -383,7 +364,7 @@ mod tests {
            .iter()
            .map(|(key, value)| MoveValue {
                key: key.clone(),
-                dest_key: tombstone_manager.to_tombstone(key),
+                dest_key: to_tombstone(key),
                value: value.clone(),
            })
            .collect::<Vec<_>>();
@@ -428,7 +409,7 @@ mod tests {
            .iter()
            .map(|(key, value)| MoveValue {
                key: key.clone(),
-                dest_key: tombstone_manager.to_tombstone(key),
+                dest_key: to_tombstone(key),
                value: value.clone(),
            })
            .collect::<Vec<_>>();
@@ -481,7 +462,7 @@ mod tests {
            .iter()
            .map(|(key, value)| MoveValue {
                key: key.clone(),
-                dest_key: tombstone_manager.to_tombstone(key),
+                dest_key: to_tombstone(key),
                value: value.clone(),
            })
            .collect::<Vec<_>>();
@@ -521,7 +502,7 @@ mod tests {
            .iter()
            .map(|(key, value)| MoveValue {
                key: key.clone(),
-                dest_key: tombstone_manager.to_tombstone(key),
+                dest_key: to_tombstone(key),
                value: value.clone(),
            })
            .collect::<Vec<_>>();
@@ -556,7 +537,7 @@ mod tests {
            .iter()
            .map(|(key, value)| MoveValue {
                key: key.clone(),
-                dest_key: tombstone_manager.to_tombstone(key),
+                dest_key: to_tombstone(key),
                value: value.clone(),
            })
            .collect::<Vec<_>>();
--- a/src/common/meta/src/kv_backend/memory.rs
+++ b/src/common/meta/src/kv_backend/memory.rs
@@ -54,20 +54,20 @@ impl<T> MemoryKvBackend<T> {
        kvs.clear();
    }

-    #[cfg(any(test, feature = "testing"))]
+    #[cfg(test)]
    /// Returns true if the `kvs` is empty.
    pub fn is_empty(&self) -> bool {
        self.kvs.read().unwrap().is_empty()
    }

-    #[cfg(any(test, feature = "testing"))]
+    #[cfg(test)]
    /// Returns the `kvs`.
    pub fn dump(&self) -> BTreeMap<Vec<u8>, Vec<u8>> {
        let kvs = self.kvs.read().unwrap();
        kvs.clone()
    }

-    #[cfg(any(test, feature = "testing"))]
+    #[cfg(test)]
    /// Returns the length of `kvs`
    pub fn len(&self) -> usize {
        self.kvs.read().unwrap().len()
--- 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/rpc/ddl.rs
+++ b/src/common/meta/src/rpc/ddl.rs
@@ -69,8 +69,6 @@ pub enum DdlTask {
    AlterDatabase(AlterDatabaseTask),
    CreateFlow(CreateFlowTask),
    DropFlow(DropFlowTask),
-    #[cfg(feature = "enterprise")]
-    DropTrigger(trigger::DropTriggerTask),
    CreateView(CreateViewTask),
    DropView(DropViewTask),
    #[cfg(feature = "enterprise")]
@@ -261,18 +259,6 @@ impl TryFrom<Task> for DdlTask {
                    .fail()
                }
            }
-            Task::DropTriggerTask(drop_trigger) => {
-                #[cfg(feature = "enterprise")]
-                return Ok(DdlTask::DropTrigger(drop_trigger.try_into()?));
-                #[cfg(not(feature = "enterprise"))]
-                {
-                    let _ = drop_trigger;
-                    crate::error::UnsupportedSnafu {
-                        operation: "drop trigger",
-                    }
-                    .fail()
-                }
-            }
        }
    }
 }
@@ -325,8 +311,6 @@ impl TryFrom<SubmitDdlTaskRequest> for PbDdlTaskRequest {
            DdlTask::DropView(task) => Task::DropViewTask(task.into()),
            #[cfg(feature = "enterprise")]
            DdlTask::CreateTrigger(task) => Task::CreateTriggerTask(task.into()),
-            #[cfg(feature = "enterprise")]
-            DdlTask::DropTrigger(task) => Task::DropTriggerTask(task.into()),
        };

        Ok(Self {
--- a/src/common/meta/src/rpc/ddl/trigger.rs
+++ b/src/common/meta/src/rpc/ddl/trigger.rs
@@ -1,13 +1,10 @@
 use std::collections::HashMap;
 use std::time::Duration;

-use api::v1::meta::{
-    CreateTriggerTask as PbCreateTriggerTask, DropTriggerTask as PbDropTriggerTask,
-};
+use api::v1::meta::CreateTriggerTask as PbCreateTriggerTask;
 use api::v1::notify_channel::ChannelType as PbChannelType;
 use api::v1::{
-    CreateTriggerExpr as PbCreateTriggerExpr, DropTriggerExpr as PbDropTriggerExpr,
-    NotifyChannel as PbNotifyChannel, WebhookOptions as PbWebhookOptions,
+    CreateTriggerExpr, NotifyChannel as PbNotifyChannel, WebhookOptions as PbWebhookOptions,
 };
 use serde::{Deserialize, Serialize};
 use snafu::OptionExt;
@@ -59,7 +56,7 @@ impl From<CreateTriggerTask> for PbCreateTriggerTask {
            .map(PbNotifyChannel::from)
            .collect();

-        let expr = PbCreateTriggerExpr {
+        let expr = CreateTriggerExpr {
            catalog_name: task.catalog_name,
            trigger_name: task.trigger_name,
            create_if_not_exists: task.if_not_exists,
@@ -142,86 +139,17 @@ impl TryFrom<PbNotifyChannel> for NotifyChannel {
    }
 }

-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct DropTriggerTask {
-    pub catalog_name: String,
-    pub trigger_name: String,
-    pub drop_if_exists: bool,
-}
-
-impl From<DropTriggerTask> for PbDropTriggerTask {
-    fn from(task: DropTriggerTask) -> Self {
-        let expr = PbDropTriggerExpr {
-            catalog_name: task.catalog_name,
-            trigger_name: task.trigger_name,
-            drop_if_exists: task.drop_if_exists,
-        };
-
-        PbDropTriggerTask {
-            drop_trigger: Some(expr),
-        }
-    }
-}
-
-impl TryFrom<PbDropTriggerTask> for DropTriggerTask {
-    type Error = error::Error;
-
-    fn try_from(task: PbDropTriggerTask) -> Result<Self> {
-        let expr = task.drop_trigger.context(error::InvalidProtoMsgSnafu {
-            err_msg: "expected drop_trigger",
-        })?;
-
-        Ok(DropTriggerTask {
-            catalog_name: expr.catalog_name,
-            trigger_name: expr.trigger_name,
-            drop_if_exists: expr.drop_if_exists,
-        })
-    }
-}
-
 impl DdlTask {
    /// Creates a [`DdlTask`] to create a trigger.
    pub fn new_create_trigger(expr: CreateTriggerTask) -> Self {
        DdlTask::CreateTrigger(expr)
    }
-
-    /// Creates a [`DdlTask`] to drop a trigger.
-    pub fn new_drop_trigger(expr: DropTriggerTask) -> Self {
-        DdlTask::DropTrigger(expr)
-    }
 }

 #[cfg(test)]
 mod tests {
    use super::*;

-    #[test]
-    fn test_convert_drop_trigger_task() {
-        let original = DropTriggerTask {
-            catalog_name: "test_catalog".to_string(),
-            trigger_name: "test_trigger".to_string(),
-            drop_if_exists: true,
-        };
-
-        let pb_task: PbDropTriggerTask = original.clone().into();
-
-        let expr = pb_task.drop_trigger.as_ref().unwrap();
-        assert_eq!(expr.catalog_name, "test_catalog");
-        assert_eq!(expr.trigger_name, "test_trigger");
-        assert!(expr.drop_if_exists);
-
-        let round_tripped = DropTriggerTask::try_from(pb_task).unwrap();
-
-        assert_eq!(original.catalog_name, round_tripped.catalog_name);
-        assert_eq!(original.trigger_name, round_tripped.trigger_name);
-        assert_eq!(original.drop_if_exists, round_tripped.drop_if_exists);
-
-        // Test invalid case where drop_trigger is None
-        let invalid_task = PbDropTriggerTask { drop_trigger: None };
-        let result = DropTriggerTask::try_from(invalid_task);
-        assert!(result.is_err());
-    }
-
    #[test]
    fn test_convert_create_trigger_task() {
        let original = CreateTriggerTask {
--- 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/query/src/lib.rs
+++ b/src/common/query/src/lib.rs
@@ -14,7 +14,7 @@

 pub mod columnar_value;
 pub mod error;
-pub mod function;
+mod function;
 pub mod logical_plan;
 pub mod prelude;
 pub mod request;
--- 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/common/telemetry/Cargo.toml
+++ b/src/common/telemetry/Cargo.toml
@@ -12,6 +12,7 @@ deadlock_detection = ["parking_lot/deadlock_detection"]
 workspace = true

 [dependencies]
+atty = "0.2"
 backtrace = "0.3"
 common-error.workspace = true
 console-subscriber = { version = "0.1", optional = true }
@@ -22,7 +23,7 @@ once_cell.workspace = true
 opentelemetry = { version = "0.21.0", default-features = false, features = [
    "trace",
 ] }
-opentelemetry-otlp = { version = "0.14.0", features = ["tokio", "http-proto", "reqwest-client"] }
+opentelemetry-otlp = { version = "0.14.0", features = ["tokio"] }
 opentelemetry-semantic-conventions = "0.13.0"
 opentelemetry_sdk = { version = "0.21.0", features = ["rt-tokio"] }
 parking_lot.workspace = true
--- a/src/common/telemetry/src/logging.rs
+++ b/src/common/telemetry/src/logging.rs
@@ -14,13 +14,12 @@

 //! logging stuffs, inspired by databend
 use std::env;
-use std::io::IsTerminal;
 use std::sync::{Arc, Mutex, Once};
 use std::time::Duration;

 use once_cell::sync::{Lazy, OnceCell};
 use opentelemetry::{global, KeyValue};
-use opentelemetry_otlp::{Protocol, SpanExporterBuilder, WithExportConfig};
+use opentelemetry_otlp::WithExportConfig;
 use opentelemetry_sdk::propagation::TraceContextPropagator;
 use opentelemetry_sdk::trace::Sampler;
 use opentelemetry_semantic_conventions::resource;
@@ -36,11 +35,7 @@ use tracing_subscriber::{filter, EnvFilter, Registry};

 use crate::tracing_sampler::{create_sampler, TracingSampleOptions};

-/// The default endpoint when use gRPC exporter protocol.
-pub const DEFAULT_OTLP_GRPC_ENDPOINT: &str = "http://localhost:4317";
-
-/// The default endpoint when use HTTP exporter protocol.
-pub const DEFAULT_OTLP_HTTP_ENDPOINT: &str = "http://localhost:4318";
+pub const DEFAULT_OTLP_ENDPOINT: &str = "http://localhost:4317";

 /// The default logs directory.
 pub const DEFAULT_LOGGING_DIR: &str = "logs";
@@ -71,25 +66,11 @@ pub struct LoggingOptions {
    /// Whether to enable tracing with OTLP. Default is false.
    pub enable_otlp_tracing: bool,

-    /// The endpoint of OTLP. Default is "http://localhost:4318".
+    /// The endpoint of OTLP. Default is "http://localhost:4317".
    pub otlp_endpoint: Option<String>,

    /// The tracing sample ratio.
    pub tracing_sample_ratio: Option<TracingSampleOptions>,
-
-    /// The protocol of OTLP export.
-    pub otlp_export_protocol: Option<OtlpExportProtocol>,
-}
-
-/// The protocol of OTLP export.
-#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
-#[serde(rename_all = "snake_case")]
-pub enum OtlpExportProtocol {
-    /// GRPC protocol.
-    Grpc,
-
-    /// HTTP protocol with binary protobuf.
-    Http,
 }

 /// The options of slow query.
@@ -165,7 +146,6 @@ impl Default for LoggingOptions {
            append_stdout: true,
            // Rotation hourly, 24 files per day, keeps info log files of 30 days
            max_log_files: 720,
-            otlp_export_protocol: None,
        }
    }
 }
@@ -241,14 +221,14 @@ pub fn init_global_logging(
                    Layer::new()
                        .json()
                        .with_writer(writer)
-                        .with_ansi(std::io::stdout().is_terminal())
+                        .with_ansi(atty::is(atty::Stream::Stdout))
                        .boxed(),
                )
            } else {
                Some(
                    Layer::new()
                        .with_writer(writer)
-                        .with_ansi(std::io::stdout().is_terminal())
+                        .with_ansi(atty::is(atty::Stream::Stdout))
                        .boxed(),
                )
            }
@@ -407,9 +387,22 @@ pub fn init_global_logging(
                    KeyValue::new(resource::PROCESS_PID, std::process::id().to_string()),
                ]));

+            let exporter = opentelemetry_otlp::new_exporter().tonic().with_endpoint(
+                opts.otlp_endpoint
+                    .as_ref()
+                    .map(|e| {
+                        if e.starts_with("http") {
+                            e.to_string()
+                        } else {
+                            format!("http://{}", e)
+                        }
+                    })
+                    .unwrap_or(DEFAULT_OTLP_ENDPOINT.to_string()),
+            );
+
            let tracer = opentelemetry_otlp::new_pipeline()
                .tracing()
-                .with_exporter(build_otlp_exporter(opts))
+                .with_exporter(exporter)
                .with_trace_config(trace_config)
                .install_batch(opentelemetry_sdk::runtime::Tokio)
                .expect("otlp tracer install failed");
@@ -427,42 +420,6 @@ pub fn init_global_logging(
    guards
 }

-fn build_otlp_exporter(opts: &LoggingOptions) -> SpanExporterBuilder {
-    let protocol = opts
-        .otlp_export_protocol
-        .clone()
-        .unwrap_or(OtlpExportProtocol::Http);
-
-    let endpoint = opts
-        .otlp_endpoint
-        .as_ref()
-        .map(|e| {
-            if e.starts_with("http") {
-                e.to_string()
-            } else {
-                format!("http://{}", e)
-            }
-        })
-        .unwrap_or_else(|| match protocol {
-            OtlpExportProtocol::Grpc => DEFAULT_OTLP_GRPC_ENDPOINT.to_string(),
-            OtlpExportProtocol::Http => DEFAULT_OTLP_HTTP_ENDPOINT.to_string(),
-        });
-
-    match protocol {
-        OtlpExportProtocol::Grpc => SpanExporterBuilder::Tonic(
-            opentelemetry_otlp::new_exporter()
-                .tonic()
-                .with_endpoint(endpoint),
-        ),
-        OtlpExportProtocol::Http => SpanExporterBuilder::Http(
-            opentelemetry_otlp::new_exporter()
-                .http()
-                .with_endpoint(endpoint)
-                .with_protocol(Protocol::HttpBinary),
-        ),
-    }
-}
-
 fn build_slow_query_logger<S>(
    opts: &LoggingOptions,
    slow_query_opts: Option<&SlowQueryOptions>,
--- a/src/datanode/src/config.rs
+++ b/src/datanode/src/config.rs
@@ -14,7 +14,10 @@

 //! Datanode configurations

+use core::time::Duration;
+
 use common_base::readable_size::ReadableSize;
+use common_base::secrets::{ExposeSecret, SecretString};
 use common_config::{Configurable, DEFAULT_DATA_HOME};
 pub use common_procedure::options::ProcedureConfig;
 use common_telemetry::logging::{LoggingOptions, TracingOptions};
@@ -24,7 +27,6 @@ use file_engine::config::EngineConfig as FileEngineConfig;
 use meta_client::MetaClientOptions;
 use metric_engine::config::EngineConfig as MetricEngineConfig;
 use mito2::config::MitoConfig;
-pub(crate) use object_store::config::ObjectStoreConfig;
 use query::options::QueryOptions;
 use serde::{Deserialize, Serialize};
 use servers::export_metrics::ExportMetricsOption;
@@ -34,6 +36,53 @@ use servers::http::HttpOptions;

 pub const DEFAULT_OBJECT_STORE_CACHE_SIZE: ReadableSize = ReadableSize::gb(5);

+/// Object storage config
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[serde(tag = "type")]
+pub enum ObjectStoreConfig {
+    File(FileConfig),
+    S3(S3Config),
+    Oss(OssConfig),
+    Azblob(AzblobConfig),
+    Gcs(GcsConfig),
+}
+
+impl ObjectStoreConfig {
+    /// Returns the object storage type name, such as `S3`, `Oss` etc.
+    pub fn provider_name(&self) -> &'static str {
+        match self {
+            Self::File(_) => "File",
+            Self::S3(_) => "S3",
+            Self::Oss(_) => "Oss",
+            Self::Azblob(_) => "Azblob",
+            Self::Gcs(_) => "Gcs",
+        }
+    }
+
+    /// Returns true when it's a remote object storage such as AWS s3 etc.
+    pub fn is_object_storage(&self) -> bool {
+        !matches!(self, Self::File(_))
+    }
+
+    /// Returns the object storage configuration name, return the provider name if it's empty.
+    pub fn config_name(&self) -> &str {
+        let name = match self {
+            // file storage doesn't support name
+            Self::File(_) => self.provider_name(),
+            Self::S3(s3) => &s3.name,
+            Self::Oss(oss) => &oss.name,
+            Self::Azblob(az) => &az.name,
+            Self::Gcs(gcs) => &gcs.name,
+        };
+
+        if name.trim().is_empty() {
+            return self.provider_name();
+        }
+
+        name
+    }
+}
+
 /// Storage engine config
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 #[serde(default)]
@@ -63,6 +112,248 @@ impl Default for StorageConfig {
    }
 }

+#[derive(Debug, Clone, Serialize, Default, Deserialize, Eq, PartialEq)]
+#[serde(default)]
+pub struct FileConfig {}
+
+#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
+#[serde(default)]
+pub struct ObjectStorageCacheConfig {
+    /// The local file cache directory
+    pub cache_path: Option<String>,
+    /// The cache capacity in bytes
+    pub cache_capacity: Option<ReadableSize>,
+}
+
+/// The http client options to the storage.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[serde(default)]
+pub struct HttpClientConfig {
+    /// The maximum idle connection per host allowed in the pool.
+    pub(crate) pool_max_idle_per_host: u32,
+
+    /// The timeout for only the connect phase of a http client.
+    #[serde(with = "humantime_serde")]
+    pub(crate) connect_timeout: Duration,
+
+    /// The total request timeout, applied from when the request starts connecting until the response body has finished.
+    /// Also considered a total deadline.
+    #[serde(with = "humantime_serde")]
+    pub(crate) timeout: Duration,
+
+    /// The timeout for idle sockets being kept-alive.
+    #[serde(with = "humantime_serde")]
+    pub(crate) pool_idle_timeout: Duration,
+}
+
+impl Default for HttpClientConfig {
+    fn default() -> Self {
+        Self {
+            pool_max_idle_per_host: 1024,
+            connect_timeout: Duration::from_secs(30),
+            timeout: Duration::from_secs(30),
+            pool_idle_timeout: Duration::from_secs(90),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(default)]
+pub struct S3Config {
+    pub name: String,
+    pub bucket: String,
+    pub root: String,
+    #[serde(skip_serializing)]
+    pub access_key_id: SecretString,
+    #[serde(skip_serializing)]
+    pub secret_access_key: SecretString,
+    pub endpoint: Option<String>,
+    pub region: Option<String>,
+    /// Enable virtual host style so that opendal will send API requests in virtual host style instead of path style.
+    /// By default, opendal will send API to https://s3.us-east-1.amazonaws.com/bucket_name
+    /// Enabled, opendal will send API to https://bucket_name.s3.us-east-1.amazonaws.com
+    pub enable_virtual_host_style: bool,
+    #[serde(flatten)]
+    pub cache: ObjectStorageCacheConfig,
+    pub http_client: HttpClientConfig,
+}
+
+impl PartialEq for S3Config {
+    fn eq(&self, other: &Self) -> bool {
+        self.name == other.name
+            && self.bucket == other.bucket
+            && self.root == other.root
+            && self.access_key_id.expose_secret() == other.access_key_id.expose_secret()
+            && self.secret_access_key.expose_secret() == other.secret_access_key.expose_secret()
+            && self.endpoint == other.endpoint
+            && self.region == other.region
+            && self.enable_virtual_host_style == other.enable_virtual_host_style
+            && self.cache == other.cache
+            && self.http_client == other.http_client
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(default)]
+pub struct OssConfig {
+    pub name: String,
+    pub bucket: String,
+    pub root: String,
+    #[serde(skip_serializing)]
+    pub access_key_id: SecretString,
+    #[serde(skip_serializing)]
+    pub access_key_secret: SecretString,
+    pub endpoint: String,
+    #[serde(flatten)]
+    pub cache: ObjectStorageCacheConfig,
+    pub http_client: HttpClientConfig,
+}
+
+impl PartialEq for OssConfig {
+    fn eq(&self, other: &Self) -> bool {
+        self.name == other.name
+            && self.bucket == other.bucket
+            && self.root == other.root
+            && self.access_key_id.expose_secret() == other.access_key_id.expose_secret()
+            && self.access_key_secret.expose_secret() == other.access_key_secret.expose_secret()
+            && self.endpoint == other.endpoint
+            && self.cache == other.cache
+            && self.http_client == other.http_client
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(default)]
+pub struct AzblobConfig {
+    pub name: String,
+    pub container: String,
+    pub root: String,
+    #[serde(skip_serializing)]
+    pub account_name: SecretString,
+    #[serde(skip_serializing)]
+    pub account_key: SecretString,
+    pub endpoint: String,
+    pub sas_token: Option<String>,
+    #[serde(flatten)]
+    pub cache: ObjectStorageCacheConfig,
+    pub http_client: HttpClientConfig,
+}
+
+impl PartialEq for AzblobConfig {
+    fn eq(&self, other: &Self) -> bool {
+        self.name == other.name
+            && self.container == other.container
+            && self.root == other.root
+            && self.account_name.expose_secret() == other.account_name.expose_secret()
+            && self.account_key.expose_secret() == other.account_key.expose_secret()
+            && self.endpoint == other.endpoint
+            && self.sas_token == other.sas_token
+            && self.cache == other.cache
+            && self.http_client == other.http_client
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(default)]
+pub struct GcsConfig {
+    pub name: String,
+    pub root: String,
+    pub bucket: String,
+    pub scope: String,
+    #[serde(skip_serializing)]
+    pub credential_path: SecretString,
+    #[serde(skip_serializing)]
+    pub credential: SecretString,
+    pub endpoint: String,
+    #[serde(flatten)]
+    pub cache: ObjectStorageCacheConfig,
+    pub http_client: HttpClientConfig,
+}
+
+impl PartialEq for GcsConfig {
+    fn eq(&self, other: &Self) -> bool {
+        self.name == other.name
+            && self.root == other.root
+            && self.bucket == other.bucket
+            && self.scope == other.scope
+            && self.credential_path.expose_secret() == other.credential_path.expose_secret()
+            && self.credential.expose_secret() == other.credential.expose_secret()
+            && self.endpoint == other.endpoint
+            && self.cache == other.cache
+            && self.http_client == other.http_client
+    }
+}
+
+impl Default for S3Config {
+    fn default() -> Self {
+        Self {
+            name: String::default(),
+            bucket: String::default(),
+            root: String::default(),
+            access_key_id: SecretString::from(String::default()),
+            secret_access_key: SecretString::from(String::default()),
+            enable_virtual_host_style: false,
+            endpoint: Option::default(),
+            region: Option::default(),
+            cache: ObjectStorageCacheConfig::default(),
+            http_client: HttpClientConfig::default(),
+        }
+    }
+}
+
+impl Default for OssConfig {
+    fn default() -> Self {
+        Self {
+            name: String::default(),
+            bucket: String::default(),
+            root: String::default(),
+            access_key_id: SecretString::from(String::default()),
+            access_key_secret: SecretString::from(String::default()),
+            endpoint: String::default(),
+            cache: ObjectStorageCacheConfig::default(),
+            http_client: HttpClientConfig::default(),
+        }
+    }
+}
+
+impl Default for AzblobConfig {
+    fn default() -> Self {
+        Self {
+            name: String::default(),
+            container: String::default(),
+            root: String::default(),
+            account_name: SecretString::from(String::default()),
+            account_key: SecretString::from(String::default()),
+            endpoint: String::default(),
+            sas_token: Option::default(),
+            cache: ObjectStorageCacheConfig::default(),
+            http_client: HttpClientConfig::default(),
+        }
+    }
+}
+
+impl Default for GcsConfig {
+    fn default() -> Self {
+        Self {
+            name: String::default(),
+            root: String::default(),
+            bucket: String::default(),
+            scope: String::default(),
+            credential_path: SecretString::from(String::default()),
+            credential: SecretString::from(String::default()),
+            endpoint: String::default(),
+            cache: ObjectStorageCacheConfig::default(),
+            http_client: HttpClientConfig::default(),
+        }
+    }
+}
+
+impl Default for ObjectStoreConfig {
+    fn default() -> Self {
+        ObjectStoreConfig::File(FileConfig {})
+    }
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
 #[serde(default)]
 pub struct DatanodeOptions {
@@ -172,6 +463,37 @@ mod tests {
        let _parsed: DatanodeOptions = toml::from_str(&toml_string).unwrap();
    }

+    #[test]
+    fn test_config_name() {
+        let object_store_config = ObjectStoreConfig::default();
+        assert_eq!("File", object_store_config.config_name());
+
+        let s3_config = ObjectStoreConfig::S3(S3Config::default());
+        assert_eq!("S3", s3_config.config_name());
+        assert_eq!("S3", s3_config.provider_name());
+
+        let s3_config = ObjectStoreConfig::S3(S3Config {
+            name: "test".to_string(),
+            ..Default::default()
+        });
+        assert_eq!("test", s3_config.config_name());
+        assert_eq!("S3", s3_config.provider_name());
+    }
+
+    #[test]
+    fn test_is_object_storage() {
+        let store = ObjectStoreConfig::default();
+        assert!(!store.is_object_storage());
+        let s3_config = ObjectStoreConfig::S3(S3Config::default());
+        assert!(s3_config.is_object_storage());
+        let oss_config = ObjectStoreConfig::Oss(OssConfig::default());
+        assert!(oss_config.is_object_storage());
+        let gcs_config = ObjectStoreConfig::Gcs(GcsConfig::default());
+        assert!(gcs_config.is_object_storage());
+        let azblob_config = ObjectStoreConfig::Azblob(AzblobConfig::default());
+        assert!(azblob_config.is_object_storage());
+    }
+
    #[test]
    fn test_secstr() {
        let toml_str = r#"
@@ -192,48 +514,4 @@ mod tests {
            _ => unreachable!(),
        }
    }
-    #[test]
-    fn test_skip_ssl_validation_config() {
-        // Test with skip_ssl_validation = true
-        let toml_str_true = r#"
-            [storage]
-            type = "S3"
-            [storage.http_client]
-            skip_ssl_validation = true
-        "#;
-        let opts: DatanodeOptions = toml::from_str(toml_str_true).unwrap();
-        match &opts.storage.store {
-            ObjectStoreConfig::S3(cfg) => {
-                assert!(cfg.http_client.skip_ssl_validation);
-            }
-            _ => panic!("Expected S3 config"),
-        }
-
-        // Test with skip_ssl_validation = false
-        let toml_str_false = r#"
-            [storage]
-            type = "S3"
-            [storage.http_client]
-            skip_ssl_validation = false
-        "#;
-        let opts: DatanodeOptions = toml::from_str(toml_str_false).unwrap();
-        match &opts.storage.store {
-            ObjectStoreConfig::S3(cfg) => {
-                assert!(!cfg.http_client.skip_ssl_validation);
-            }
-            _ => panic!("Expected S3 config"),
-        }
-        // Test default value (should be false)
-        let toml_str_default = r#"
-            [storage]
-            type = "S3"
-        "#;
-        let opts: DatanodeOptions = toml::from_str(toml_str_default).unwrap();
-        match &opts.storage.store {
-            ObjectStoreConfig::S3(cfg) => {
-                assert!(!cfg.http_client.skip_ssl_validation);
-            }
-            _ => panic!("Expected S3 config"),
-        }
-    }
 }
--- a/src/datanode/src/error.rs
+++ b/src/datanode/src/error.rs
@@ -142,6 +142,14 @@ pub enum Error {
        source: Box<log_store::error::Error>,
    },

+    #[snafu(display("Failed to init backend"))]
+    InitBackend {
+        #[snafu(source)]
+        error: object_store::Error,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
    #[snafu(display("Invalid SQL, error: {}", msg))]
    InvalidSql { msg: String },

@@ -379,29 +387,6 @@ pub enum Error {
        #[snafu(implicit)]
        location: Location,
    },
-
-    #[snafu(display("Failed to serialize json"))]
-    SerializeJson {
-        #[snafu(source)]
-        error: serde_json::Error,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
-    #[snafu(display("Failed object store operation"))]
-    ObjectStore {
-        source: object_store::error::Error,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
-    #[snafu(display("Failed to build cache store"))]
-    BuildCacheStore {
-        #[snafu(source)]
-        error: object_store::Error,
-        #[snafu(implicit)]
-        location: Location,
-    },
 }

 pub type Result<T> = std::result::Result<T, Error>;
@@ -454,6 +439,8 @@ impl ErrorExt for Error {

            StartServer { source, .. } | ShutdownServer { source, .. } => source.status_code(),

+            InitBackend { .. } => StatusCode::StorageUnavailable,
+
            OpenLogStore { source, .. } => source.status_code(),
            MetaClientInit { source, .. } => source.status_code(),
            UnsupportedOutput { .. } => StatusCode::Unsupported,
@@ -470,10 +457,6 @@ impl ErrorExt for Error {
                StatusCode::RegionBusy
            }
            MissingCache { .. } => StatusCode::Internal,
-            SerializeJson { .. } => StatusCode::Internal,
-
-            ObjectStore { source, .. } => source.status_code(),
-            BuildCacheStore { .. } => StatusCode::StorageUnavailable,
        }
    }

--- a/src/datanode/src/region_server.rs
+++ b/src/datanode/src/region_server.rs
@@ -20,14 +20,12 @@ use std::time::Duration;

 use api::region::RegionResponse;
 use api::v1::region::sync_request::ManifestInfo;
-use api::v1::region::{
-    region_request, ListMetadataRequest, RegionResponse as RegionResponseV1, SyncRequest,
-};
+use api::v1::region::{region_request, RegionResponse as RegionResponseV1, SyncRequest};
 use api::v1::{ResponseHeader, Status};
 use arrow_flight::{FlightData, Ticket};
 use async_trait::async_trait;
 use bytes::Bytes;
-use common_error::ext::{BoxedError, ErrorExt};
+use common_error::ext::BoxedError;
 use common_error::status_code::StatusCode;
 use common_query::request::QueryRequest;
 use common_query::OutputData;
@@ -49,7 +47,6 @@ pub use query::dummy_catalog::{
    DummyCatalogList, DummyTableProviderFactory, TableProviderFactoryRef,
 };
 use query::QueryEngineRef;
-use serde_json;
 use servers::error::{self as servers_error, ExecuteGrpcRequestSnafu, Result as ServerResult};
 use servers::grpc::flight::{FlightCraft, FlightRecordBatchStream, TonicStream};
 use servers::grpc::region_server::RegionServerHandler;
@@ -74,10 +71,10 @@ use tonic::{Request, Response, Result as TonicResult};
 use crate::error::{
    self, BuildRegionRequestsSnafu, ConcurrentQueryLimiterClosedSnafu,
    ConcurrentQueryLimiterTimeoutSnafu, DataFusionSnafu, DecodeLogicalPlanSnafu,
-    ExecuteLogicalPlanSnafu, FindLogicalRegionsSnafu, GetRegionMetadataSnafu,
-    HandleBatchDdlRequestSnafu, HandleBatchOpenRequestSnafu, HandleRegionRequestSnafu,
-    NewPlanDecoderSnafu, RegionEngineNotFoundSnafu, RegionNotFoundSnafu, RegionNotReadySnafu,
-    Result, SerializeJsonSnafu, StopRegionEngineSnafu, UnexpectedSnafu, UnsupportedOutputSnafu,
+    ExecuteLogicalPlanSnafu, FindLogicalRegionsSnafu, HandleBatchDdlRequestSnafu,
+    HandleBatchOpenRequestSnafu, HandleRegionRequestSnafu, NewPlanDecoderSnafu,
+    RegionEngineNotFoundSnafu, RegionNotFoundSnafu, RegionNotReadySnafu, Result,
+    StopRegionEngineSnafu, UnexpectedSnafu, UnsupportedOutputSnafu,
 };
 use crate::event_listener::RegionServerEventListenerRef;

@@ -141,12 +138,12 @@ impl RegionServer {

    /// Finds the region's engine by its id. If the region is not ready, returns `None`.
    pub fn find_engine(&self, region_id: RegionId) -> Result<Option<RegionEngineRef>> {
-        match self.inner.get_engine(region_id, &RegionChange::None) {
-            Ok(CurrentEngine::Engine(engine)) => Ok(Some(engine)),
-            Ok(CurrentEngine::EarlyReturn(_)) => Ok(None),
-            Err(error::Error::RegionNotFound { .. }) => Ok(None),
-            Err(err) => Err(err),
-        }
+        self.inner
+            .get_engine(region_id, &RegionChange::None)
+            .map(|x| match x {
+                CurrentEngine::Engine(engine) => Some(engine),
+                CurrentEngine::EarlyReturn(_) => None,
+            })
    }

    #[tracing::instrument(skip_all)]
@@ -415,7 +412,6 @@ impl RegionServer {
        Ok(RegionResponse {
            affected_rows,
            extensions,
-            metadata: Vec::new(),
        })
    }

@@ -445,7 +441,6 @@ impl RegionServer {
        Ok(RegionResponse {
            affected_rows,
            extensions,
-            metadata: Vec::new(),
        })
    }

@@ -478,48 +473,6 @@ impl RegionServer {
            .map(|_| RegionResponse::new(AffectedRows::default()))
    }

-    /// Handles the ListMetadata request and retrieves metadata for specified regions.
-    ///
-    /// Returns the results as a JSON-serialized list in the [RegionResponse]. It serializes
-    /// non-existing regions as `null`.
-    #[tracing::instrument(skip_all)]
-    async fn handle_list_metadata_request(
-        &self,
-        request: &ListMetadataRequest,
-    ) -> Result<RegionResponse> {
-        let mut region_metadatas = Vec::new();
-        // Collect metadata for each region
-        for region_id in &request.region_ids {
-            let region_id = RegionId::from_u64(*region_id);
-            // Get the engine.
-            let Some(engine) = self.find_engine(region_id)? else {
-                region_metadatas.push(None);
-                continue;
-            };
-
-            match engine.get_metadata(region_id).await {
-                Ok(metadata) => region_metadatas.push(Some(metadata)),
-                Err(err) => {
-                    if err.status_code() == StatusCode::RegionNotFound {
-                        region_metadatas.push(None);
-                    } else {
-                        Err(err).with_context(|_| GetRegionMetadataSnafu {
-                            engine: engine.name(),
-                            region_id,
-                        })?;
-                    }
-                }
-            }
-        }
-
-        // Serialize metadata to JSON
-        let json_result = serde_json::to_vec(&region_metadatas).context(SerializeJsonSnafu)?;
-
-        let response = RegionResponse::from_metadata(json_result);
-
-        Ok(response)
-    }
-
    /// Sync region manifest and registers new opened logical regions.
    pub async fn sync_region(
        &self,
@@ -551,10 +504,6 @@ impl RegionServerHandler for RegionServer {
            region_request::Body::Sync(sync_request) => {
                self.handle_sync_region_request(sync_request).await
            }
-            region_request::Body::ListMetadata(list_metadata_request) => {
-                self.handle_list_metadata_request(list_metadata_request)
-                    .await
-            }
            _ => self.handle_requests_in_serial(request).await,
        }
        .map_err(BoxedError::new)
@@ -569,7 +518,6 @@ impl RegionServerHandler for RegionServer {
            }),
            affected_rows: response.affected_rows as _,
            extensions: response.extensions,
-            metadata: response.metadata,
        })
    }
 }
@@ -949,7 +897,6 @@ impl RegionServerInner {
                Ok(RegionResponse {
                    affected_rows: result.affected_rows,
                    extensions: result.extensions,
-                    metadata: Vec::new(),
                })
            }
            Err(err) => {
@@ -1020,7 +967,6 @@ impl RegionServerInner {
                Ok(RegionResponse {
                    affected_rows: result.affected_rows,
                    extensions: result.extensions,
-                    metadata: Vec::new(),
                })
            }
            Err(err) => {
@@ -1296,11 +1242,8 @@ mod tests {

    use std::assert_matches::assert_matches;

-    use api::v1::SemanticType;
    use common_error::ext::ErrorExt;
-    use datatypes::prelude::ConcreteDataType;
    use mito2::test_util::CreateRequestBuilder;
-    use store_api::metadata::{ColumnMetadata, RegionMetadata, RegionMetadataBuilder};
    use store_api::region_engine::RegionEngine;
    use store_api::region_request::{RegionDropRequest, RegionOpenRequest, RegionTruncateRequest};
    use store_api::storage::RegionId;
@@ -1662,175 +1605,4 @@ mod tests {
        let forth_query = p.acquire().await;
        assert!(forth_query.is_ok());
    }
-
-    fn mock_region_metadata(region_id: RegionId) -> RegionMetadata {
-        let mut metadata_builder = RegionMetadataBuilder::new(region_id);
-        metadata_builder.push_column_metadata(ColumnMetadata {
-            column_schema: datatypes::schema::ColumnSchema::new(
-                "timestamp",
-                ConcreteDataType::timestamp_nanosecond_datatype(),
-                false,
-            ),
-            semantic_type: SemanticType::Timestamp,
-            column_id: 0,
-        });
-        metadata_builder.push_column_metadata(ColumnMetadata {
-            column_schema: datatypes::schema::ColumnSchema::new(
-                "file",
-                ConcreteDataType::string_datatype(),
-                true,
-            ),
-            semantic_type: SemanticType::Tag,
-            column_id: 1,
-        });
-        metadata_builder.push_column_metadata(ColumnMetadata {
-            column_schema: datatypes::schema::ColumnSchema::new(
-                "message",
-                ConcreteDataType::string_datatype(),
-                true,
-            ),
-            semantic_type: SemanticType::Field,
-            column_id: 2,
-        });
-        metadata_builder.primary_key(vec![1]);
-        metadata_builder.build().unwrap()
-    }
-
-    #[tokio::test]
-    async fn test_handle_list_metadata_request() {
-        common_telemetry::init_default_ut_logging();
-
-        let mut mock_region_server = mock_region_server();
-        let region_id_1 = RegionId::new(1, 0);
-        let region_id_2 = RegionId::new(2, 0);
-
-        let metadata_1 = mock_region_metadata(region_id_1);
-        let metadata_2 = mock_region_metadata(region_id_2);
-        let metadatas = vec![Some(metadata_1.clone()), Some(metadata_2.clone())];
-
-        let metadata_1 = Arc::new(metadata_1);
-        let metadata_2 = Arc::new(metadata_2);
-        let (engine, _) = MockRegionEngine::with_metadata_mock_fn(
-            MITO_ENGINE_NAME,
-            Box::new(move |region_id| {
-                if region_id == region_id_1 {
-                    Ok(metadata_1.clone())
-                } else if region_id == region_id_2 {
-                    Ok(metadata_2.clone())
-                } else {
-                    error::RegionNotFoundSnafu { region_id }.fail()
-                }
-            }),
-        );
-
-        mock_region_server.register_engine(engine.clone());
-        mock_region_server
-            .inner
-            .region_map
-            .insert(region_id_1, RegionEngineWithStatus::Ready(engine.clone()));
-        mock_region_server
-            .inner
-            .region_map
-            .insert(region_id_2, RegionEngineWithStatus::Ready(engine.clone()));
-
-        // All regions exist.
-        let list_metadata_request = ListMetadataRequest {
-            region_ids: vec![region_id_1.as_u64(), region_id_2.as_u64()],
-        };
-        let response = mock_region_server
-            .handle_list_metadata_request(&list_metadata_request)
-            .await
-            .unwrap();
-        let decoded_metadata: Vec<Option<RegionMetadata>> =
-            serde_json::from_slice(&response.metadata).unwrap();
-        assert_eq!(metadatas, decoded_metadata);
-    }
-
-    #[tokio::test]
-    async fn test_handle_list_metadata_not_found() {
-        common_telemetry::init_default_ut_logging();
-
-        let mut mock_region_server = mock_region_server();
-        let region_id_1 = RegionId::new(1, 0);
-        let region_id_2 = RegionId::new(2, 0);
-
-        let metadata_1 = mock_region_metadata(region_id_1);
-        let metadatas = vec![Some(metadata_1.clone()), None];
-
-        let metadata_1 = Arc::new(metadata_1);
-        let (engine, _) = MockRegionEngine::with_metadata_mock_fn(
-            MITO_ENGINE_NAME,
-            Box::new(move |region_id| {
-                if region_id == region_id_1 {
-                    Ok(metadata_1.clone())
-                } else {
-                    error::RegionNotFoundSnafu { region_id }.fail()
-                }
-            }),
-        );
-
-        mock_region_server.register_engine(engine.clone());
-        mock_region_server
-            .inner
-            .region_map
-            .insert(region_id_1, RegionEngineWithStatus::Ready(engine.clone()));
-
-        // Not in region map.
-        let list_metadata_request = ListMetadataRequest {
-            region_ids: vec![region_id_1.as_u64(), region_id_2.as_u64()],
-        };
-        let response = mock_region_server
-            .handle_list_metadata_request(&list_metadata_request)
-            .await
-            .unwrap();
-        let decoded_metadata: Vec<Option<RegionMetadata>> =
-            serde_json::from_slice(&response.metadata).unwrap();
-        assert_eq!(metadatas, decoded_metadata);
-
-        // Not in region engine.
-        mock_region_server
-            .inner
-            .region_map
-            .insert(region_id_2, RegionEngineWithStatus::Ready(engine.clone()));
-        let response = mock_region_server
-            .handle_list_metadata_request(&list_metadata_request)
-            .await
-            .unwrap();
-        let decoded_metadata: Vec<Option<RegionMetadata>> =
-            serde_json::from_slice(&response.metadata).unwrap();
-        assert_eq!(metadatas, decoded_metadata);
-    }
-
-    #[tokio::test]
-    async fn test_handle_list_metadata_failed() {
-        common_telemetry::init_default_ut_logging();
-
-        let mut mock_region_server = mock_region_server();
-        let region_id_1 = RegionId::new(1, 0);
-
-        let (engine, _) = MockRegionEngine::with_metadata_mock_fn(
-            MITO_ENGINE_NAME,
-            Box::new(move |region_id| {
-                error::UnexpectedSnafu {
-                    violated: format!("Failed to get region {region_id}"),
-                }
-                .fail()
-            }),
-        );
-
-        mock_region_server.register_engine(engine.clone());
-        mock_region_server
-            .inner
-            .region_map
-            .insert(region_id_1, RegionEngineWithStatus::Ready(engine.clone()));
-
-        // Failed to get.
-        let list_metadata_request = ListMetadataRequest {
-            region_ids: vec![region_id_1.as_u64()],
-        };
-        mock_region_server
-            .handle_list_metadata_request(&list_metadata_request)
-            .await
-            .unwrap_err();
-    }
 }
--- a/src/datanode/src/service.rs
+++ b/src/datanode/src/service.rs
@@ -17,7 +17,7 @@ use std::sync::Arc;

 use common_config::Configurable;
 use servers::grpc::builder::GrpcServerBuilder;
-use servers::grpc::GrpcServer;
+use servers::grpc::{GrpcServer, GrpcServerConfig};
 use servers::http::HttpServerBuilder;
 use servers::metrics_handler::MetricsHandler;
 use servers::server::{ServerHandler, ServerHandlers};
@@ -92,7 +92,13 @@ impl<'a> DatanodeServiceBuilder<'a> {
        opts: &DatanodeOptions,
        region_server: &RegionServer,
    ) -> GrpcServerBuilder {
-        GrpcServerBuilder::new(opts.grpc.as_config(), region_server.runtime())
+        let config = GrpcServerConfig {
+            max_recv_message_size: opts.grpc.max_recv_message_size.as_bytes() as usize,
+            max_send_message_size: opts.grpc.max_send_message_size.as_bytes() as usize,
+            tls: opts.grpc.tls.clone(),
+        };
+
+        GrpcServerBuilder::new(config, region_server.runtime())
            .flight_handler(Arc::new(region_server.clone()))
            .region_server_handler(Arc::new(region_server.clone()))
    }
--- a/src/datanode/src/store.rs
+++ b/src/datanode/src/store.rs
@@ -14,22 +14,45 @@

 //! object storage utilities

+mod azblob;
+pub mod fs;
+mod gcs;
+mod oss;
+mod s3;
+use std::path;
 use std::path::Path;
 use std::sync::Arc;
 use std::time::Duration;

 use common_telemetry::{info, warn};
-use object_store::factory::new_raw_object_store;
+use mito2::access_layer::{ATOMIC_WRITE_DIR, OLD_ATOMIC_WRITE_DIR};
 use object_store::layers::{LruCacheLayer, RetryInterceptor, RetryLayer};
 use object_store::services::Fs;
-use object_store::util::{clean_temp_dir, join_dir, with_instrument_layers};
-use object_store::{
-    Access, Error, ObjectStore, ObjectStoreBuilder, ATOMIC_WRITE_DIR, OLD_ATOMIC_WRITE_DIR,
-};
+use object_store::util::{join_dir, normalize_dir, with_instrument_layers};
+use object_store::{Access, Error, HttpClient, ObjectStore, ObjectStoreBuilder};
 use snafu::prelude::*;

-use crate::config::{ObjectStoreConfig, DEFAULT_OBJECT_STORE_CACHE_SIZE};
-use crate::error::{self, CreateDirSnafu, Result};
+use crate::config::{HttpClientConfig, ObjectStoreConfig, DEFAULT_OBJECT_STORE_CACHE_SIZE};
+use crate::error::{self, BuildHttpClientSnafu, CreateDirSnafu, Result};
+
+pub(crate) async fn new_raw_object_store(
+    store: &ObjectStoreConfig,
+    data_home: &str,
+) -> Result<ObjectStore> {
+    let data_home = normalize_dir(data_home);
+    let object_store = match store {
+        ObjectStoreConfig::File(file_config) => {
+            fs::new_fs_object_store(&data_home, file_config).await
+        }
+        ObjectStoreConfig::S3(s3_config) => s3::new_s3_object_store(s3_config).await,
+        ObjectStoreConfig::Oss(oss_config) => oss::new_oss_object_store(oss_config).await,
+        ObjectStoreConfig::Azblob(azblob_config) => {
+            azblob::new_azblob_object_store(azblob_config).await
+        }
+        ObjectStoreConfig::Gcs(gcs_config) => gcs::new_gcs_object_store(gcs_config).await,
+    }?;
+    Ok(object_store)
+}

 fn with_retry_layers(object_store: ObjectStore) -> ObjectStore {
    object_store.layer(
@@ -43,9 +66,7 @@ pub(crate) async fn new_object_store_without_cache(
    store: &ObjectStoreConfig,
    data_home: &str,
 ) -> Result<ObjectStore> {
-    let object_store = new_raw_object_store(store, data_home)
-        .await
-        .context(error::ObjectStoreSnafu)?;
+    let object_store = new_raw_object_store(store, data_home).await?;
    // Enable retry layer and cache layer for non-fs object storages
    let object_store = if store.is_object_storage() {
        // Adds retry layer
@@ -62,9 +83,7 @@ pub(crate) async fn new_object_store(
    store: ObjectStoreConfig,
    data_home: &str,
 ) -> Result<ObjectStore> {
-    let object_store = new_raw_object_store(&store, data_home)
-        .await
-        .context(error::ObjectStoreSnafu)?;
+    let object_store = new_raw_object_store(&store, data_home).await?;
    // Enable retry layer and cache layer for non-fs object storages
    let object_store = if store.is_object_storage() {
        let object_store = if let Some(cache_layer) = build_cache_layer(&store, data_home).await? {
@@ -151,20 +170,20 @@ async fn build_cache_layer(
        && !path.trim().is_empty()
    {
        let atomic_temp_dir = join_dir(path, ATOMIC_WRITE_DIR);
-        clean_temp_dir(&atomic_temp_dir).context(error::ObjectStoreSnafu)?;
+        clean_temp_dir(&atomic_temp_dir)?;

        // Compatible code. Remove this after a major release.
        let old_atomic_temp_dir = join_dir(path, OLD_ATOMIC_WRITE_DIR);
-        clean_temp_dir(&old_atomic_temp_dir).context(error::ObjectStoreSnafu)?;
+        clean_temp_dir(&old_atomic_temp_dir)?;

        let cache_store = Fs::default()
            .root(path)
            .atomic_write_dir(&atomic_temp_dir)
            .build()
-            .context(error::BuildCacheStoreSnafu)?;
+            .context(error::InitBackendSnafu)?;

        let cache_layer = LruCacheLayer::new(Arc::new(cache_store), cache_capacity.0 as usize)
-            .context(error::BuildCacheStoreSnafu)?;
+            .context(error::InitBackendSnafu)?;
        cache_layer.recover_cache(false).await;
        info!(
            "Enabled local object storage cache, path: {}, capacity: {}.",
@@ -177,6 +196,26 @@ async fn build_cache_layer(
    }
 }

+pub(crate) fn clean_temp_dir(dir: &str) -> Result<()> {
+    if path::Path::new(&dir).exists() {
+        info!("Begin to clean temp storage directory: {}", dir);
+        std::fs::remove_dir_all(dir).context(error::RemoveDirSnafu { dir })?;
+        info!("Cleaned temp storage directory: {}", dir);
+    }
+
+    Ok(())
+}
+
+pub(crate) fn build_http_client(config: &HttpClientConfig) -> Result<HttpClient> {
+    let client = reqwest::ClientBuilder::new()
+        .pool_max_idle_per_host(config.pool_max_idle_per_host as usize)
+        .connect_timeout(config.connect_timeout)
+        .pool_idle_timeout(config.pool_idle_timeout)
+        .timeout(config.timeout)
+        .build()
+        .context(BuildHttpClientSnafu)?;
+    Ok(HttpClient::with(client))
+}
 struct PrintDetailedError;

 // PrintDetailedError is a retry interceptor that prints error in Debug format in retrying.
--- a/src/datanode/src/store/azblob.rs
+++ b/src/datanode/src/store/azblob.rs
@@ -0,0 +1,50 @@
+// 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_base::secrets::ExposeSecret;
+use common_telemetry::info;
+use object_store::services::Azblob;
+use object_store::{util, ObjectStore};
+use snafu::prelude::*;
+
+use crate::config::AzblobConfig;
+use crate::error::{self, Result};
+use crate::store::build_http_client;
+
+pub(crate) async fn new_azblob_object_store(azblob_config: &AzblobConfig) -> Result<ObjectStore> {
+    let root = util::normalize_dir(&azblob_config.root);
+
+    info!(
+        "The azure storage container is: {}, root is: {}",
+        azblob_config.container, &root
+    );
+
+    let client = build_http_client(&azblob_config.http_client)?;
+
+    let mut builder = Azblob::default()
+        .root(&root)
+        .container(&azblob_config.container)
+        .endpoint(&azblob_config.endpoint)
+        .account_name(azblob_config.account_name.expose_secret())
+        .account_key(azblob_config.account_key.expose_secret())
+        .http_client(client);
+
+    if let Some(token) = &azblob_config.sas_token {
+        builder = builder.sas_token(token);
+    };
+
+    Ok(ObjectStore::new(builder)
+        .context(error::InitBackendSnafu)?
+        .finish())
+}
--- a/src/datanode/src/store/fs.rs
+++ b/src/datanode/src/store/fs.rs
@@ -0,0 +1,53 @@
+// 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::{fs, path};
+
+use common_telemetry::info;
+use mito2::access_layer::{ATOMIC_WRITE_DIR, OLD_ATOMIC_WRITE_DIR};
+use object_store::services::Fs;
+use object_store::util::join_dir;
+use object_store::ObjectStore;
+use snafu::prelude::*;
+
+use crate::config::FileConfig;
+use crate::error::{self, Result};
+use crate::store;
+
+/// A helper function to create a file system object store.
+pub async fn new_fs_object_store(
+    data_home: &str,
+    _file_config: &FileConfig,
+) -> Result<ObjectStore> {
+    fs::create_dir_all(path::Path::new(&data_home))
+        .context(error::CreateDirSnafu { dir: data_home })?;
+    info!("The file storage home is: {}", data_home);
+
+    let atomic_write_dir = join_dir(data_home, ATOMIC_WRITE_DIR);
+    store::clean_temp_dir(&atomic_write_dir)?;
+
+    // Compatible code. Remove this after a major release.
+    let old_atomic_temp_dir = join_dir(data_home, OLD_ATOMIC_WRITE_DIR);
+    store::clean_temp_dir(&old_atomic_temp_dir)?;
+
+    let builder = Fs::default()
+        .root(data_home)
+        .atomic_write_dir(&atomic_write_dir);
+
+    let object_store = ObjectStore::new(builder)
+        .context(error::InitBackendSnafu)?
+        .finish();
+
+    Ok(object_store)
+}
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
discord9	b4b1c335ac	chore: comment Signed-off-by: discord9 <discord9@163.com>	2025-06-12 17:44:54 +08:00
discord9	41e97a0657	chore: per review Signed-off-by: discord9 <discord9@163.com>	2025-06-12 17:44:54 +08:00
discord9	0eca6333e5	feat: parallelism hint in grpc Signed-off-by: discord9 <discord9@163.com>	2025-06-12 17:44:54 +08:00