chore: rm unwrap

Signed-off-by: discord9 <discord9@163.com>
docs: explain fast path
2025-12-26 08:00:01 +00:00 · 2025-07-10 10:55:36 +08:00 · 2025-07-10 10:48:38 +08:00 · 2025-07-10 00:00:20 +08:00 · 2025-07-09 13:23:15 +00:00 · 2025-07-09 11:46:07 +00:00
454 changed files with 9005 additions and 21871 deletions
--- a/.github/actions/setup-etcd-cluster/action.yml
+++ b/.github/actions/setup-etcd-cluster/action.yml
@@ -12,7 +12,7 @@ runs:
  steps:
  - name: Install Etcd cluster
    shell: bash
-    run: |
+    run: | 
      helm upgrade \
        --install etcd oci://registry-1.docker.io/bitnamicharts/etcd \
        --set replicaCount=${{ inputs.etcd-replicas }} \
@@ -24,9 +24,4 @@ runs:
        --set auth.rbac.token.enabled=false \
        --set persistence.size=2Gi \
        --create-namespace \
-        --set global.security.allowInsecureImages=true \
-        --set image.registry=docker.io \
-        --set image.repository=greptime/etcd \
-        --set image.tag=3.6.1-debian-12-r3 \
-        --version 12.0.8 \
        -n ${{ inputs.namespace }}
--- a/.github/actions/setup-kafka-cluster/action.yml
+++ b/.github/actions/setup-kafka-cluster/action.yml
@@ -12,7 +12,7 @@ runs:
  steps:
  - name: Install Kafka cluster
    shell: bash
-    run: |
+    run: | 
      helm upgrade \
        --install kafka oci://registry-1.docker.io/bitnamicharts/kafka \
        --set controller.replicaCount=${{ inputs.controller-replicas }} \
@@ -23,8 +23,4 @@ runs:
        --set listeners.controller.protocol=PLAINTEXT \
        --set listeners.client.protocol=PLAINTEXT \
        --create-namespace \
-        --set image.registry=docker.io \
-        --set image.repository=greptime/kafka \
-        --set image.tag=3.9.0-debian-12-r1 \
-        --version 31.0.0 \
        -n ${{ inputs.namespace }}
--- a/.github/actions/setup-postgres-cluster/action.yml
+++ b/.github/actions/setup-postgres-cluster/action.yml
@@ -6,7 +6,9 @@ inputs:
    description: "Number of PostgreSQL replicas"
  namespace:
    default: "postgres-namespace"
-    description: "The PostgreSQL namespace"
+  postgres-version:
+    default: "14.2"
+    description: "PostgreSQL version"
  storage-size:
    default: "1Gi"
    description: "Storage size for PostgreSQL"
@@ -20,11 +22,7 @@ runs:
      helm upgrade \
        --install postgresql oci://registry-1.docker.io/bitnamicharts/postgresql \
        --set replicaCount=${{ inputs.postgres-replicas }} \
-        --set global.security.allowInsecureImages=true \
-        --set image.registry=docker.io \
-        --set image.repository=greptime/postgresql \
-        --set image.tag=17.5.0-debian-12-r3 \
-        --version 16.7.4 \
+        --set image.tag=${{ inputs.postgres-version }} \
        --set persistence.size=${{ inputs.storage-size }} \
        --set postgresql.username=greptimedb \
        --set postgresql.password=admin \
--- a/.github/scripts/pull-test-deps-images.sh
+++ b/.github/scripts/pull-test-deps-images.sh
@@ -1,34 +0,0 @@
-#!/bin/bash
-
-# This script is used to pull the test dependency images that are stored in public ECR one by one to avoid rate limiting.
-
-set -e
-
-MAX_RETRIES=3
-
-IMAGES=(
-  "greptime/zookeeper:3.7"
-  "greptime/kafka:3.9.0-debian-12-r1"
-  "greptime/etcd:3.6.1-debian-12-r3"
-  "greptime/minio:2024"
-  "greptime/mysql:5.7"
-)
-
-for image in "${IMAGES[@]}"; do
-  for ((attempt=1; attempt<=MAX_RETRIES; attempt++)); do
-    if docker pull "$image"; then
-      # Successfully pulled the image.
-      break
-    else
-      # Use some simple exponential backoff to avoid rate limiting.
-      if [ $attempt -lt $MAX_RETRIES ]; then
-        sleep_seconds=$((attempt * 5))
-        echo "Attempt $attempt failed for $image, waiting $sleep_seconds seconds"
-        sleep $sleep_seconds  # 5s, 10s delays
-      else
-        echo "Failed to pull $image after $MAX_RETRIES attempts"
-        exit 1
-      fi
-    fi
-  done
-done
--- a/.github/workflows/develop.yml
+++ b/.github/workflows/develop.yml
@@ -719,10 +719,6 @@ jobs:
          save-if: ${{ github.ref == 'refs/heads/main' }}
      - name: Install latest nextest release
        uses: taiki-e/install-action@nextest
-
-      - name: Pull test dependencies images
-        run: ./.github/scripts/pull-test-deps-images.sh
-
      - name: Setup external services
        working-directory: tests-integration/fixtures
        run: docker compose up -d --wait
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -339,7 +339,7 @@ jobs:
      build-windows-artifacts,
      release-images-to-dockerhub,
    ]
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-latest-16-cores
    # When we push to ACR, it's easy to fail due to some unknown network issues.
    # However, we don't want to fail the whole workflow because of this.
    # The ACR have daily sync with DockerHub, so don't worry about the image not being updated.
--- a/.github/workflows/semantic-pull-request.yml
+++ b/.github/workflows/semantic-pull-request.yml
@@ -11,17 +11,17 @@ concurrency:
  group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
  cancel-in-progress: true

+permissions:
+  issues: write
+  contents: write
+  pull-requests: write
+
 jobs:
  check:
    runs-on: ubuntu-latest
-    permissions:
-      pull-requests: write  # Add permissions to modify PRs
-      issues: write
    timeout-minutes: 10
    steps:
      - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
      - uses: ./.github/actions/setup-cyborg
      - name: Check Pull Request
        working-directory: cyborg
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -211,7 +211,7 @@ checksum = "d301b3b94cb4b2f23d7917810addbbaff90738e0ca2be692bd027e70d7e0330c"

 [[package]]
 name = "api"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "common-base",
 "common-decimal",
@@ -944,7 +944,7 @@ dependencies = [

 [[package]]
 name = "auth"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "async-trait",
@@ -1586,7 +1586,7 @@ dependencies = [

 [[package]]
 name = "cache"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "catalog",
 "common-error",
@@ -1621,7 +1621,7 @@ checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5"

 [[package]]
 name = "catalog"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "arrow 54.2.1",
@@ -1659,8 +1659,6 @@ dependencies = [
 "partition",
 "paste",
 "prometheus",
- "promql-parser",
- "rand 0.9.0",
 "rustc-hash 2.0.0",
 "serde_json",
 "session",
@@ -1961,7 +1959,7 @@ checksum = "1462739cb27611015575c0c11df5df7601141071f07518d56fcc1be504cbec97"

 [[package]]
 name = "cli"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "async-stream",
 "async-trait",
@@ -1998,14 +1996,14 @@ dependencies = [
 "operator",
 "query",
 "rand 0.9.0",
- "reqwest",
+ "reqwest 0.12.9",
 "serde",
 "serde_json",
 "servers",
 "session",
 "snafu 0.8.5",
 "store-api",
- "substrait 0.15.5",
+ "substrait 0.16.0",
 "table",
 "tempfile",
 "tokio",
@@ -2014,7 +2012,7 @@ dependencies = [

 [[package]]
 name = "client"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "arc-swap",
@@ -2044,7 +2042,7 @@ dependencies = [
 "rand 0.9.0",
 "serde_json",
 "snafu 0.8.5",
- "substrait 0.15.5",
+ "substrait 0.16.0",
 "substrait 0.37.3",
 "tokio",
 "tokio-stream",
@@ -2085,7 +2083,7 @@ dependencies = [

 [[package]]
 name = "cmd"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "async-trait",
 "auth",
@@ -2129,13 +2127,14 @@ dependencies = [
 "mito2",
 "moka",
 "nu-ansi-term",
+ "object-store",
 "plugins",
 "prometheus",
 "prost 0.13.5",
 "query",
 "rand 0.9.0",
 "regex",
- "reqwest",
+ "reqwest 0.12.9",
 "rexpect",
 "serde",
 "serde_json",
@@ -2145,7 +2144,7 @@ dependencies = [
 "snafu 0.8.5",
 "stat",
 "store-api",
- "substrait 0.15.5",
+ "substrait 0.16.0",
 "table",
 "temp-env",
 "tempfile",
@@ -2192,7 +2191,7 @@ checksum = "55b672471b4e9f9e95499ea597ff64941a309b2cdbffcc46f2cc5e2d971fd335"

 [[package]]
 name = "common-base"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "anymap2",
 "async-trait",
@@ -2214,11 +2213,11 @@ dependencies = [

 [[package]]
 name = "common-catalog"
-version = "0.15.5"
+version = "0.16.0"

 [[package]]
 name = "common-config"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "common-base",
 "common-error",
@@ -2231,6 +2230,7 @@ dependencies = [
 "humantime-serde",
 "meta-client",
 "num_cpus",
+ "object-store",
 "serde",
 "serde_json",
 "serde_with",
@@ -2243,7 +2243,7 @@ dependencies = [

 [[package]]
 name = "common-datasource"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "arrow 54.2.1",
 "arrow-schema 54.3.1",
@@ -2280,7 +2280,7 @@ dependencies = [

 [[package]]
 name = "common-decimal"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "bigdecimal 0.4.8",
 "common-error",
@@ -2293,7 +2293,7 @@ dependencies = [

 [[package]]
 name = "common-error"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "common-macro",
 "http 1.1.0",
@@ -2304,7 +2304,7 @@ dependencies = [

 [[package]]
 name = "common-frontend"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "async-trait",
 "common-error",
@@ -2313,7 +2313,6 @@ dependencies = [
 "common-meta",
 "greptime-proto",
 "meta-client",
- "session",
 "snafu 0.8.5",
 "tokio",
 "tonic 0.12.3",
@@ -2321,7 +2320,7 @@ dependencies = [

 [[package]]
 name = "common-function"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -2344,6 +2343,7 @@ dependencies = [
 "datafusion",
 "datafusion-common",
 "datafusion-expr",
+ "datafusion-functions-aggregate-common",
 "datatypes",
 "derive_more",
 "geo",
@@ -2374,7 +2374,7 @@ dependencies = [

 [[package]]
 name = "common-greptimedb-telemetry"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "async-trait",
 "common-runtime",
@@ -2382,7 +2382,7 @@ dependencies = [
 "common-test-util",
 "common-version",
 "hyper 0.14.30",
- "reqwest",
+ "reqwest 0.12.9",
 "serde",
 "tempfile",
 "tokio",
@@ -2391,7 +2391,7 @@ dependencies = [

 [[package]]
 name = "common-grpc"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "arrow-flight",
@@ -2423,7 +2423,7 @@ dependencies = [

 [[package]]
 name = "common-grpc-expr"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "common-base",
@@ -2442,7 +2442,7 @@ dependencies = [

 [[package]]
 name = "common-macro"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "arc-swap",
 "common-query",
@@ -2456,7 +2456,7 @@ dependencies = [

 [[package]]
 name = "common-mem-prof"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "anyhow",
 "common-error",
@@ -2472,7 +2472,7 @@ dependencies = [

 [[package]]
 name = "common-meta"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "anymap2",
 "api",
@@ -2492,7 +2492,6 @@ dependencies = [
 "common-procedure-test",
 "common-query",
 "common-recordbatch",
- "common-runtime",
 "common-telemetry",
 "common-test-util",
 "common-time",
@@ -2538,7 +2537,7 @@ dependencies = [

 [[package]]
 name = "common-options"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "common-grpc",
 "humantime-serde",
@@ -2547,11 +2546,11 @@ dependencies = [

 [[package]]
 name = "common-plugins"
-version = "0.15.5"
+version = "0.16.0"

 [[package]]
 name = "common-pprof"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "common-error",
 "common-macro",
@@ -2563,7 +2562,7 @@ dependencies = [

 [[package]]
 name = "common-procedure"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "async-stream",
 "async-trait",
@@ -2590,17 +2589,16 @@ dependencies = [

 [[package]]
 name = "common-procedure-test"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "async-trait",
 "common-procedure",
 "snafu 0.8.5",
- "tokio",
 ]

 [[package]]
 name = "common-query"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "async-trait",
@@ -2626,7 +2624,7 @@ dependencies = [

 [[package]]
 name = "common-recordbatch"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "arc-swap",
 "common-error",
@@ -2646,7 +2644,7 @@ dependencies = [

 [[package]]
 name = "common-runtime"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "async-trait",
 "clap 4.5.19",
@@ -2676,14 +2674,32 @@ dependencies = [

 [[package]]
 name = "common-session"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "strum 0.27.1",
 ]

+[[package]]
+name = "common-sql"
+version = "0.16.0"
+dependencies = [
+ "common-base",
+ "common-datasource",
+ "common-decimal",
+ "common-error",
+ "common-macro",
+ "common-time",
+ "datafusion-sql",
+ "datatypes",
+ "hex",
+ "jsonb",
+ "snafu 0.8.5",
+ "sqlparser 0.54.0 (git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=0cf6c04490d59435ee965edd2078e8855bd8471e)",
+]
+
 [[package]]
 name = "common-telemetry"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "backtrace",
 "common-error",
@@ -2711,7 +2727,7 @@ dependencies = [

 [[package]]
 name = "common-test-util"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "client",
 "common-grpc",
@@ -2724,7 +2740,7 @@ dependencies = [

 [[package]]
 name = "common-time"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "arrow 54.2.1",
 "chrono",
@@ -2742,7 +2758,7 @@ dependencies = [

 [[package]]
 name = "common-version"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "build-data",
 "cargo-manifest",
@@ -2753,7 +2769,7 @@ dependencies = [

 [[package]]
 name = "common-wal"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "common-base",
 "common-error",
@@ -2776,7 +2792,7 @@ dependencies = [

 [[package]]
 name = "common-workload"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "common-telemetry",
@@ -3732,7 +3748,7 @@ dependencies = [

 [[package]]
 name = "datanode"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "arrow-flight",
@@ -3778,14 +3794,14 @@ dependencies = [
 "prometheus",
 "prost 0.13.5",
 "query",
- "reqwest",
+ "reqwest 0.12.9",
 "serde",
 "serde_json",
 "servers",
 "session",
 "snafu 0.8.5",
 "store-api",
- "substrait 0.15.5",
+ "substrait 0.16.0",
 "table",
 "tokio",
 "toml 0.8.19",
@@ -3794,7 +3810,7 @@ dependencies = [

 [[package]]
 name = "datatypes"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "arrow 54.2.1",
 "arrow-array 54.2.1",
@@ -4454,7 +4470,7 @@ checksum = "e8c02a5121d4ea3eb16a80748c74f5549a5665e4c21333c6098f283870fbdea6"

 [[package]]
 name = "file-engine"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "async-trait",
@@ -4591,7 +4607,7 @@ checksum = "8bf7cc16383c4b8d58b9905a8509f02926ce3058053c056376248d958c9df1e8"

 [[package]]
 name = "flow"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "arrow 54.2.1",
@@ -4656,7 +4672,7 @@ dependencies = [
 "sql",
 "store-api",
 "strum 0.27.1",
- "substrait 0.15.5",
+ "substrait 0.16.0",
 "table",
 "tokio",
 "tonic 0.12.3",
@@ -4711,7 +4727,7 @@ checksum = "6c2141d6d6c8512188a7891b4b01590a45f6dac67afb4f255c4124dbb86d4eaa"

 [[package]]
 name = "frontend"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "arc-swap",
@@ -4771,7 +4787,7 @@ dependencies = [
 "sqlparser 0.54.0 (git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=0cf6c04490d59435ee965edd2078e8855bd8471e)",
 "store-api",
 "strfmt",
- "substrait 0.15.5",
+ "substrait 0.16.0",
 "table",
 "tokio",
 "tokio-util",
@@ -5161,7 +5177,7 @@ dependencies = [
 [[package]]
 name = "greptime-proto"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=f3103a8c9b8ce162457d0a3e3ca00d53d1a8bd06#f3103a8c9b8ce162457d0a3e3ca00d53d1a8bd06"
+source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=ceb1af4fa9309ce65bda0367db7b384df2bb4d4f#ceb1af4fa9309ce65bda0367db7b384df2bb4d4f"
 dependencies = [
 "prost 0.13.5",
 "serde",
@@ -5932,7 +5948,7 @@ dependencies = [

 [[package]]
 name = "index"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "async-trait",
 "asynchronous-codec",
@@ -5944,7 +5960,6 @@ dependencies = [
 "common-runtime",
 "common-telemetry",
 "common-test-util",
- "criterion 0.4.0",
 "fastbloom",
 "fst",
 "futures",
@@ -5957,7 +5972,6 @@ dependencies = [
 "prost 0.13.5",
 "puffin",
 "rand 0.9.0",
- "rand_chacha 0.9.0",
 "regex",
 "regex-automata 0.4.8",
 "roaring",
@@ -6819,7 +6833,7 @@ checksum = "a7a70ba024b9dc04c27ea2f0c0548feb474ec5c54bba33a7f72f873a39d07b24"

 [[package]]
 name = "log-query"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "chrono",
 "common-error",
@@ -6831,7 +6845,7 @@ dependencies = [

 [[package]]
 name = "log-store"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "async-stream",
 "async-trait",
@@ -7129,7 +7143,7 @@ dependencies = [

 [[package]]
 name = "meta-client"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "async-trait",
@@ -7157,13 +7171,10 @@ dependencies = [

 [[package]]
 name = "meta-srv"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "async-trait",
- "axum 0.8.1",
- "axum-extra",
- "axum-macros",
 "bytes",
 "chrono",
 "clap 4.5.19",
@@ -7197,7 +7208,6 @@ dependencies = [
 "http-body-util",
 "humantime",
 "humantime-serde",
- "hyper 0.14.30",
 "hyper-util",
 "itertools 0.14.0",
 "lazy_static",
@@ -7225,7 +7235,6 @@ dependencies = [
 "toml 0.8.19",
 "tonic 0.12.3",
 "tower 0.5.2",
- "tower-http 0.6.2",
 "tracing",
 "tracing-subscriber",
 "typetag",
@@ -7254,7 +7263,7 @@ dependencies = [

 [[package]]
 name = "metric-engine"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "aquamarine",
@@ -7279,7 +7288,6 @@ dependencies = [
 "lazy_static",
 "mito-codec",
 "mito2",
- "moka",
 "mur3",
 "object-store",
 "prometheus",
@@ -7346,7 +7354,7 @@ dependencies = [

 [[package]]
 name = "mito-codec"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "bytes",
@@ -7369,7 +7377,7 @@ dependencies = [

 [[package]]
 name = "mito2"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "aquamarine",
@@ -8119,18 +8127,25 @@ dependencies = [

 [[package]]
 name = "object-store"
-version = "0.15.5"
+version = "0.16.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",
 ]
@@ -8265,7 +8280,7 @@ dependencies = [
 "prometheus",
 "quick-xml 0.36.2",
 "reqsign",
- "reqwest",
+ "reqwest 0.12.9",
 "serde",
 "serde_json",
 "sha2",
@@ -8337,6 +8352,19 @@ 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"
@@ -8347,10 +8375,12 @@ 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",
@@ -8433,7 +8463,7 @@ dependencies = [

 [[package]]
 name = "operator"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -8458,6 +8488,7 @@ dependencies = [
 "common-query",
 "common-recordbatch",
 "common-runtime",
+ "common-sql",
 "common-telemetry",
 "common-test-util",
 "common-time",
@@ -8488,7 +8519,7 @@ dependencies = [
 "sql",
 "sqlparser 0.54.0 (git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=0cf6c04490d59435ee965edd2078e8855bd8471e)",
 "store-api",
- "substrait 0.15.5",
+ "substrait 0.16.0",
 "table",
 "tokio",
 "tokio-util",
@@ -8755,7 +8786,7 @@ dependencies = [

 [[package]]
 name = "partition"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "async-trait",
@@ -9043,7 +9074,7 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"

 [[package]]
 name = "pipeline"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -9186,7 +9217,7 @@ dependencies = [

 [[package]]
 name = "plugins"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "auth",
 "clap 4.5.19",
@@ -9499,7 +9530,7 @@ dependencies = [

 [[package]]
 name = "promql"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "ahash 0.8.11",
 "async-trait",
@@ -9523,8 +9554,9 @@ dependencies = [

 [[package]]
 name = "promql-parser"
-version = "0.5.1"
-source = "git+https://github.com/GreptimeTeam/promql-parser.git?rev=0410e8b459dda7cb222ce9596f8bf3971bd07bd2#0410e8b459dda7cb222ce9596f8bf3971bd07bd2"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "328fe69c2443ec4f8e6c33ea925dde04a1026e6c95928e89ed02343944cac9bf"
 dependencies = [
 "cfgrammar",
 "chrono",
@@ -9781,7 +9813,7 @@ dependencies = [

 [[package]]
 name = "puffin"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "async-compression 0.4.13",
 "async-trait",
@@ -9823,7 +9855,7 @@ dependencies = [

 [[package]]
 name = "query"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -9889,7 +9921,7 @@ dependencies = [
 "sqlparser 0.54.0 (git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=0cf6c04490d59435ee965edd2078e8855bd8471e)",
 "statrs",
 "store-api",
- "substrait 0.15.5",
+ "substrait 0.16.0",
 "table",
 "tokio",
 "tokio-stream",
@@ -10337,7 +10369,7 @@ dependencies = [
 "percent-encoding",
 "quick-xml 0.35.0",
 "rand 0.8.5",
- "reqwest",
+ "reqwest 0.12.9",
 "rsa",
 "rust-ini 0.21.1",
 "serde",
@@ -10346,6 +10378,42 @@ 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"
@@ -11175,7 +11243,7 @@ dependencies = [

 [[package]]
 name = "servers"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -11208,6 +11276,7 @@ dependencies = [
 "common-recordbatch",
 "common-runtime",
 "common-session",
+ "common-sql",
 "common-telemetry",
 "common-test-util",
 "common-time",
@@ -11261,7 +11330,7 @@ dependencies = [
 "quoted-string",
 "rand 0.9.0",
 "regex",
- "reqwest",
+ "reqwest 0.12.9",
 "rust-embed",
 "rustls",
 "rustls-pemfile",
@@ -11296,7 +11365,7 @@ dependencies = [

 [[package]]
 name = "session"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "arc-swap",
@@ -11635,7 +11704,7 @@ dependencies = [

 [[package]]
 name = "sql"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "chrono",
@@ -11646,6 +11715,7 @@ dependencies = [
 "common-error",
 "common-macro",
 "common-query",
+ "common-sql",
 "common-time",
 "datafusion",
 "datafusion-common",
@@ -11690,7 +11760,7 @@ dependencies = [

 [[package]]
 name = "sqlness-runner"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "async-trait",
 "clap 4.5.19",
@@ -11705,7 +11775,7 @@ dependencies = [
 "local-ip-address",
 "mysql",
 "num_cpus",
- "reqwest",
+ "reqwest 0.12.9",
 "serde",
 "serde_json",
 "sha2",
@@ -11990,7 +12060,7 @@ dependencies = [

 [[package]]
 name = "stat"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "nix 0.30.1",
 ]
@@ -12016,7 +12086,7 @@ dependencies = [

 [[package]]
 name = "store-api"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "aquamarine",
@@ -12177,7 +12247,7 @@ dependencies = [

 [[package]]
 name = "substrait"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "async-trait",
 "bytes",
@@ -12355,9 +12425,30 @@ 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.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "async-trait",
@@ -12618,7 +12709,7 @@ checksum = "3369f5ac52d5eb6ab48c6b4ffdc8efbcad6b89c765749064ba298f2c68a16a76"

 [[package]]
 name = "tests-fuzz"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "arbitrary",
 "async-trait",
@@ -12645,7 +12736,7 @@ dependencies = [
 "paste",
 "rand 0.9.0",
 "rand_chacha 0.9.0",
- "reqwest",
+ "reqwest 0.12.9",
 "schemars",
 "serde",
 "serde_json",
@@ -12662,7 +12753,7 @@ dependencies = [

 [[package]]
 name = "tests-integration"
-version = "0.15.5"
+version = "0.16.0"
 dependencies = [
 "api",
 "arrow-flight",
@@ -12729,7 +12820,7 @@ dependencies = [
 "sql",
 "sqlx",
 "store-api",
- "substrait 0.15.5",
+ "substrait 0.16.0",
 "table",
 "tempfile",
 "time",
@@ -12739,7 +12830,6 @@ dependencies = [
 "tonic 0.12.3",
 "tower 0.5.2",
 "url",
- "urlencoding",
 "uuid",
 "yaml-rust",
 "zstd 0.13.2",
@@ -14530,6 +14620,16 @@ 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
@@ -30,6 +30,7 @@ members = [
    "src/common/recordbatch",
    "src/common/runtime",
    "src/common/session",
+    "src/common/sql",
    "src/common/stat",
    "src/common/substrait",
    "src/common/telemetry",
@@ -71,11 +72,13 @@ members = [
 resolver = "2"

 [workspace.package]
-version = "0.15.5"
+version = "0.16.0"
 edition = "2021"
 license = "Apache-2.0"

 [workspace.lints]
+clippy.print_stdout = "warn"
+clippy.print_stderr = "warn"
 clippy.dbg_macro = "warn"
 clippy.implicit_clone = "warn"
 clippy.result_large_err = "allow"
@@ -121,6 +124,7 @@ 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 +139,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 = "f3103a8c9b8ce162457d0a3e3ca00d53d1a8bd06" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "ceb1af4fa9309ce65bda0367db7b384df2bb4d4f" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"
@@ -168,9 +172,7 @@ parquet = { version = "54.2", default-features = false, features = ["arrow", "as
 paste = "1.0"
 pin-project = "1.0"
 prometheus = { version = "0.13.3", features = ["process"] }
-promql-parser = { git = "https://github.com/GreptimeTeam/promql-parser.git", rev = "0410e8b459dda7cb222ce9596f8bf3971bd07bd2", features = [
-    "ser",
-] }
+promql-parser = { version = "0.6", features = ["ser"] }
 prost = { version = "0.13", features = ["no-recursion-limit"] }
 raft-engine = { version = "0.4.1", default-features = false }
 rand = "0.9"
@@ -221,8 +223,6 @@ tokio-util = { version = "0.7", features = ["io-util", "compat"] }
 toml = "0.8.8"
 tonic = { version = "0.12", features = ["tls", "gzip", "zstd"] }
 tower = "0.5"
-tower-http = "0.6"
-tracing = "0.1"
 tracing-appender = "0.2"
 tracing-subscriber = { version = "0.3", features = ["env-filter", "json", "fmt"] }
 typetag = "0.2"
@@ -261,6 +261,7 @@ common-query = { path = "src/common/query" }
 common-recordbatch = { path = "src/common/recordbatch" }
 common-runtime = { path = "src/common/runtime" }
 common-session = { path = "src/common/session" }
+common-sql = { path = "src/common/sql" }
 common-telemetry = { path = "src/common/telemetry" }
 common-test-util = { path = "src/common/test-util" }
 common-time = { path = "src/common/time" }
--- a/2
+++ b/2
@@ -8,7 +8,7 @@ CARGO_BUILD_OPTS := --locked
 IMAGE_REGISTRY ?= docker.io
 IMAGE_NAMESPACE ?= greptime
 IMAGE_TAG ?= latest
-DEV_BUILDER_IMAGE_TAG ?= 2025-05-19-f55023f3-20250829091211
+DEV_BUILDER_IMAGE_TAG ?= 2025-05-19-b2377d4b-20250520045554
 BUILDX_MULTI_PLATFORM_BUILD ?= false
 BUILDX_BUILDER_NAME ?= gtbuilder
 BASE_IMAGE ?= ubuntu
--- a/README.md
+++ b/README.md
@@ -75,9 +75,9 @@
 | --------- | ----------- |
 | [Unified Observability Data](https://docs.greptime.com/user-guide/concepts/why-greptimedb) | Store metrics, logs, and traces as timestamped, contextual wide events. Query via [SQL](https://docs.greptime.com/user-guide/query-data/sql), [PromQL](https://docs.greptime.com/user-guide/query-data/promql), and [streaming](https://docs.greptime.com/user-guide/flow-computation/overview). |
 | [High Performance & Cost Effective](https://docs.greptime.com/user-guide/manage-data/data-index) | Written in Rust, with a distributed query engine, [rich indexing](https://docs.greptime.com/user-guide/manage-data/data-index), and optimized columnar storage, delivering sub-second responses at PB scale. |
-| [Cloud-Native Architecture](https://docs.greptime.com/user-guide/concepts/architecture) | Designed for [Kubernetes](https://docs.greptime.com/user-guide/deployments/deploy-on-kubernetes/greptimedb-operator-management), with compute/storage separation, native object storage (AWS S3, Azure Blob, etc.) and seamless cross-cloud access. |
+| [Cloud-Native Architecture](https://docs.greptime.com/user-guide/concepts/architecture) | Designed for [Kubernetes](https://docs.greptime.com/user-guide/deployments-administration/deploy-on-kubernetes/greptimedb-operator-management), with compute/storage separation, native object storage (AWS S3, Azure Blob, etc.) and seamless cross-cloud access. |
 | [Developer-Friendly](https://docs.greptime.com/user-guide/protocols/overview) | Access via SQL/PromQL interfaces, REST API, MySQL/PostgreSQL protocols, and popular ingestion [protocols](https://docs.greptime.com/user-guide/protocols/overview). |
-| [Flexible Deployment](https://docs.greptime.com/user-guide/deployments/overview) | Deploy anywhere: edge (including ARM/[Android](https://docs.greptime.com/user-guide/deployments/run-on-android)) or cloud, with unified APIs and efficient data sync. |
+| [Flexible Deployment](https://docs.greptime.com/user-guide/deployments-administration/overview) | Deploy anywhere: edge (including ARM/[Android](https://docs.greptime.com/user-guide/deployments-administration/run-on-android)) or cloud, with unified APIs and efficient data sync. |

 Learn more in [Why GreptimeDB](https://docs.greptime.com/user-guide/concepts/why-greptimedb) and [Observability 2.0 and the Database for It](https://greptime.com/blogs/2025-04-25-greptimedb-observability2-new-database).

--- a/config/config.md
+++ b/config/config.md
@@ -147,7 +147,6 @@
 | `region_engine.mito.write_cache_ttl` | String | Unset | TTL for write cache. |
 | `region_engine.mito.sst_write_buffer_size` | String | `8MB` | Buffer size for SST writing. |
 | `region_engine.mito.parallel_scan_channel_size` | Integer | `32` | Capacity of the channel to send data from parallel scan tasks to the main task. |
-| `region_engine.mito.max_concurrent_scan_files` | Integer | `128` | Maximum number of SST files to scan concurrently. |
 | `region_engine.mito.allow_stale_entries` | Bool | `false` | Whether to allow stale WAL entries read during replay. |
 | `region_engine.mito.min_compaction_interval` | String | `0m` | Minimum time interval between two compactions.<br/>To align with the old behavior, the default value is 0 (no restrictions). |
 | `region_engine.mito.index` | -- | -- | The options for index in Mito engine. |
@@ -186,10 +185,11 @@
 | `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:4317` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4318` | 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,10 +289,11 @@
 | `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:4317` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4318` | 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. |
@@ -324,6 +325,7 @@
 | `selector` | String | `round_robin` | Datanode selector type.<br/>- `round_robin` (default value)<br/>- `lease_based`<br/>- `load_based`<br/>For details, please see "https://docs.greptime.com/developer-guide/metasrv/selector". |
 | `use_memory_store` | Bool | `false` | Store data in memory. |
 | `enable_region_failover` | Bool | `false` | Whether to enable region failover.<br/>This feature is only available on GreptimeDB running on cluster mode and<br/>- Using Remote WAL<br/>- Using shared storage (e.g., s3). |
+| `region_failure_detector_initialization_delay` | String | `10m` | The delay before starting region failure detection.<br/>This delay helps prevent Metasrv from triggering unnecessary region failovers before all Datanodes are fully started.<br/>Especially useful when the cluster is not deployed with GreptimeDB Operator and maintenance mode is not enabled. |
 | `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. |
@@ -371,10 +373,11 @@
 | `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:4317` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4318` | 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. |
@@ -433,8 +436,8 @@
 | `wal.provider` | String | `raft_engine` | The provider of the WAL.<br/>- `raft_engine`: the wal is stored in the local file system by raft-engine.<br/>- `kafka`: it's remote wal that data is stored in Kafka. |
 | `wal.dir` | String | Unset | The directory to store the WAL files.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.file_size` | String | `128MB` | The size of the WAL segment file.<br/>**It's only used when the provider is `raft_engine`**. |
-| `wal.purge_threshold` | String | `1GB` | The threshold of the WAL size to trigger a flush.<br/>**It's only used when the provider is `raft_engine`**. |
-| `wal.purge_interval` | String | `1m` | The interval to trigger a flush.<br/>**It's only used when the provider is `raft_engine`**. |
+| `wal.purge_threshold` | String | `1GB` | The threshold of the WAL size to trigger a purge.<br/>**It's only used when the provider is `raft_engine`**. |
+| `wal.purge_interval` | String | `1m` | The interval to trigger a purge.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.read_batch_size` | Integer | `128` | The read batch size.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.sync_write` | Bool | `false` | Whether to use sync write.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.enable_log_recycle` | Bool | `true` | Whether to reuse logically truncated log files.<br/>**It's only used when the provider is `raft_engine`**. |
@@ -497,7 +500,6 @@
 | `region_engine.mito.write_cache_ttl` | String | Unset | TTL for write cache. |
 | `region_engine.mito.sst_write_buffer_size` | String | `8MB` | Buffer size for SST writing. |
 | `region_engine.mito.parallel_scan_channel_size` | Integer | `32` | Capacity of the channel to send data from parallel scan tasks to the main task. |
-| `region_engine.mito.max_concurrent_scan_files` | Integer | `128` | Maximum number of SST files to scan concurrently. |
 | `region_engine.mito.allow_stale_entries` | Bool | `false` | Whether to allow stale WAL entries read during replay. |
 | `region_engine.mito.min_compaction_interval` | String | `0m` | Minimum time interval between two compactions.<br/>To align with the old behavior, the default value is 0 (no restrictions). |
 | `region_engine.mito.index` | -- | -- | The options for index in Mito engine. |
@@ -536,10 +538,11 @@
 | `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:4317` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4318` | 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. |
@@ -586,11 +589,14 @@
 | `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:4317` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4318` | 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. |
 | `tracing.tokio_console_addr` | String | Unset | The tokio console address. |
+| `query` | -- | -- | -- |
+| `query.parallelism` | Integer | `1` | Parallelism of the query engine for query sent by flownode.<br/>Default to 1, so it won't use too much cpu or memory |
--- a/config/datanode.example.toml
+++ b/config/datanode.example.toml
@@ -129,11 +129,11 @@ dir = "./greptimedb_data/wal"
 ## **It's only used when the provider is `raft_engine`**.
 file_size = "128MB"

-## The threshold of the WAL size to trigger a flush.
+## The threshold of the WAL size to trigger a purge.
 ## **It's only used when the provider is `raft_engine`**.
 purge_threshold = "1GB"

-## The interval to trigger a flush.
+## The interval to trigger a purge.
 ## **It's only used when the provider is `raft_engine`**.
 purge_interval = "1m"

@@ -474,9 +474,6 @@ sst_write_buffer_size = "8MB"
 ## Capacity of the channel to send data from parallel scan tasks to the main task.
 parallel_scan_channel_size = 32

-## Maximum number of SST files to scan concurrently.
-max_concurrent_scan_files = 128
-
 ## Whether to allow stale WAL entries read during replay.
 allow_stale_entries = false

@@ -632,7 +629,7 @@ level = "info"
 enable_otlp_tracing = false

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

 ## Whether to append logs to stdout.
 append_stdout = true
@@ -643,6 +640,9 @@ 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:4317"
+otlp_endpoint = "http://localhost:4318"

 ## Whether to append logs to stdout.
 append_stdout = true
@@ -94,6 +94,9 @@ 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
@@ -105,3 +108,8 @@ default_ratio = 1.0
 ## The tokio console address.
 ## @toml2docs:none-default
 #+ tokio_console_addr = "127.0.0.1"
+
+[query]
+## Parallelism of the query engine for query sent by flownode.
+## Default to 1, so it won't use too much cpu or memory
+parallelism = 1
--- 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:4317"
+otlp_endpoint = "http://localhost:4318"

 ## Whether to append logs to stdout.
 append_stdout = true
@@ -229,6 +229,9 @@ 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,6 +43,11 @@ use_memory_store = false
 ## - Using shared storage (e.g., s3).
 enable_region_failover = false

+## The delay before starting region failure detection.
+## This delay helps prevent Metasrv from triggering unnecessary region failovers before all Datanodes are fully started.
+## Especially useful when the cluster is not deployed with GreptimeDB Operator and maintenance mode is not enabled.
+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
@@ -220,7 +225,7 @@ level = "info"
 enable_otlp_tracing = false

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

 ## Whether to append logs to stdout.
 append_stdout = true
@@ -231,6 +236,9 @@ 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
@@ -565,9 +565,6 @@ sst_write_buffer_size = "8MB"
 ## Capacity of the channel to send data from parallel scan tasks to the main task.
 parallel_scan_channel_size = 32

-## Maximum number of SST files to scan concurrently.
-max_concurrent_scan_files = 128
-
 ## Whether to allow stale WAL entries read during replay.
 allow_stale_entries = false

@@ -723,7 +720,7 @@ level = "info"
 enable_otlp_tracing = false

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

 ## Whether to append logs to stdout.
 append_stdout = true
@@ -734,6 +731,9 @@ 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/cyborg/bin/follow-up-docs-issue.ts
+++ b/cyborg/bin/follow-up-docs-issue.ts
@@ -55,12 +55,25 @@ async function main() {
        await client.rest.issues.addLabels({
            owner, repo, issue_number: number, labels: [labelDocsRequired],
        })
+
+        // Get available assignees for the docs repo
+        const assigneesResponse = await docsClient.rest.issues.listAssignees({
+            owner: 'GreptimeTeam',
+            repo: 'docs',
+        })
+        const validAssignees = assigneesResponse.data.map(assignee => assignee.login)
+        core.info(`Available assignees: ${validAssignees.join(', ')}`)
+
+        // Check if the actor is a valid assignee, otherwise fallback to fengjiachun
+        const assignee = validAssignees.includes(actor) ? actor : 'fengjiachun'
+        core.info(`Assigning issue to: ${assignee}`)
+
        await docsClient.rest.issues.create({
            owner: 'GreptimeTeam',
            repo: 'docs',
            title: `Update docs for ${title}`,
            body: `A document change request is generated from ${html_url}`,
-            assignee: actor,
+            assignee: assignee,
        }).then((res) => {
            core.info(`Created issue ${res.data}`)
        })
--- a/docs/benchmarks/log/README.md
+++ b/docs/benchmarks/log/README.md
@@ -48,4 +48,4 @@ Please refer to [SQL query](./query.sql) for GreptimeDB and Clickhouse, and [que

 ## Addition
 - You can tune GreptimeDB's configuration to get better performance.
- You can setup GreptimeDB to use S3 as storage, see [here](https://docs.greptime.com/user-guide/deployments/configuration#storage-options).
+- You can setup GreptimeDB to use S3 as storage, see [here](https://docs.greptime.com/user-guide/deployments-administration/configuration#storage-options).
--- a/grafana/README.md
+++ b/grafana/README.md
@@ -83,7 +83,7 @@ If you use the [Helm Chart](https://github.com/GreptimeTeam/helm-charts) to depl
 - `monitoring.enabled=true`: Deploys a standalone GreptimeDB instance dedicated to monitoring the cluster;
 - `grafana.enabled=true`: Deploys Grafana and automatically imports the monitoring dashboard;

-The standalone GreptimeDB instance will collect metrics from your cluster, and the dashboard will be available in the Grafana UI. For detailed deployment instructions, please refer to our [Kubernetes deployment guide](https://docs.greptime.com/user-guide/deployments-administration/deploy-on-kubernetes/getting-started).
+The standalone GreptimeDB instance will collect metrics from your cluster, and the dashboard will be available in the Grafana UI. For detailed deployment instructions, please refer to our [Kubernetes deployment guide](https://docs.greptime.com/user-guide/deployments-administration-administration/deploy-on-kubernetes/getting-started).

 ### Self-host Prometheus and import dashboards manually

--- a/licenserc.toml
+++ b/licenserc.toml
@@ -31,8 +31,10 @@ 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",
+    "src/mito2/src/extension.rs",
 ]

 [properties]
--- a/src/api/src/v1/column_def.rs
+++ b/src/api/src/v1/column_def.rs
@@ -24,7 +24,7 @@ use greptime_proto::v1::{
 };
 use snafu::ResultExt;

-use crate::error::{self, ConvertColumnDefaultConstraintSnafu, Result};
+use crate::error::{self, Result};
 use crate::helper::ColumnDataTypeWrapper;
 use crate::v1::{ColumnDef, ColumnOptions, SemanticType};

@@ -77,48 +77,6 @@ pub fn try_as_column_schema(column_def: &ColumnDef) -> Result<ColumnSchema> {
        })
 }

-/// Tries to construct a `ColumnDef` from the given `ColumnSchema`.
-///
-/// TODO(weny): Add tests for this function.
-pub fn try_as_column_def(column_schema: &ColumnSchema, is_primary_key: bool) -> Result<ColumnDef> {
-    let column_datatype =
-        ColumnDataTypeWrapper::try_from(column_schema.data_type.clone()).map(|w| w.to_parts())?;
-
-    let semantic_type = if column_schema.is_time_index() {
-        SemanticType::Timestamp
-    } else if is_primary_key {
-        SemanticType::Tag
-    } else {
-        SemanticType::Field
-    } as i32;
-    let comment = column_schema
-        .metadata()
-        .get(COMMENT_KEY)
-        .cloned()
-        .unwrap_or_default();
-
-    let default_constraint = match column_schema.default_constraint() {
-        None => vec![],
-        Some(v) => v
-            .clone()
-            .try_into()
-            .context(ConvertColumnDefaultConstraintSnafu {
-                column: &column_schema.name,
-            })?,
-    };
-    let options = options_from_column_schema(column_schema);
-    Ok(ColumnDef {
-        name: column_schema.name.clone(),
-        data_type: column_datatype.0 as i32,
-        is_nullable: column_schema.is_nullable(),
-        default_constraint,
-        semantic_type,
-        comment,
-        datatype_extension: column_datatype.1,
-        options,
-    })
-}
-
 /// Constructs a `ColumnOptions` from the given `ColumnSchema`.
 pub fn options_from_column_schema(column_schema: &ColumnSchema) -> Option<ColumnOptions> {
    let mut options = ColumnOptions::default();
--- a/src/catalog/Cargo.toml
+++ b/src/catalog/Cargo.toml
@@ -5,6 +5,7 @@ edition.workspace = true
 license.workspace = true

 [features]
+enterprise = []
 testing = []

 [lints]
@@ -43,8 +44,6 @@ moka = { workspace = true, features = ["future", "sync"] }
 partition.workspace = true
 paste.workspace = true
 prometheus.workspace = true
-promql-parser.workspace = true
-rand.workspace = true
 rustc-hash.workspace = true
 serde_json.workspace = true
 session.workspace = true
--- a/src/catalog/src/information_extension.rs
+++ b/src/catalog/src/information_extension.rs
@@ -16,8 +16,8 @@ use api::v1::meta::ProcedureStatus;
 use common_error::ext::BoxedError;
 use common_meta::cluster::{ClusterInfo, NodeInfo};
 use common_meta::datanode::RegionStat;
+use common_meta::ddl::{ExecutorContext, ProcedureExecutor};
 use common_meta::key::flow::flow_state::FlowStat;
-use common_meta::procedure_executor::{ExecutorContext, ProcedureExecutor};
 use common_meta::rpc::procedure;
 use common_procedure::{ProcedureInfo, ProcedureState};
 use meta_client::MetaClientRef;
--- a/src/catalog/src/kvbackend.rs
+++ b/src/catalog/src/kvbackend.rs
@@ -14,9 +14,11 @@

 pub use client::{CachedKvBackend, CachedKvBackendBuilder, MetaKvBackend};

+mod builder;
 mod client;
 mod manager;
 mod table_cache;

+pub use builder::KvBackendCatalogManagerBuilder;
 pub use manager::KvBackendCatalogManager;
 pub use table_cache::{new_table_cache, TableCache, TableCacheRef};
--- a/src/catalog/src/kvbackend/builder.rs
+++ b/src/catalog/src/kvbackend/builder.rs
@@ -0,0 +1,131 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use common_catalog::consts::DEFAULT_CATALOG_NAME;
+use common_meta::cache::LayeredCacheRegistryRef;
+use common_meta::key::flow::FlowMetadataManager;
+use common_meta::key::TableMetadataManager;
+use common_meta::kv_backend::KvBackendRef;
+use common_procedure::ProcedureManagerRef;
+use moka::sync::Cache;
+use partition::manager::PartitionRuleManager;
+
+#[cfg(feature = "enterprise")]
+use crate::information_schema::InformationSchemaTableFactoryRef;
+use crate::information_schema::{InformationExtensionRef, InformationSchemaProvider};
+use crate::kvbackend::manager::{SystemCatalog, CATALOG_CACHE_MAX_CAPACITY};
+use crate::kvbackend::KvBackendCatalogManager;
+use crate::process_manager::ProcessManagerRef;
+use crate::system_schema::pg_catalog::PGCatalogProvider;
+
+pub struct KvBackendCatalogManagerBuilder {
+    information_extension: InformationExtensionRef,
+    backend: KvBackendRef,
+    cache_registry: LayeredCacheRegistryRef,
+    procedure_manager: Option<ProcedureManagerRef>,
+    process_manager: Option<ProcessManagerRef>,
+    #[cfg(feature = "enterprise")]
+    extra_information_table_factories:
+        std::collections::HashMap<String, InformationSchemaTableFactoryRef>,
+}
+
+impl KvBackendCatalogManagerBuilder {
+    pub fn new(
+        information_extension: InformationExtensionRef,
+        backend: KvBackendRef,
+        cache_registry: LayeredCacheRegistryRef,
+    ) -> Self {
+        Self {
+            information_extension,
+            backend,
+            cache_registry,
+            procedure_manager: None,
+            process_manager: None,
+            #[cfg(feature = "enterprise")]
+            extra_information_table_factories: std::collections::HashMap::new(),
+        }
+    }
+
+    pub fn with_procedure_manager(mut self, procedure_manager: ProcedureManagerRef) -> Self {
+        self.procedure_manager = Some(procedure_manager);
+        self
+    }
+
+    pub fn with_process_manager(mut self, process_manager: ProcessManagerRef) -> Self {
+        self.process_manager = Some(process_manager);
+        self
+    }
+
+    /// Sets the extra information tables.
+    #[cfg(feature = "enterprise")]
+    pub fn with_extra_information_table_factories(
+        mut self,
+        factories: std::collections::HashMap<String, InformationSchemaTableFactoryRef>,
+    ) -> Self {
+        self.extra_information_table_factories = factories;
+        self
+    }
+
+    pub fn build(self) -> Arc<KvBackendCatalogManager> {
+        let Self {
+            information_extension,
+            backend,
+            cache_registry,
+            procedure_manager,
+            process_manager,
+            #[cfg(feature = "enterprise")]
+            extra_information_table_factories,
+        } = self;
+        Arc::new_cyclic(|me| KvBackendCatalogManager {
+            information_extension,
+            partition_manager: Arc::new(PartitionRuleManager::new(
+                backend.clone(),
+                cache_registry
+                    .get()
+                    .expect("Failed to get table_route_cache"),
+            )),
+            table_metadata_manager: Arc::new(TableMetadataManager::new(backend.clone())),
+            system_catalog: SystemCatalog {
+                catalog_manager: me.clone(),
+                catalog_cache: Cache::new(CATALOG_CACHE_MAX_CAPACITY),
+                pg_catalog_cache: Cache::new(CATALOG_CACHE_MAX_CAPACITY),
+                information_schema_provider: {
+                    let provider = InformationSchemaProvider::new(
+                        DEFAULT_CATALOG_NAME.to_string(),
+                        me.clone(),
+                        Arc::new(FlowMetadataManager::new(backend.clone())),
+                        process_manager.clone(),
+                        backend.clone(),
+                    );
+                    #[cfg(feature = "enterprise")]
+                    let provider = provider
+                        .with_extra_table_factories(extra_information_table_factories.clone());
+                    Arc::new(provider)
+                },
+                pg_catalog_provider: Arc::new(PGCatalogProvider::new(
+                    DEFAULT_CATALOG_NAME.to_string(),
+                    me.clone(),
+                )),
+                backend,
+                process_manager,
+                #[cfg(feature = "enterprise")]
+                extra_information_table_factories,
+            },
+            cache_registry,
+            procedure_manager,
+        })
+    }
+}
--- a/src/catalog/src/kvbackend/manager.rs
+++ b/src/catalog/src/kvbackend/manager.rs
@@ -30,13 +30,13 @@ use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::schema_name::SchemaNameKey;
 use common_meta::key::table_info::{TableInfoManager, TableInfoValue};
 use common_meta::key::table_name::TableNameKey;
-use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
+use common_meta::key::TableMetadataManagerRef;
 use common_meta::kv_backend::KvBackendRef;
 use common_procedure::ProcedureManagerRef;
 use futures_util::stream::BoxStream;
 use futures_util::{StreamExt, TryStreamExt};
 use moka::sync::Cache;
-use partition::manager::{PartitionRuleManager, PartitionRuleManagerRef};
+use partition::manager::PartitionRuleManagerRef;
 use session::context::{Channel, QueryContext};
 use snafu::prelude::*;
 use store_api::metric_engine_consts::METRIC_ENGINE_NAME;
@@ -52,6 +52,8 @@ use crate::error::{
    CacheNotFoundSnafu, GetTableCacheSnafu, InvalidTableInfoInCatalogSnafu, ListCatalogsSnafu,
    ListSchemasSnafu, ListTablesSnafu, Result, TableMetadataManagerSnafu,
 };
+#[cfg(feature = "enterprise")]
+use crate::information_schema::InformationSchemaTableFactoryRef;
 use crate::information_schema::{InformationExtensionRef, InformationSchemaProvider};
 use crate::kvbackend::TableCacheRef;
 use crate::process_manager::ProcessManagerRef;
@@ -67,60 +69,22 @@ use crate::CatalogManager;
 #[derive(Clone)]
 pub struct KvBackendCatalogManager {
    /// Provides the extension methods for the `information_schema` tables
-    information_extension: InformationExtensionRef,
+    pub(super) information_extension: InformationExtensionRef,
    /// Manages partition rules.
-    partition_manager: PartitionRuleManagerRef,
+    pub(super) partition_manager: PartitionRuleManagerRef,
    /// Manages table metadata.
-    table_metadata_manager: TableMetadataManagerRef,
+    pub(super) table_metadata_manager: TableMetadataManagerRef,
    /// A sub-CatalogManager that handles system tables
-    system_catalog: SystemCatalog,
+    pub(super) system_catalog: SystemCatalog,
    /// Cache registry for all caches.
-    cache_registry: LayeredCacheRegistryRef,
+    pub(super) cache_registry: LayeredCacheRegistryRef,
    /// Only available in `Standalone` mode.
-    procedure_manager: Option<ProcedureManagerRef>,
+    pub(super) procedure_manager: Option<ProcedureManagerRef>,
 }

-const CATALOG_CACHE_MAX_CAPACITY: u64 = 128;
+pub(super) const CATALOG_CACHE_MAX_CAPACITY: u64 = 128;

 impl KvBackendCatalogManager {
-    pub fn new(
-        information_extension: InformationExtensionRef,
-        backend: KvBackendRef,
-        cache_registry: LayeredCacheRegistryRef,
-        procedure_manager: Option<ProcedureManagerRef>,
-        process_manager: Option<ProcessManagerRef>,
-    ) -> Arc<Self> {
-        Arc::new_cyclic(|me| Self {
-            information_extension,
-            partition_manager: Arc::new(PartitionRuleManager::new(
-                backend.clone(),
-                cache_registry
-                    .get()
-                    .expect("Failed to get table_route_cache"),
-            )),
-            table_metadata_manager: Arc::new(TableMetadataManager::new(backend.clone())),
-            system_catalog: SystemCatalog {
-                catalog_manager: me.clone(),
-                catalog_cache: Cache::new(CATALOG_CACHE_MAX_CAPACITY),
-                pg_catalog_cache: Cache::new(CATALOG_CACHE_MAX_CAPACITY),
-                information_schema_provider: Arc::new(InformationSchemaProvider::new(
-                    DEFAULT_CATALOG_NAME.to_string(),
-                    me.clone(),
-                    Arc::new(FlowMetadataManager::new(backend.clone())),
-                    process_manager.clone(),
-                )),
-                pg_catalog_provider: Arc::new(PGCatalogProvider::new(
-                    DEFAULT_CATALOG_NAME.to_string(),
-                    me.clone(),
-                )),
-                backend,
-                process_manager,
-            },
-            cache_registry,
-            procedure_manager,
-        })
-    }
-
    pub fn view_info_cache(&self) -> Result<ViewInfoCacheRef> {
        self.cache_registry.get().context(CacheNotFoundSnafu {
            name: "view_info_cache",
@@ -507,16 +471,19 @@ fn build_table(table_info_value: TableInfoValue) -> Result<TableRef> {
 /// - information_schema.{tables}
 /// - pg_catalog.{tables}
 #[derive(Clone)]
-struct SystemCatalog {
-    catalog_manager: Weak<KvBackendCatalogManager>,
-    catalog_cache: Cache<String, Arc<InformationSchemaProvider>>,
-    pg_catalog_cache: Cache<String, Arc<PGCatalogProvider>>,
+pub(super) struct SystemCatalog {
+    pub(super) catalog_manager: Weak<KvBackendCatalogManager>,
+    pub(super) catalog_cache: Cache<String, Arc<InformationSchemaProvider>>,
+    pub(super) pg_catalog_cache: Cache<String, Arc<PGCatalogProvider>>,

    // system_schema_provider for default catalog
-    information_schema_provider: Arc<InformationSchemaProvider>,
-    pg_catalog_provider: Arc<PGCatalogProvider>,
-    backend: KvBackendRef,
-    process_manager: Option<ProcessManagerRef>,
+    pub(super) information_schema_provider: Arc<InformationSchemaProvider>,
+    pub(super) pg_catalog_provider: Arc<PGCatalogProvider>,
+    pub(super) backend: KvBackendRef,
+    pub(super) process_manager: Option<ProcessManagerRef>,
+    #[cfg(feature = "enterprise")]
+    pub(super) extra_information_table_factories:
+        std::collections::HashMap<String, InformationSchemaTableFactoryRef>,
 }

 impl SystemCatalog {
@@ -580,12 +547,17 @@ impl SystemCatalog {
        if schema == INFORMATION_SCHEMA_NAME {
            let information_schema_provider =
                self.catalog_cache.get_with_by_ref(catalog, move || {
-                    Arc::new(InformationSchemaProvider::new(
+                    let provider = InformationSchemaProvider::new(
                        catalog.to_string(),
                        self.catalog_manager.clone(),
                        Arc::new(FlowMetadataManager::new(self.backend.clone())),
                        self.process_manager.clone(),
-                    ))
+                        self.backend.clone(),
+                    );
+                    #[cfg(feature = "enterprise")]
+                    let provider = provider
+                        .with_extra_table_factories(self.extra_information_table_factories.clone());
+                    Arc::new(provider)
                });
            information_schema_provider.table(table_name)
        } else if schema == PG_CATALOG_NAME && channel == Channel::Postgres {
--- a/src/catalog/src/memory/manager.rs
+++ b/src/catalog/src/memory/manager.rs
@@ -352,11 +352,13 @@ impl MemoryCatalogManager {
    }

    fn create_catalog_entry(self: &Arc<Self>, catalog: String) -> SchemaEntries {
+        let backend = Arc::new(MemoryKvBackend::new());
        let information_schema_provider = InformationSchemaProvider::new(
            catalog,
            Arc::downgrade(self) as Weak<dyn CatalogManager>,
-            Arc::new(FlowMetadataManager::new(Arc::new(MemoryKvBackend::new()))),
+            Arc::new(FlowMetadataManager::new(backend.clone())),
            None, // we don't need ProcessManager on regions server.
+            backend,
        );
        let information_schema = information_schema_provider.tables().clone();

--- a/src/catalog/src/process_manager.rs
+++ b/src/catalog/src/process_manager.rs
@@ -14,24 +14,17 @@

 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
-use std::fmt::{Debug, Display, Formatter};
+use std::fmt::{Debug, Formatter};
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::{Arc, RwLock};
-use std::time::{Duration, Instant, UNIX_EPOCH};

 use api::v1::frontend::{KillProcessRequest, ListProcessRequest, ProcessInfo};
 use common_base::cancellation::CancellationHandle;
 use common_frontend::selector::{FrontendSelector, MetaClientSelector};
-use common_frontend::slow_query_event::SlowQueryEvent;
-use common_telemetry::{debug, error, info, warn};
+use common_telemetry::{debug, info, warn};
 use common_time::util::current_time_millis;
 use meta_client::MetaClientRef;
-use promql_parser::parser::EvalStmt;
-use rand::random;
-use session::context::QueryContextRef;
 use snafu::{ensure, OptionExt, ResultExt};
-use sql::statements::statement::Statement;
-use tokio::sync::mpsc::Sender;

 use crate::error;
 use crate::metrics::{PROCESS_KILL_COUNT, PROCESS_LIST_COUNT};
@@ -51,23 +44,6 @@ pub struct ProcessManager {
    frontend_selector: Option<MetaClientSelector>,
 }

-/// Represents a parsed query statement, functionally equivalent to [query::parser::QueryStatement].
-/// This enum is defined here to avoid cyclic dependencies with the query parser module.
-#[derive(Debug, Clone)]
-pub enum QueryStatement {
-    Sql(Statement),
-    Promql(EvalStmt),
-}
-
-impl Display for QueryStatement {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        match self {
-            QueryStatement::Sql(stmt) => write!(f, "{}", stmt),
-            QueryStatement::Promql(eval_stmt) => write!(f, "{}", eval_stmt),
-        }
-    }
-}
-
 impl ProcessManager {
    /// Create a [ProcessManager] instance with server address and kv client.
    pub fn new(server_addr: String, meta_client: Option<MetaClientRef>) -> Self {
@@ -91,7 +67,6 @@ impl ProcessManager {
        query: String,
        client: String,
        query_id: Option<ProcessId>,
-        _slow_query_timer: Option<SlowQueryTimer>,
    ) -> Ticket {
        let id = query_id.unwrap_or_else(|| self.next_id.fetch_add(1, Ordering::Relaxed));
        let process = ProcessInfo {
@@ -118,7 +93,6 @@ impl ProcessManager {
            manager: self.clone(),
            id,
            cancellation_handle,
-            _slow_query_timer,
        }
    }

@@ -249,7 +223,6 @@ pub struct Ticket {
    pub(crate) manager: ProcessManagerRef,
    pub(crate) id: ProcessId,
    pub cancellation_handle: Arc<CancellationHandle>,
-    _slow_query_timer: Option<SlowQueryTimer>,
 }

 impl Drop for Ticket {
@@ -290,107 +263,6 @@ impl Debug for CancellableProcess {
    }
 }

-/// SlowQueryTimer is used to log slow query when it's dropped.
-/// In drop(), it will check if the query is slow and send the slow query event to the handler.
-pub struct SlowQueryTimer {
-    start: Instant,
-    stmt: QueryStatement,
-    query_ctx: QueryContextRef,
-    threshold: Option<Duration>,
-    sample_ratio: Option<f64>,
-    tx: Sender<SlowQueryEvent>,
-}
-
-impl SlowQueryTimer {
-    pub fn new(
-        stmt: QueryStatement,
-        query_ctx: QueryContextRef,
-        threshold: Option<Duration>,
-        sample_ratio: Option<f64>,
-        tx: Sender<SlowQueryEvent>,
-    ) -> Self {
-        Self {
-            start: Instant::now(),
-            stmt,
-            query_ctx,
-            threshold,
-            sample_ratio,
-            tx,
-        }
-    }
-}
-
-impl SlowQueryTimer {
-    fn send_slow_query_event(&self, elapsed: Duration, threshold: Duration) {
-        let mut slow_query_event = SlowQueryEvent {
-            cost: elapsed.as_millis() as u64,
-            threshold: threshold.as_millis() as u64,
-            query: "".to_string(),
-            query_ctx: self.query_ctx.clone(),
-
-            // The following fields are only used for PromQL queries.
-            is_promql: false,
-            promql_range: None,
-            promql_step: None,
-            promql_start: None,
-            promql_end: None,
-        };
-
-        match &self.stmt {
-            QueryStatement::Promql(stmt) => {
-                slow_query_event.is_promql = true;
-                slow_query_event.query = stmt.expr.to_string();
-                slow_query_event.promql_step = Some(stmt.interval.as_millis() as u64);
-
-                let start = stmt
-                    .start
-                    .duration_since(UNIX_EPOCH)
-                    .unwrap_or_default()
-                    .as_millis() as i64;
-
-                let end = stmt
-                    .end
-                    .duration_since(UNIX_EPOCH)
-                    .unwrap_or_default()
-                    .as_millis() as i64;
-
-                slow_query_event.promql_range = Some((end - start) as u64);
-                slow_query_event.promql_start = Some(start);
-                slow_query_event.promql_end = Some(end);
-            }
-            QueryStatement::Sql(stmt) => {
-                slow_query_event.query = stmt.to_string();
-            }
-        }
-
-        // Send SlowQueryEvent to the handler.
-        if let Err(e) = self.tx.try_send(slow_query_event) {
-            error!(e; "Failed to send slow query event");
-        }
-    }
-}
-
-impl Drop for SlowQueryTimer {
-    fn drop(&mut self) {
-        if let Some(threshold) = self.threshold {
-            // Calculate the elaspsed duration since the timer is created.
-            let elapsed = self.start.elapsed();
-            if elapsed > threshold {
-                if let Some(ratio) = self.sample_ratio {
-                    // Only capture a portion of slow queries based on sample_ratio.
-                    // Generate a random number in [0, 1) and compare it with sample_ratio.
-                    if ratio >= 1.0 || random::<f64>() <= ratio {
-                        self.send_slow_query_event(elapsed, threshold);
-                    }
-                } else {
-                    // Captures all slow queries if sample_ratio is not set.
-                    self.send_slow_query_event(elapsed, threshold);
-                }
-            }
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
    use std::sync::Arc;
@@ -406,7 +278,6 @@ mod tests {
            "SELECT * FROM table".to_string(),
            "".to_string(),
            None,
-            None,
        );

        let running_processes = process_manager.local_processes(None).unwrap();
@@ -430,7 +301,6 @@ mod tests {
            "SELECT * FROM table".to_string(),
            "client1".to_string(),
            Some(custom_id),
-            None,
        );

        assert_eq!(ticket.id, custom_id);
@@ -451,7 +321,6 @@ mod tests {
            "SELECT * FROM table1".to_string(),
            "client1".to_string(),
            None,
-            None,
        );

        let ticket2 = process_manager.clone().register_query(
@@ -460,7 +329,6 @@ mod tests {
            "SELECT * FROM table2".to_string(),
            "client2".to_string(),
            None,
-            None,
        );

        let running_processes = process_manager.local_processes(Some("public")).unwrap();
@@ -482,7 +350,6 @@ mod tests {
            "SELECT * FROM table1".to_string(),
            "client1".to_string(),
            None,
-            None,
        );

        let _ticket2 = process_manager.clone().register_query(
@@ -491,7 +358,6 @@ mod tests {
            "SELECT * FROM table2".to_string(),
            "client2".to_string(),
            None,
-            None,
        );

        // Test listing processes for specific catalog
@@ -518,7 +384,6 @@ mod tests {
            "SELECT * FROM table".to_string(),
            "client1".to_string(),
            None,
-            None,
        );
        assert_eq!(process_manager.local_processes(None).unwrap().len(), 1);
        process_manager.deregister_query("public".to_string(), ticket.id);
@@ -535,7 +400,6 @@ mod tests {
            "SELECT * FROM table".to_string(),
            "client1".to_string(),
            None,
-            None,
        );

        assert!(!ticket.cancellation_handle.is_cancelled());
@@ -553,7 +417,6 @@ mod tests {
            "SELECT * FROM table".to_string(),
            "client1".to_string(),
            None,
-            None,
        );
        assert!(!ticket.cancellation_handle.is_cancelled());
        let killed = process_manager
@@ -599,7 +462,6 @@ mod tests {
            "SELECT COUNT(*) FROM users WHERE age > 18".to_string(),
            "test_client".to_string(),
            Some(42),
-            None,
        );

        let processes = process_manager.local_processes(None).unwrap();
@@ -626,7 +488,6 @@ mod tests {
                "SELECT * FROM table".to_string(),
                "client1".to_string(),
                None,
-                None,
            );

            // Process should be registered
--- a/src/catalog/src/system_schema.rs
+++ b/src/catalog/src/system_schema.rs
@@ -15,7 +15,7 @@
 pub mod information_schema;
 mod memory_table;
 pub mod pg_catalog;
-mod predicate;
+pub mod predicate;
 mod utils;

 use std::collections::HashMap;
@@ -96,7 +96,7 @@ trait SystemSchemaProviderInner {
    }
 }

-pub(crate) trait SystemTable {
+pub trait SystemTable {
    fn table_id(&self) -> TableId;

    fn table_name(&self) -> &'static str;
@@ -110,7 +110,7 @@ pub(crate) trait SystemTable {
    }
 }

-pub(crate) type SystemTableRef = Arc<dyn SystemTable + Send + Sync>;
+pub type SystemTableRef = Arc<dyn SystemTable + Send + Sync>;

 struct SystemTableDataSource {
    table: SystemTableRef,
--- 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;
-mod process_list;
+pub mod process_list;
 pub mod region_peers;
 mod region_statistics;
 mod runtime_metrics;
@@ -38,6 +38,7 @@ use common_meta::cluster::NodeInfo;
 use common_meta::datanode::RegionStat;
 use common_meta::key::flow::flow_state::FlowStat;
 use common_meta::key::flow::FlowMetadataManager;
+use common_meta::kv_backend::KvBackendRef;
 use common_procedure::ProcedureInfo;
 use common_recordbatch::SendableRecordBatchStream;
 use datatypes::schema::SchemaRef;
@@ -112,6 +113,25 @@ macro_rules! setup_memory_table {
    };
 }

+#[cfg(feature = "enterprise")]
+pub struct MakeInformationTableRequest {
+    pub catalog_name: String,
+    pub catalog_manager: Weak<dyn CatalogManager>,
+    pub kv_backend: KvBackendRef,
+}
+
+/// A factory trait for making information schema tables.
+///
+/// This trait allows for extensibility of the information schema by providing
+/// a way to dynamically create custom information schema tables.
+#[cfg(feature = "enterprise")]
+pub trait InformationSchemaTableFactory {
+    fn make_information_table(&self, req: MakeInformationTableRequest) -> SystemTableRef;
+}
+
+#[cfg(feature = "enterprise")]
+pub type InformationSchemaTableFactoryRef = Arc<dyn InformationSchemaTableFactory + Send + Sync>;
+
 /// The `information_schema` tables info provider.
 pub struct InformationSchemaProvider {
    catalog_name: String,
@@ -119,6 +139,10 @@ pub struct InformationSchemaProvider {
    process_manager: Option<ProcessManagerRef>,
    flow_metadata_manager: Arc<FlowMetadataManager>,
    tables: HashMap<String, TableRef>,
+    #[allow(dead_code)]
+    kv_backend: KvBackendRef,
+    #[cfg(feature = "enterprise")]
+    extra_table_factories: HashMap<String, InformationSchemaTableFactoryRef>,
 }

 impl SystemSchemaProvider for InformationSchemaProvider {
@@ -128,6 +152,7 @@ impl SystemSchemaProvider for InformationSchemaProvider {
        &self.tables
    }
 }
+
 impl SystemSchemaProviderInner for InformationSchemaProvider {
    fn catalog_name(&self) -> &str {
        &self.catalog_name
@@ -215,7 +240,22 @@ impl SystemSchemaProviderInner for InformationSchemaProvider {
                .process_manager
                .as_ref()
                .map(|p| Arc::new(InformationSchemaProcessList::new(p.clone())) as _),
-            _ => None,
+            table_name => {
+                #[cfg(feature = "enterprise")]
+                return self.extra_table_factories.get(table_name).map(|factory| {
+                    let req = MakeInformationTableRequest {
+                        catalog_name: self.catalog_name.clone(),
+                        catalog_manager: self.catalog_manager.clone(),
+                        kv_backend: self.kv_backend.clone(),
+                    };
+                    factory.make_information_table(req)
+                });
+                #[cfg(not(feature = "enterprise"))]
+                {
+                    let _ = table_name;
+                    None
+                }
+            }
        }
    }
 }
@@ -226,6 +266,7 @@ impl InformationSchemaProvider {
        catalog_manager: Weak<dyn CatalogManager>,
        flow_metadata_manager: Arc<FlowMetadataManager>,
        process_manager: Option<ProcessManagerRef>,
+        kv_backend: KvBackendRef,
    ) -> Self {
        let mut provider = Self {
            catalog_name,
@@ -233,6 +274,9 @@ impl InformationSchemaProvider {
            flow_metadata_manager,
            process_manager,
            tables: HashMap::new(),
+            kv_backend,
+            #[cfg(feature = "enterprise")]
+            extra_table_factories: HashMap::new(),
        };

        provider.build_tables();
@@ -240,6 +284,16 @@ impl InformationSchemaProvider {
        provider
    }

+    #[cfg(feature = "enterprise")]
+    pub(crate) fn with_extra_table_factories(
+        mut self,
+        factories: HashMap<String, InformationSchemaTableFactoryRef>,
+    ) -> Self {
+        self.extra_table_factories = factories;
+        self.build_tables();
+        self
+    }
+
    fn build_tables(&mut self) {
        let mut tables = HashMap::new();

@@ -290,16 +344,19 @@ impl InformationSchemaProvider {
        if let Some(process_list) = self.build_table(PROCESS_LIST) {
            tables.insert(PROCESS_LIST.to_string(), process_list);
        }
+        #[cfg(feature = "enterprise")]
+        for name in self.extra_table_factories.keys() {
+            tables.insert(name.to_string(), self.build_table(name).expect(name));
+        }
        // Add memory tables
        for name in MEMORY_TABLES.iter() {
            tables.insert((*name).to_string(), self.build_table(name).expect(name));
        }
-
        self.tables = tables;
    }
 }

-trait InformationTable {
+pub trait InformationTable {
    fn table_id(&self) -> TableId;

    fn table_name(&self) -> &'static str;
--- 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`
-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";
+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";

 /// `information_schema.process_list` table implementation that tracks running
 /// queries in current cluster.
--- a/src/catalog/src/system_schema/information_schema/region_statistics.rs
+++ b/src/catalog/src/system_schema/information_schema/region_statistics.rs
@@ -40,7 +40,6 @@ const REGION_ID: &str = "region_id";
 const TABLE_ID: &str = "table_id";
 const REGION_NUMBER: &str = "region_number";
 const REGION_ROWS: &str = "region_rows";
-const WRITTEN_BYTES: &str = "written_bytes_since_open";
 const DISK_SIZE: &str = "disk_size";
 const MEMTABLE_SIZE: &str = "memtable_size";
 const MANIFEST_SIZE: &str = "manifest_size";
@@ -57,7 +56,6 @@ const INIT_CAPACITY: usize = 42;
 /// - `table_id`: The table id.
 /// - `region_number`: The region number.
 /// - `region_rows`: The number of rows in region.
-/// - `written_bytes_since_open`: The total bytes written of the region since region opened.
 /// - `memtable_size`: The memtable size in bytes.
 /// - `disk_size`: The approximate disk size in bytes.
 /// - `manifest_size`: The manifest size in bytes.
@@ -85,7 +83,6 @@ impl InformationSchemaRegionStatistics {
            ColumnSchema::new(TABLE_ID, ConcreteDataType::uint32_datatype(), false),
            ColumnSchema::new(REGION_NUMBER, ConcreteDataType::uint32_datatype(), false),
            ColumnSchema::new(REGION_ROWS, ConcreteDataType::uint64_datatype(), true),
-            ColumnSchema::new(WRITTEN_BYTES, ConcreteDataType::uint64_datatype(), true),
            ColumnSchema::new(DISK_SIZE, ConcreteDataType::uint64_datatype(), true),
            ColumnSchema::new(MEMTABLE_SIZE, ConcreteDataType::uint64_datatype(), true),
            ColumnSchema::new(MANIFEST_SIZE, ConcreteDataType::uint64_datatype(), true),
@@ -148,7 +145,6 @@ struct InformationSchemaRegionStatisticsBuilder {
    table_ids: UInt32VectorBuilder,
    region_numbers: UInt32VectorBuilder,
    region_rows: UInt64VectorBuilder,
-    written_bytes: UInt64VectorBuilder,
    disk_sizes: UInt64VectorBuilder,
    memtable_sizes: UInt64VectorBuilder,
    manifest_sizes: UInt64VectorBuilder,
@@ -167,7 +163,6 @@ impl InformationSchemaRegionStatisticsBuilder {
            table_ids: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
            region_numbers: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
            region_rows: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
-            written_bytes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
            disk_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
            memtable_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
            manifest_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
@@ -198,7 +193,6 @@ impl InformationSchemaRegionStatisticsBuilder {
            (TABLE_ID, &Value::from(region_stat.id.table_id())),
            (REGION_NUMBER, &Value::from(region_stat.id.region_number())),
            (REGION_ROWS, &Value::from(region_stat.num_rows)),
-            (WRITTEN_BYTES, &Value::from(region_stat.written_bytes)),
            (DISK_SIZE, &Value::from(region_stat.approximate_bytes)),
            (MEMTABLE_SIZE, &Value::from(region_stat.memtable_size)),
            (MANIFEST_SIZE, &Value::from(region_stat.manifest_size)),
@@ -217,7 +211,6 @@ impl InformationSchemaRegionStatisticsBuilder {
        self.region_numbers
            .push(Some(region_stat.id.region_number()));
        self.region_rows.push(Some(region_stat.num_rows));
-        self.written_bytes.push(Some(region_stat.written_bytes));
        self.disk_sizes.push(Some(region_stat.approximate_bytes));
        self.memtable_sizes.push(Some(region_stat.memtable_size));
        self.manifest_sizes.push(Some(region_stat.manifest_size));
@@ -233,7 +226,6 @@ impl InformationSchemaRegionStatisticsBuilder {
            Arc::new(self.table_ids.finish()),
            Arc::new(self.region_numbers.finish()),
            Arc::new(self.region_rows.finish()),
-            Arc::new(self.written_bytes.finish()),
            Arc::new(self.disk_sizes.finish()),
            Arc::new(self.memtable_sizes.finish()),
            Arc::new(self.manifest_sizes.finish()),
--- a/src/catalog/src/system_schema/information_schema/table_names.rs
+++ b/src/catalog/src/system_schema/information_schema/table_names.rs
@@ -48,3 +48,4 @@ pub const FLOWS: &str = "flows";
 pub const PROCEDURE_INFO: &str = "procedure_info";
 pub const REGION_STATISTICS: &str = "region_statistics";
 pub const PROCESS_LIST: &str = "process_list";
+pub const TRIGGER_LIST: &str = "trigger_list";
--- a/src/catalog/src/table_source.rs
+++ b/src/catalog/src/table_source.rs
@@ -207,6 +207,7 @@ mod tests {
    use session::context::QueryContext;

    use super::*;
+    use crate::kvbackend::KvBackendCatalogManagerBuilder;
    use crate::memory::MemoryCatalogManager;

    #[test]
@@ -323,13 +324,13 @@ mod tests {
            .build(),
        );

-        let catalog_manager = KvBackendCatalogManager::new(
+        let catalog_manager = KvBackendCatalogManagerBuilder::new(
            Arc::new(NoopInformationExtension),
            backend.clone(),
            layered_cache_registry,
-            None,
-            None,
-        );
+        )
+        .build();
+
        let table_metadata_manager = TableMetadataManager::new(backend);
        let mut view_info = common_meta::key::test_utils::new_test_table_info(1024, vec![]);
        view_info.table_type = TableType::View;
--- a/src/cli/src/lib.rs
+++ b/src/cli/src/lib.rs
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+#![allow(clippy::print_stdout)]
 mod bench;
 mod data;
 mod database;
--- a/src/cli/src/metadata/control/get.rs
+++ b/src/cli/src/metadata/control/get.rs
@@ -29,7 +29,7 @@ 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_formatter};
+use crate::metadata::control::utils::{decode_key_value, get_table_id_by_name, json_fromatter};
 use crate::Tool;

 /// Getting metadata from metadata store.
@@ -206,7 +206,7 @@ impl Tool for GetTableTool {
            println!(
                "{}\n{}",
                TableInfoKey::new(table_id),
-                json_formatter(self.pretty, &*table_info)
+                json_fromatter(self.pretty, &*table_info)
            );
        } else {
            println!("Table info not found");
@@ -221,7 +221,7 @@ impl Tool for GetTableTool {
            println!(
                "{}\n{}",
                TableRouteKey::new(table_id),
-                json_formatter(self.pretty, &table_route)
+                json_fromatter(self.pretty, &table_route)
            );
        } else {
            println!("Table route not found");
--- a/src/cli/src/metadata/control/utils.rs
+++ b/src/cli/src/metadata/control/utils.rs
@@ -27,7 +27,7 @@ pub fn decode_key_value(kv: KeyValue) -> CommonMetaResult<(String, String)> {
 }

 /// Formats a value as a JSON string.
-pub fn json_formatter<T>(pretty: bool, value: &T) -> String
+pub fn json_fromatter<T>(pretty: bool, value: &T) -> String
 where
    T: Serialize,
 {
--- a/src/cli/src/metadata/snapshot.rs
+++ b/src/cli/src/metadata/snapshot.rs
@@ -301,7 +301,6 @@ struct MetaInfoTool {

 #[async_trait]
 impl Tool for MetaInfoTool {
-    #[allow(clippy::print_stdout)]
    async fn do_work(&self) -> std::result::Result<(), BoxedError> {
        let result = MetadataSnapshotManager::info(
            &self.inner,
--- a/src/client/src/database.rs
+++ b/src/client/src/database.rs
@@ -31,7 +31,7 @@ use base64::prelude::BASE64_STANDARD;
 use base64::Engine;
 use common_catalog::build_db_string;
 use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
-use common_error::ext::{BoxedError, ErrorExt};
+use common_error::ext::BoxedError;
 use common_grpc::flight::do_put::DoPutResponse;
 use common_grpc::flight::{FlightDecoder, FlightMessage};
 use common_query::Output;
@@ -48,7 +48,7 @@ use tonic::transport::Channel;

 use crate::error::{
    ConvertFlightDataSnafu, Error, FlightGetSnafu, IllegalFlightMessagesSnafu,
-    InvalidTonicMetadataValueSnafu, ServerSnafu,
+    InvalidTonicMetadataValueSnafu,
 };
 use crate::{error, from_grpc_response, Client, Result};

@@ -196,12 +196,22 @@ impl Database {

    /// Retry if connection fails, max_retries is the max number of retries, so the total wait time
    /// is `max_retries * GRPC_CONN_TIMEOUT`
-    pub async fn handle_with_retry(&self, request: Request, max_retries: u32) -> Result<u32> {
+    pub async fn handle_with_retry(
+        &self,
+        request: Request,
+        max_retries: u32,
+        hints: &[(&str, &str)],
+    ) -> Result<u32> {
        let mut client = make_database_client(&self.client)?.inner;
        let mut retries = 0;
+
        let request = self.to_rpc_request(request);
+
        loop {
-            let raw_response = client.handle(request.clone()).await;
+            let mut tonic_request = tonic::Request::new(request.clone());
+            let metadata = tonic_request.metadata_mut();
+            Self::put_hints(metadata, hints)?;
+            let raw_response = client.handle(tonic_request).await;
            match (raw_response, retries < max_retries) {
                (Ok(resp), _) => return from_grpc_response(resp.into_inner()),
                (Err(err), true) => {
@@ -211,18 +221,12 @@ impl Database {
                        retries += 1;
                        warn!("Retrying {} times with error = {:?}", retries, err);
                        continue;
-                    } else {
-                        error!(
-                            err; "Failed to send request to grpc handle, retries = {}, not retryable error, aborting",
-                            retries
-                        );
-                        return Err(err.into());
                    }
                }
                (Err(err), false) => {
                    error!(
-                        err; "Failed to send request to grpc handle after {} retries",
-                        retries,
+                        "Failed to send request to grpc handle after {} retries, error = {:?}",
+                        retries, err
                    );
                    return Err(err.into());
                }
@@ -298,21 +302,16 @@ impl Database {
        let response = client.mut_inner().do_get(request).await.or_else(|e| {
            let tonic_code = e.code();
            let e: Error = e.into();
-            let code = e.status_code();
-            let msg = e.to_string();
-            let error =
-                Err(BoxedError::new(ServerSnafu { code, msg }.build())).with_context(|_| {
-                    FlightGetSnafu {
-                        addr: client.addr().to_string(),
-                        tonic_code,
-                    }
-                });
            error!(
                "Failed to do Flight get, addr: {}, code: {}, source: {:?}",
                client.addr(),
                tonic_code,
-                error
+                e
            );
+            let error = Err(BoxedError::new(e)).with_context(|_| FlightGetSnafu {
+                addr: client.addr().to_string(),
+                tonic_code,
+            });
            error
        })?;

@@ -442,8 +441,11 @@ mod tests {

    use api::v1::auth_header::AuthScheme;
    use api::v1::{AuthHeader, Basic};
+    use common_error::status_code::StatusCode;
+    use tonic::{Code, Status};

    use super::*;
+    use crate::error::TonicSnafu;

    #[test]
    fn test_flight_ctx() {
@@ -466,4 +468,19 @@ mod tests {
            })
        )
    }
+
+    #[test]
+    fn test_from_tonic_status() {
+        let expected = TonicSnafu {
+            code: StatusCode::Internal,
+            msg: "blabla".to_string(),
+            tonic_code: Code::Internal,
+        }
+        .build();
+
+        let status = Status::new(Code::Internal, "blabla");
+        let actual: Error = status.into();
+
+        assert_eq!(expected.to_string(), actual.to_string());
+    }
 }
--- a/src/client/src/error.rs
+++ b/src/client/src/error.rs
@@ -14,13 +14,13 @@

 use std::any::Any;

+use common_error::define_from_tonic_status;
 use common_error::ext::{BoxedError, ErrorExt};
-use common_error::status_code::{convert_tonic_code_to_status_code, StatusCode};
-use common_error::{GREPTIME_DB_HEADER_ERROR_CODE, GREPTIME_DB_HEADER_ERROR_MSG};
+use common_error::status_code::StatusCode;
 use common_macro::stack_trace_debug;
 use snafu::{location, Location, Snafu};
 use tonic::metadata::errors::InvalidMetadataValue;
-use tonic::{Code, Status};
+use tonic::Code;

 #[derive(Snafu)]
 #[snafu(visibility(pub))]
@@ -124,6 +124,15 @@ pub enum Error {
        location: Location,
        source: datatypes::error::Error,
    },
+
+    #[snafu(display("{}", msg))]
+    Tonic {
+        code: StatusCode,
+        msg: String,
+        tonic_code: Code,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }

 pub type Result<T> = std::result::Result<T, Error>;
@@ -135,7 +144,7 @@ impl ErrorExt for Error {
            | Error::MissingField { .. }
            | Error::IllegalDatabaseResponse { .. } => StatusCode::Internal,

-            Error::Server { code, .. } => *code,
+            Error::Server { code, .. } | Error::Tonic { code, .. } => *code,
            Error::FlightGet { source, .. }
            | Error::RegionServer { source, .. }
            | Error::FlowServer { source, .. } => source.status_code(),
@@ -153,34 +162,7 @@ impl ErrorExt for Error {
    }
 }

-impl From<Status> for Error {
-    fn from(e: Status) -> Self {
-        fn get_metadata_value(e: &Status, key: &str) -> Option<String> {
-            e.metadata()
-                .get(key)
-                .and_then(|v| String::from_utf8(v.as_bytes().to_vec()).ok())
-        }
-
-        let code = get_metadata_value(&e, GREPTIME_DB_HEADER_ERROR_CODE).and_then(|s| {
-            if let Ok(code) = s.parse::<u32>() {
-                StatusCode::from_u32(code)
-            } else {
-                None
-            }
-        });
-        let tonic_code = e.code();
-        let code = code.unwrap_or_else(|| convert_tonic_code_to_status_code(tonic_code));
-
-        let msg = get_metadata_value(&e, GREPTIME_DB_HEADER_ERROR_MSG)
-            .unwrap_or_else(|| e.message().to_string());
-
-        Self::Server {
-            code,
-            msg,
-            location: location!(),
-        }
-    }
-}
+define_from_tonic_status!(Error, Tonic);

 impl Error {
    pub fn should_retry(&self) -> bool {
--- a/src/client/src/region.rs
+++ b/src/client/src/region.rs
@@ -21,7 +21,7 @@ use arc_swap::ArcSwapOption;
 use arrow_flight::Ticket;
 use async_stream::stream;
 use async_trait::async_trait;
-use common_error::ext::{BoxedError, ErrorExt};
+use common_error::ext::BoxedError;
 use common_error::status_code::StatusCode;
 use common_grpc::flight::{FlightDecoder, FlightMessage};
 use common_meta::error::{self as meta_error, Result as MetaResult};
@@ -107,24 +107,18 @@ impl RegionRequester {
            .mut_inner()
            .do_get(ticket)
            .await
-            .map_err(|e| {
+            .or_else(|e| {
                let tonic_code = e.code();
                let e: error::Error = e.into();
-                let code = e.status_code();
-                let msg = e.to_string();
-                let error = ServerSnafu { code, msg }
-                    .fail::<()>()
-                    .map_err(BoxedError::new)
-                    .with_context(|_| FlightGetSnafu {
-                        tonic_code,
-                        addr: flight_client.addr().to_string(),
-                    })
-                    .unwrap_err();
                error!(
                    e; "Failed to do Flight get, addr: {}, code: {}",
                    flight_client.addr(),
                    tonic_code
                );
+                let error = Err(BoxedError::new(e)).with_context(|_| FlightGetSnafu {
+                    addr: flight_client.addr().to_string(),
+                    tonic_code,
+                });
                error
            })?;

@@ -163,70 +157,19 @@ impl RegionRequester {
            let _span = tracing_context.attach(common_telemetry::tracing::info_span!(
                "poll_flight_data_stream"
            ));
-
-            let mut buffered_message: Option<FlightMessage> = None;
-            let mut stream_ended = false;
-
-            while !stream_ended {
-                // get the next message from the buffered message or read from the flight message stream
-                let flight_message_item = if let Some(msg) = buffered_message.take() {
-                    Some(Ok(msg))
-                } else {
-                    flight_message_stream.next().await
-                };
-
-                let flight_message = match flight_message_item {
-                    Some(Ok(message)) => message,
-                    Some(Err(e)) => {
-                        yield Err(BoxedError::new(e)).context(ExternalSnafu);
-                        break;
-                    }
-                    None => break,
-                };
+            while let Some(flight_message) = flight_message_stream.next().await {
+                let flight_message = flight_message
+                    .map_err(BoxedError::new)
+                    .context(ExternalSnafu)?;

                match flight_message {
                    FlightMessage::RecordBatch(record_batch) => {
-                        let result_to_yield = RecordBatch::try_from_df_record_batch(
+                        yield RecordBatch::try_from_df_record_batch(
                            schema_cloned.clone(),
                            record_batch,
-                        );
-
-                        // get the next message from the stream. normally it should be a metrics message.
-                        if let Some(next_flight_message_result) = flight_message_stream.next().await
-                        {
-                            match next_flight_message_result {
-                                Ok(FlightMessage::Metrics(s)) => {
-                                    let m = serde_json::from_str(&s).ok().map(Arc::new);
-                                    metrics_ref.swap(m);
-                                }
-                                Ok(FlightMessage::RecordBatch(rb)) => {
-                                    // for some reason it's not a metrics message, so we need to buffer this record batch
-                                    // and yield it in the next iteration.
-                                    buffered_message = Some(FlightMessage::RecordBatch(rb));
-                                }
-                                Ok(_) => {
-                                    yield IllegalFlightMessagesSnafu {
-                                        reason: "A RecordBatch message can only be succeeded by a Metrics message or another RecordBatch message"
-                                    }
-                                    .fail()
-                                    .map_err(BoxedError::new)
-                                    .context(ExternalSnafu);
-                                    break;
-                                }
-                                Err(e) => {
-                                    yield Err(BoxedError::new(e)).context(ExternalSnafu);
-                                    break;
-                                }
-                            }
-                        } else {
-                            // the stream has ended
-                            stream_ended = true;
-                        }
-
-                        yield result_to_yield;
+                        )
                    }
                    FlightMessage::Metrics(s) => {
-                        // just a branch in case of some metrics message comes after other things.
                        let m = serde_json::from_str(&s).ok().map(Arc::new);
                        metrics_ref.swap(m);
                        break;
--- a/src/cmd/Cargo.toml
+++ b/src/cmd/Cargo.toml
@@ -16,7 +16,7 @@ default = [
    "meta-srv/pg_kvbackend",
    "meta-srv/mysql_kvbackend",
 ]
-enterprise = ["common-meta/enterprise", "frontend/enterprise", "meta-srv/enterprise"]
+enterprise = ["common-meta/enterprise", "frontend/enterprise", "meta-srv/enterprise", "catalog/enterprise"]
 tokio-console = ["common-telemetry/tokio-console"]

 [lints]
@@ -66,6 +66,7 @@ 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 datanode::config::{FileConfig, GcsConfig, ObjectStoreConfig, S3Config};
+    use object_store::config::{FileConfig, GcsConfig, ObjectStoreConfig, S3Config};
    use servers::heartbeat_options::HeartbeatOptions;

    use super::*;
--- a/src/cmd/src/flownode.rs
+++ b/src/cmd/src/flownode.rs
@@ -18,7 +18,7 @@ use std::time::Duration;

 use cache::{build_fundamental_cache_registry, with_default_composite_cache_registry};
 use catalog::information_extension::DistributedInformationExtension;
-use catalog::kvbackend::{CachedKvBackendBuilder, KvBackendCatalogManager, MetaKvBackend};
+use catalog::kvbackend::{CachedKvBackendBuilder, KvBackendCatalogManagerBuilder, MetaKvBackend};
 use clap::Parser;
 use client::client_manager::NodeClients;
 use common_base::Plugins;
@@ -342,13 +342,12 @@ impl StartCommand {

        let information_extension =
            Arc::new(DistributedInformationExtension::new(meta_client.clone()));
-        let catalog_manager = KvBackendCatalogManager::new(
+        let catalog_manager = KvBackendCatalogManagerBuilder::new(
            information_extension,
            cached_meta_backend.clone(),
            layered_cache_registry.clone(),
-            None,
-            None,
-        );
+        )
+        .build();

        let table_metadata_manager =
            Arc::new(TableMetadataManager::new(cached_meta_backend.clone()));
@@ -371,8 +370,11 @@ impl StartCommand {

        let flow_metadata_manager = Arc::new(FlowMetadataManager::new(cached_meta_backend.clone()));
        let flow_auth_header = get_flow_auth_options(&opts).context(StartFlownodeSnafu)?;
-        let frontend_client =
-            FrontendClient::from_meta_client(meta_client.clone(), flow_auth_header);
+        let frontend_client = FrontendClient::from_meta_client(
+            meta_client.clone(),
+            flow_auth_header,
+            opts.query.clone(),
+        );
        let frontend_client = Arc::new(frontend_client);
        let flownode_builder = FlownodeBuilder::new(
            opts.clone(),
--- a/src/cmd/src/frontend.rs
+++ b/src/cmd/src/frontend.rs
@@ -19,7 +19,7 @@ use std::time::Duration;
 use async_trait::async_trait;
 use cache::{build_fundamental_cache_registry, with_default_composite_cache_registry};
 use catalog::information_extension::DistributedInformationExtension;
-use catalog::kvbackend::{CachedKvBackendBuilder, KvBackendCatalogManager, MetaKvBackend};
+use catalog::kvbackend::{CachedKvBackendBuilder, KvBackendCatalogManagerBuilder, MetaKvBackend};
 use catalog::process_manager::ProcessManager;
 use clap::Parser;
 use client::client_manager::NodeClients;
@@ -102,7 +102,7 @@ impl App for Instance {
 #[derive(Parser)]
 pub struct Command {
    #[clap(subcommand)]
-    subcmd: SubCommand,
+    pub subcmd: SubCommand,
 }

 impl Command {
@@ -116,7 +116,7 @@ impl Command {
 }

 #[derive(Parser)]
-enum SubCommand {
+pub enum SubCommand {
    Start(StartCommand),
 }

@@ -153,7 +153,7 @@ pub struct StartCommand {
    #[clap(long)]
    postgres_addr: Option<String>,
    #[clap(short, long)]
-    config_file: Option<String>,
+    pub config_file: Option<String>,
    #[clap(short, long)]
    influxdb_enable: Option<bool>,
    #[clap(long, value_delimiter = ',', num_args = 1..)]
@@ -169,7 +169,7 @@ pub struct StartCommand {
    #[clap(long)]
    disable_dashboard: Option<bool>,
    #[clap(long, default_value = "GREPTIMEDB_FRONTEND")]
-    env_prefix: String,
+    pub env_prefix: String,
 }

 impl StartCommand {
@@ -350,13 +350,20 @@ impl StartCommand {
            addrs::resolve_addr(&opts.grpc.bind_addr, Some(&opts.grpc.server_addr)),
            Some(meta_client.clone()),
        ));
-        let catalog_manager = KvBackendCatalogManager::new(
+
+        let builder = KvBackendCatalogManagerBuilder::new(
            information_extension,
            cached_meta_backend.clone(),
            layered_cache_registry.clone(),
-            None,
-            Some(process_manager.clone()),
-        );
+        )
+        .with_process_manager(process_manager.clone());
+        #[cfg(feature = "enterprise")]
+        let builder = if let Some(factories) = plugins.get() {
+            builder.with_extra_information_table_factories(factories)
+        } else {
+            builder
+        };
+        let catalog_manager = builder.build();

        let executor = HandlerGroupExecutor::new(vec![
            Arc::new(ParseMailboxMessageHandler),
--- a/src/cmd/src/metasrv.rs
+++ b/src/cmd/src/metasrv.rs
@@ -340,7 +340,7 @@ impl StartCommand {
            .await
            .context(StartMetaServerSnafu)?;

-        let builder = meta_srv::bootstrap::metasrv_builder(&opts, plugins.clone(), None)
+        let builder = meta_srv::bootstrap::metasrv_builder(&opts, plugins, None)
            .await
            .context(error::BuildMetaServerSnafu)?;
        let metasrv = builder.build().await.context(error::BuildMetaServerSnafu)?;
--- a/src/cmd/src/standalone.rs
+++ b/src/cmd/src/standalone.rs
@@ -20,7 +20,7 @@ use std::{fs, path};
 use async_trait::async_trait;
 use cache::{build_fundamental_cache_registry, with_default_composite_cache_registry};
 use catalog::information_schema::InformationExtension;
-use catalog::kvbackend::KvBackendCatalogManager;
+use catalog::kvbackend::KvBackendCatalogManagerBuilder;
 use catalog::process_manager::ProcessManager;
 use clap::Parser;
 use client::api::v1::meta::RegionRole;
@@ -34,14 +34,13 @@ 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::{DdlContext, NoopRegionFailureDetectorControl};
+use common_meta::ddl::{DdlContext, NoopRegionFailureDetectorControl, ProcedureExecutorRef};
 use common_meta::ddl_manager::DdlManager;
 use common_meta::key::flow::flow_state::FlowStat;
 use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
 use common_meta::peer::Peer;
-use common_meta::procedure_executor::LocalProcedureExecutor;
 use common_meta::region_keeper::MemoryRegionKeeper;
 use common_meta::region_registry::LeaderRegionRegistry;
 use common_meta::sequence::SequenceBuilder;
@@ -258,15 +257,34 @@ 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]
@@ -526,13 +544,20 @@ impl StartCommand {
        ));

        let process_manager = Arc::new(ProcessManager::new(opts.grpc.server_addr.clone(), None));
-        let catalog_manager = KvBackendCatalogManager::new(
+        let builder = KvBackendCatalogManagerBuilder::new(
            information_extension.clone(),
            kv_backend.clone(),
            layered_cache_registry.clone(),
-            Some(procedure_manager.clone()),
-            Some(process_manager.clone()),
-        );
+        )
+        .with_procedure_manager(procedure_manager.clone())
+        .with_process_manager(process_manager.clone());
+        #[cfg(feature = "enterprise")]
+        let builder = if let Some(factories) = plugins.get() {
+            builder.with_extra_information_table_factories(factories)
+        } else {
+            builder
+        };
+        let catalog_manager = builder.build();

        let table_metadata_manager =
            Self::create_table_metadata_manager(kv_backend.clone()).await?;
@@ -546,14 +571,15 @@ impl StartCommand {
        // for standalone not use grpc, but get a handler to frontend grpc client without
        // actually make a connection
        let (frontend_client, frontend_instance_handler) =
-            FrontendClient::from_empty_grpc_handler();
+            FrontendClient::from_empty_grpc_handler(opts.query.clone());
+        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(),
-            Arc::new(frontend_client.clone()),
+            frontend_client.clone(),
        );
        let flownode = flow_builder
            .build()
@@ -610,8 +636,9 @@ impl StartCommand {
            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.clone(), true)
+        let ddl_manager = DdlManager::try_new(ddl_context, procedure_manager_c, true)
            .context(error::InitDdlManagerSnafu)?;
        #[cfg(feature = "enterprise")]
        let ddl_manager = {
@@ -619,11 +646,7 @@ impl StartCommand {
                plugins.get();
            ddl_manager.with_trigger_ddl_manager(trigger_ddl_manager)
        };
-
-        let procedure_executor = Arc::new(LocalProcedureExecutor::new(
-            Arc::new(ddl_manager),
-            procedure_manager.clone(),
-        ));
+        let ddl_task_executor: ProcedureExecutorRef = Arc::new(ddl_manager);

        let fe_instance = FrontendBuilder::new(
            fe_opts.clone(),
@@ -631,7 +654,7 @@ impl StartCommand {
            layered_cache_registry.clone(),
            catalog_manager.clone(),
            node_manager.clone(),
-            procedure_executor.clone(),
+            ddl_task_executor.clone(),
            process_manager,
        )
        .with_plugin(plugins.clone())
@@ -656,7 +679,7 @@ impl StartCommand {
            catalog_manager.clone(),
            kv_backend.clone(),
            layered_cache_registry.clone(),
-            procedure_executor,
+            ddl_task_executor.clone(),
            node_manager,
        )
        .await
@@ -666,7 +689,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)
+        let servers = Services::new(opts, fe_instance.clone(), plugins.clone())
            .build()
            .context(error::StartFrontendSnafu)?;

@@ -677,12 +700,22 @@ 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,
        })
    }
@@ -792,7 +825,6 @@ impl InformationExtension for StandaloneInformationExtension {
                    region_manifest: region_stat.manifest.into(),
                    data_topic_latest_entry_id: region_stat.data_topic_latest_entry_id,
                    metadata_topic_latest_entry_id: region_stat.metadata_topic_latest_entry_id,
-                    written_bytes: region_stat.written_bytes,
                }
            })
            .collect::<Vec<_>>();
@@ -823,7 +855,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 datanode::config::{FileConfig, GcsConfig};
+    use object_store::config::{FileConfig, GcsConfig};

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

        assert!(matches!(
            &dn_opts.storage.store,
-            datanode::config::ObjectStoreConfig::File(FileConfig { .. })
+            object_store::config::ObjectStoreConfig::File(FileConfig { .. })
        ));
        assert_eq!(dn_opts.storage.providers.len(), 2);
        assert!(matches!(
            dn_opts.storage.providers[0],
-            datanode::config::ObjectStoreConfig::Gcs(GcsConfig { .. })
+            object_store::config::ObjectStoreConfig::Gcs(GcsConfig { .. })
        ));
        match &dn_opts.storage.providers[1] {
-            datanode::config::ObjectStoreConfig::S3(s3_config) => {
+            object_store::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
@@ -18,17 +18,19 @@ 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_ENDPOINT};
+use common_telemetry::logging::{LoggingOptions, DEFAULT_LOGGING_DIR, DEFAULT_OTLP_HTTP_ENDPOINT};
 use common_wal::config::raft_engine::RaftEngineConfig;
 use common_wal::config::DatanodeWalConfig;
 use datanode::config::{DatanodeOptions, RegionEngineConfig, StorageConfig};
 use file_engine::config::EngineConfig as FileEngineConfig;
+use flow::FlownodeOptions;
 use frontend::frontend::FrontendOptions;
 use meta_client::MetaClientOptions;
 use meta_srv::metasrv::MetasrvOptions;
 use meta_srv::selector::SelectorType;
 use metric_engine::config::EngineConfig as MetricEngineConfig;
 use mito2::config::MitoConfig;
+use query::options::QueryOptions;
 use servers::export_metrics::ExportMetricsOption;
 use servers::grpc::GrpcOptions;
 use servers::http::HttpOptions;
@@ -81,7 +83,7 @@ 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_ENDPOINT.to_string()),
+                otlp_endpoint: Some(DEFAULT_OTLP_HTTP_ENDPOINT.to_string()),
                tracing_sample_ratio: Some(Default::default()),
                ..Default::default()
            },
@@ -124,7 +126,7 @@ 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_ENDPOINT.to_string()),
+                otlp_endpoint: Some(DEFAULT_OTLP_HTTP_ENDPOINT.to_string()),
                tracing_sample_ratio: Some(Default::default()),
                ..Default::default()
            },
@@ -172,7 +174,7 @@ fn test_load_metasrv_example_config() {
            logging: LoggingOptions {
                dir: format!("{}/{}", DEFAULT_DATA_HOME, DEFAULT_LOGGING_DIR),
                level: Some("info".to_string()),
-                otlp_endpoint: Some(DEFAULT_OTLP_ENDPOINT.to_string()),
+                otlp_endpoint: Some(DEFAULT_OTLP_HTTP_ENDPOINT.to_string()),
                tracing_sample_ratio: Some(Default::default()),
                ..Default::default()
            },
@@ -195,6 +197,57 @@ fn test_load_metasrv_example_config() {
    similar_asserts::assert_eq!(options, expected);
 }

+#[test]
+fn test_load_flownode_example_config() {
+    let example_config = common_test_util::find_workspace_path("config/flownode.example.toml");
+    let options =
+        GreptimeOptions::<FlownodeOptions>::load_layered_options(example_config.to_str(), "")
+            .unwrap();
+    let expected = GreptimeOptions::<FlownodeOptions> {
+        component: FlownodeOptions {
+            node_id: Some(14),
+            flow: Default::default(),
+            grpc: GrpcOptions {
+                bind_addr: "127.0.0.1:6800".to_string(),
+                server_addr: "127.0.0.1:6800".to_string(),
+                runtime_size: 2,
+                ..Default::default()
+            },
+            logging: LoggingOptions {
+                dir: format!("{}/{}", DEFAULT_DATA_HOME, DEFAULT_LOGGING_DIR),
+                level: Some("info".to_string()),
+                otlp_endpoint: Some(DEFAULT_OTLP_HTTP_ENDPOINT.to_string()),
+                otlp_export_protocol: Some(common_telemetry::logging::OtlpExportProtocol::Http),
+                tracing_sample_ratio: Some(Default::default()),
+                ..Default::default()
+            },
+            tracing: Default::default(),
+            heartbeat: Default::default(),
+            // flownode deliberately use a slower query parallelism
+            // to avoid overwhelming the frontend with too many queries
+            query: QueryOptions { parallelism: 1 },
+            meta_client: Some(MetaClientOptions {
+                metasrv_addrs: vec!["127.0.0.1:3002".to_string()],
+                timeout: Duration::from_secs(3),
+                heartbeat_timeout: Duration::from_millis(500),
+                ddl_timeout: Duration::from_secs(10),
+                connect_timeout: Duration::from_secs(1),
+                tcp_nodelay: true,
+                metadata_cache_max_capacity: 100000,
+                metadata_cache_ttl: Duration::from_secs(600),
+                metadata_cache_tti: Duration::from_secs(300),
+            }),
+            http: HttpOptions {
+                addr: "127.0.0.1:4000".to_string(),
+                ..Default::default()
+            },
+            user_provider: None,
+        },
+        ..Default::default()
+    };
+    similar_asserts::assert_eq!(options, expected);
+}
+
 #[test]
 fn test_load_standalone_example_config() {
    let example_config = common_test_util::find_workspace_path("config/standalone.example.toml");
@@ -229,7 +282,7 @@ 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_ENDPOINT.to_string()),
+                otlp_endpoint: Some(DEFAULT_OTLP_HTTP_ENDPOINT.to_string()),
                tracing_sample_ratio: Some(Default::default()),
                ..Default::default()
            },
--- a/src/common/catalog/src/consts.rs
+++ b/src/common/catalog/src/consts.rs
@@ -78,7 +78,7 @@ pub const INFORMATION_SCHEMA_ROUTINES_TABLE_ID: u32 = 21;
 pub const INFORMATION_SCHEMA_SCHEMA_PRIVILEGES_TABLE_ID: u32 = 22;
 /// id for information_schema.TABLE_PRIVILEGES
 pub const INFORMATION_SCHEMA_TABLE_PRIVILEGES_TABLE_ID: u32 = 23;
-/// id for information_schema.TRIGGERS
+/// id for information_schema.TRIGGERS (for mysql)
 pub const INFORMATION_SCHEMA_TRIGGERS_TABLE_ID: u32 = 24;
 /// id for information_schema.GLOBAL_STATUS
 pub const INFORMATION_SCHEMA_GLOBAL_STATUS_TABLE_ID: u32 = 25;
@@ -104,6 +104,8 @@ pub const INFORMATION_SCHEMA_PROCEDURE_INFO_TABLE_ID: u32 = 34;
 pub const INFORMATION_SCHEMA_REGION_STATISTICS_TABLE_ID: u32 = 35;
 /// id for information_schema.process_list
 pub const INFORMATION_SCHEMA_PROCESS_LIST_TABLE_ID: u32 = 36;
+/// id for information_schema.trigger_list (for greptimedb trigger)
+pub const INFORMATION_SCHEMA_TRIGGER_TABLE_ID: u32 = 37;

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

--- a/src/common/config/Cargo.toml
+++ b/src/common/config/Cargo.toml
@@ -14,6 +14,7 @@ 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::{ObjectStoreConfig, StorageConfig};
+    use datanode::config::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 {
-                    ObjectStoreConfig::S3(s3_config) => {
+                    object_store::config::ObjectStoreConfig::S3(s3_config) => {
                        assert_eq!(s3_config.bucket, "mybucket".to_string());
                    }
                    _ => panic!("unexpected store type"),
--- a/src/common/error/src/status_code.rs
+++ b/src/common/error/src/status_code.rs
@@ -119,6 +119,11 @@ pub enum StatusCode {
    FlowAlreadyExists = 8000,
    FlowNotFound = 8001,
    // ====== End of flow related status code =====
+
+    // ====== Begin of trigger related status code =====
+    TriggerAlreadyExists = 9000,
+    TriggerNotFound = 9001,
+    // ====== End of trigger related status code =====
 }

 impl StatusCode {
@@ -155,6 +160,8 @@ impl StatusCode {
            | StatusCode::RegionNotFound
            | StatusCode::FlowAlreadyExists
            | StatusCode::FlowNotFound
+            | StatusCode::TriggerAlreadyExists
+            | StatusCode::TriggerNotFound
            | StatusCode::RegionReadonly
            | StatusCode::TableColumnNotFound
            | StatusCode::TableColumnExists
@@ -198,6 +205,8 @@ impl StatusCode {
            | StatusCode::PlanQuery
            | StatusCode::FlowAlreadyExists
            | StatusCode::FlowNotFound
+            | StatusCode::TriggerAlreadyExists
+            | StatusCode::TriggerNotFound
            | StatusCode::RegionNotReady
            | StatusCode::RegionBusy
            | StatusCode::RegionReadonly
@@ -230,6 +239,48 @@ impl fmt::Display for StatusCode {
    }
 }

+#[macro_export]
+macro_rules! define_from_tonic_status {
+    ($Error: ty, $Variant: ident) => {
+        impl From<tonic::Status> for $Error {
+            fn from(e: tonic::Status) -> Self {
+                use snafu::location;
+
+                fn metadata_value(e: &tonic::Status, key: &str) -> Option<String> {
+                    e.metadata()
+                        .get(key)
+                        .and_then(|v| String::from_utf8(v.as_bytes().to_vec()).ok())
+                }
+
+                let code = metadata_value(&e, $crate::GREPTIME_DB_HEADER_ERROR_CODE)
+                    .and_then(|s| {
+                        if let Ok(code) = s.parse::<u32>() {
+                            StatusCode::from_u32(code)
+                        } else {
+                            None
+                        }
+                    })
+                    .unwrap_or_else(|| match e.code() {
+                        tonic::Code::Cancelled => StatusCode::Cancelled,
+                        tonic::Code::DeadlineExceeded => StatusCode::DeadlineExceeded,
+                        _ => StatusCode::Internal,
+                    });
+
+                let msg = metadata_value(&e, $crate::GREPTIME_DB_HEADER_ERROR_MSG)
+                    .unwrap_or_else(|| e.message().to_string());
+
+                // TODO(LFC): Make the error variant defined automatically.
+                Self::$Variant {
+                    code,
+                    msg,
+                    tonic_code: e.code(),
+                    location: location!(),
+                }
+            }
+        }
+    };
+}
+
 #[macro_export]
 macro_rules! define_into_tonic_status {
    ($Error: ty) => {
@@ -281,12 +332,14 @@ pub fn status_to_tonic_code(status_code: StatusCode) -> Code {
        | StatusCode::TableColumnExists
        | StatusCode::RegionAlreadyExists
        | StatusCode::DatabaseAlreadyExists
+        | StatusCode::TriggerAlreadyExists
        | StatusCode::FlowAlreadyExists => Code::AlreadyExists,
        StatusCode::TableNotFound
        | StatusCode::RegionNotFound
        | StatusCode::TableColumnNotFound
        | StatusCode::DatabaseNotFound
        | StatusCode::UserNotFound
+        | StatusCode::TriggerNotFound
        | StatusCode::FlowNotFound => Code::NotFound,
        StatusCode::TableUnavailable
        | StatusCode::StorageUnavailable
@@ -304,15 +357,6 @@ pub fn status_to_tonic_code(status_code: StatusCode) -> Code {
    }
 }

-/// Converts tonic [Code] to [StatusCode].
-pub fn convert_tonic_code_to_status_code(code: Code) -> StatusCode {
-    match code {
-        Code::Cancelled => StatusCode::Cancelled,
-        Code::DeadlineExceeded => StatusCode::DeadlineExceeded,
-        _ => StatusCode::Internal,
-    }
-}
-
 #[cfg(test)]
 mod tests {
    use strum::IntoEnumIterator;
--- a/src/common/frontend/Cargo.toml
+++ b/src/common/frontend/Cargo.toml
@@ -12,7 +12,6 @@ common-macro.workspace = true
 common-meta.workspace = true
 greptime-proto.workspace = true
 meta-client.workspace = true
-session.workspace = true
 snafu.workspace = true
 tonic.workspace = true

--- a/src/common/frontend/src/lib.rs
+++ b/src/common/frontend/src/lib.rs
@@ -19,7 +19,6 @@ use snafu::OptionExt;

 pub mod error;
 pub mod selector;
-pub mod slow_query_event;

 #[derive(Debug, Clone, Eq, PartialEq)]
 pub struct DisplayProcessId {
--- a/src/common/frontend/src/slow_query_event.rs
+++ b/src/common/frontend/src/slow_query_event.rs
@@ -1,28 +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 session::context::QueryContextRef;
-
-#[derive(Debug)]
-pub struct SlowQueryEvent {
-    pub cost: u64,
-    pub threshold: u64,
-    pub query: String,
-    pub is_promql: bool,
-    pub query_ctx: QueryContextRef,
-    pub promql_range: Option<u64>,
-    pub promql_step: Option<u64>,
-    pub promql_start: Option<i64>,
-    pub promql_end: Option<i64>,
-}
--- a/src/common/function/Cargo.toml
+++ b/src/common/function/Cargo.toml
@@ -33,6 +33,7 @@ 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/admin.rs
+++ b/src/common/function/src/admin.rs
@@ -16,9 +16,6 @@ mod add_region_follower;
 mod flush_compact_region;
 mod flush_compact_table;
 mod migrate_region;
-mod reconcile_catalog;
-mod reconcile_database;
-mod reconcile_table;
 mod remove_region_follower;

 use std::sync::Arc;
@@ -27,9 +24,6 @@ use add_region_follower::AddRegionFollowerFunction;
 use flush_compact_region::{CompactRegionFunction, FlushRegionFunction};
 use flush_compact_table::{CompactTableFunction, FlushTableFunction};
 use migrate_region::MigrateRegionFunction;
-use reconcile_catalog::ReconcileCatalogFunction;
-use reconcile_database::ReconcileDatabaseFunction;
-use reconcile_table::ReconcileTableFunction;
 use remove_region_follower::RemoveRegionFollowerFunction;

 use crate::flush_flow::FlushFlowFunction;
@@ -49,8 +43,5 @@ impl AdminFunction {
        registry.register_async(Arc::new(FlushTableFunction));
        registry.register_async(Arc::new(CompactTableFunction));
        registry.register_async(Arc::new(FlushFlowFunction));
-        registry.register_async(Arc::new(ReconcileCatalogFunction));
-        registry.register_async(Arc::new(ReconcileDatabaseFunction));
-        registry.register_async(Arc::new(ReconcileTableFunction));
    }
 }
--- a/src/common/function/src/admin/reconcile_catalog.rs
+++ b/src/common/function/src/admin/reconcile_catalog.rs
@@ -1,179 +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 api::v1::meta::reconcile_request::Target;
-use api::v1::meta::{ReconcileCatalog, ReconcileRequest};
-use common_macro::admin_fn;
-use common_query::error::{
-    InvalidFuncArgsSnafu, MissingProcedureServiceHandlerSnafu, Result,
-    UnsupportedInputDataTypeSnafu,
-};
-use common_query::prelude::{Signature, TypeSignature, Volatility};
-use common_telemetry::info;
-use datatypes::prelude::*;
-use session::context::QueryContextRef;
-
-use crate::handlers::ProcedureServiceHandlerRef;
-use crate::helper::{
-    cast_u32, default_parallelism, default_resolve_strategy, get_string_from_params,
-    parse_resolve_strategy,
-};
-
-const FN_NAME: &str = "reconcile_catalog";
-
-/// A function to reconcile a catalog.
-/// Returns the procedure id if success.
-///
-/// - `reconcile_catalog(resolve_strategy)`.
-/// - `reconcile_catalog(resolve_strategy, parallelism)`.
-///
-/// - `reconcile_catalog()`.
-#[admin_fn(
-    name = ReconcileCatalogFunction,
-    display_name = reconcile_catalog,
-    sig_fn = signature,
-    ret = string
-)]
-pub(crate) async fn reconcile_catalog(
-    procedure_service_handler: &ProcedureServiceHandlerRef,
-    query_ctx: &QueryContextRef,
-    params: &[ValueRef<'_>],
-) -> Result<Value> {
-    let (resolve_strategy, parallelism) = match params.len() {
-        0 => (default_resolve_strategy(), default_parallelism()),
-        1 => (
-            parse_resolve_strategy(get_string_from_params(params, 0, FN_NAME)?)?,
-            default_parallelism(),
-        ),
-        2 => {
-            let Some(parallelism) = cast_u32(&params[1])? else {
-                return UnsupportedInputDataTypeSnafu {
-                    function: FN_NAME,
-                    datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
-                }
-                .fail();
-            };
-            (
-                parse_resolve_strategy(get_string_from_params(params, 0, FN_NAME)?)?,
-                parallelism,
-            )
-        }
-        size => {
-            return InvalidFuncArgsSnafu {
-                err_msg: format!(
-                    "The length of the args is not correct, expect 0, 1 or 2, have: {}",
-                    size
-                ),
-            }
-            .fail();
-        }
-    };
-    info!(
-        "Reconciling catalog with resolve_strategy: {:?}, parallelism: {}",
-        resolve_strategy, parallelism
-    );
-    let pid = procedure_service_handler
-        .reconcile(ReconcileRequest {
-            target: Some(Target::ReconcileCatalog(ReconcileCatalog {
-                catalog_name: query_ctx.current_catalog().to_string(),
-                parallelism,
-                resolve_strategy: resolve_strategy as i32,
-            })),
-            ..Default::default()
-        })
-        .await?;
-    match pid {
-        Some(pid) => Ok(Value::from(pid)),
-        None => Ok(Value::Null),
-    }
-}
-
-fn signature() -> Signature {
-    let nums = ConcreteDataType::numerics();
-    let mut signs = Vec::with_capacity(2 + nums.len());
-    signs.extend([
-        // reconcile_catalog()
-        TypeSignature::NullAry,
-        // reconcile_catalog(resolve_strategy)
-        TypeSignature::Exact(vec![ConcreteDataType::string_datatype()]),
-    ]);
-    for sign in nums {
-        // reconcile_catalog(resolve_strategy, parallelism)
-        signs.push(TypeSignature::Exact(vec![
-            ConcreteDataType::string_datatype(),
-            sign,
-        ]));
-    }
-    Signature::one_of(signs, Volatility::Immutable)
-}
-
-#[cfg(test)]
-mod tests {
-    use std::assert_matches::assert_matches;
-    use std::sync::Arc;
-
-    use common_query::error::Error;
-    use datatypes::vectors::{StringVector, UInt64Vector, VectorRef};
-
-    use crate::admin::reconcile_catalog::ReconcileCatalogFunction;
-    use crate::function::{AsyncFunction, FunctionContext};
-
-    #[tokio::test]
-    async fn test_reconcile_catalog() {
-        common_telemetry::init_default_ut_logging();
-
-        // reconcile_catalog()
-        let f = ReconcileCatalogFunction;
-        let args = vec![];
-        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
-        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
-        assert_eq!(expect, result);
-
-        // reconcile_catalog(resolve_strategy)
-        let f = ReconcileCatalogFunction;
-        let args = vec![Arc::new(StringVector::from(vec!["UseMetasrv"])) as _];
-        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
-        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
-        assert_eq!(expect, result);
-
-        // reconcile_catalog(resolve_strategy, parallelism)
-        let f = ReconcileCatalogFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
-            Arc::new(UInt64Vector::from_slice([10])) as _,
-        ];
-        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
-        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
-        assert_eq!(expect, result);
-
-        // unsupported input data type
-        let f = ReconcileCatalogFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
-            Arc::new(StringVector::from(vec!["test"])) as _,
-        ];
-        let err = f.eval(FunctionContext::mock(), &args).await.unwrap_err();
-        assert_matches!(err, Error::UnsupportedInputDataType { .. });
-
-        // invalid function args
-        let f = ReconcileCatalogFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
-            Arc::new(UInt64Vector::from_slice([10])) as _,
-            Arc::new(StringVector::from(vec!["10"])) as _,
-        ];
-        let err = f.eval(FunctionContext::mock(), &args).await.unwrap_err();
-        assert_matches!(err, Error::InvalidFuncArgs { .. });
-    }
-}
--- a/src/common/function/src/admin/reconcile_database.rs
+++ b/src/common/function/src/admin/reconcile_database.rs
@@ -1,198 +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 api::v1::meta::reconcile_request::Target;
-use api::v1::meta::{ReconcileDatabase, ReconcileRequest};
-use common_macro::admin_fn;
-use common_query::error::{
-    InvalidFuncArgsSnafu, MissingProcedureServiceHandlerSnafu, Result,
-    UnsupportedInputDataTypeSnafu,
-};
-use common_query::prelude::{Signature, TypeSignature, Volatility};
-use common_telemetry::info;
-use datatypes::prelude::*;
-use session::context::QueryContextRef;
-
-use crate::handlers::ProcedureServiceHandlerRef;
-use crate::helper::{
-    cast_u32, default_parallelism, default_resolve_strategy, get_string_from_params,
-    parse_resolve_strategy,
-};
-
-const FN_NAME: &str = "reconcile_database";
-
-/// A function to reconcile a database.
-/// Returns the procedure id if success.
-///
-/// - `reconcile_database(database_name)`.
-/// - `reconcile_database(database_name, resolve_strategy)`.
-/// - `reconcile_database(database_name, resolve_strategy, parallelism)`.
-///
-/// The parameters:
-/// - `database_name`:  the database name
-#[admin_fn(
-    name = ReconcileDatabaseFunction,
-    display_name = reconcile_database,
-    sig_fn = signature,
-    ret = string
-)]
-pub(crate) async fn reconcile_database(
-    procedure_service_handler: &ProcedureServiceHandlerRef,
-    query_ctx: &QueryContextRef,
-    params: &[ValueRef<'_>],
-) -> Result<Value> {
-    let (database_name, resolve_strategy, parallelism) = match params.len() {
-        1 => (
-            get_string_from_params(params, 0, FN_NAME)?,
-            default_resolve_strategy(),
-            default_parallelism(),
-        ),
-        2 => (
-            get_string_from_params(params, 0, FN_NAME)?,
-            parse_resolve_strategy(get_string_from_params(params, 1, FN_NAME)?)?,
-            default_parallelism(),
-        ),
-        3 => {
-            let Some(parallelism) = cast_u32(&params[2])? else {
-                return UnsupportedInputDataTypeSnafu {
-                    function: FN_NAME,
-                    datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
-                }
-                .fail();
-            };
-            (
-                get_string_from_params(params, 0, FN_NAME)?,
-                parse_resolve_strategy(get_string_from_params(params, 1, FN_NAME)?)?,
-                parallelism,
-            )
-        }
-        size => {
-            return InvalidFuncArgsSnafu {
-                err_msg: format!(
-                    "The length of the args is not correct, expect 1, 2 or 3, have: {}",
-                    size
-                ),
-            }
-            .fail();
-        }
-    };
-    info!(
-        "Reconciling database: {}, resolve_strategy: {:?}, parallelism: {}",
-        database_name, resolve_strategy, parallelism
-    );
-    let pid = procedure_service_handler
-        .reconcile(ReconcileRequest {
-            target: Some(Target::ReconcileDatabase(ReconcileDatabase {
-                catalog_name: query_ctx.current_catalog().to_string(),
-                database_name: database_name.to_string(),
-                parallelism,
-                resolve_strategy: resolve_strategy as i32,
-            })),
-            ..Default::default()
-        })
-        .await?;
-    match pid {
-        Some(pid) => Ok(Value::from(pid)),
-        None => Ok(Value::Null),
-    }
-}
-
-fn signature() -> Signature {
-    let nums = ConcreteDataType::numerics();
-    let mut signs = Vec::with_capacity(2 + nums.len());
-    signs.extend([
-        // reconcile_database(datanode_name)
-        TypeSignature::Exact(vec![ConcreteDataType::string_datatype()]),
-        // reconcile_database(database_name, resolve_strategy)
-        TypeSignature::Exact(vec![
-            ConcreteDataType::string_datatype(),
-            ConcreteDataType::string_datatype(),
-        ]),
-    ]);
-    for sign in nums {
-        // reconcile_database(database_name, resolve_strategy, parallelism)
-        signs.push(TypeSignature::Exact(vec![
-            ConcreteDataType::string_datatype(),
-            ConcreteDataType::string_datatype(),
-            sign,
-        ]));
-    }
-    Signature::one_of(signs, Volatility::Immutable)
-}
-
-#[cfg(test)]
-mod tests {
-    use std::assert_matches::assert_matches;
-    use std::sync::Arc;
-
-    use common_query::error::Error;
-    use datatypes::vectors::{StringVector, UInt32Vector, VectorRef};
-
-    use crate::admin::reconcile_database::ReconcileDatabaseFunction;
-    use crate::function::{AsyncFunction, FunctionContext};
-
-    #[tokio::test]
-    async fn test_reconcile_catalog() {
-        common_telemetry::init_default_ut_logging();
-
-        // reconcile_database(database_name)
-        let f = ReconcileDatabaseFunction;
-        let args = vec![Arc::new(StringVector::from(vec!["test"])) as _];
-        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
-        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
-        assert_eq!(expect, result);
-
-        // reconcile_database(database_name, resolve_strategy)
-        let f = ReconcileDatabaseFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["test"])) as _,
-            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
-        ];
-        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
-        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
-        assert_eq!(expect, result);
-
-        // reconcile_database(database_name, resolve_strategy, parallelism)
-        let f = ReconcileDatabaseFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["test"])) as _,
-            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
-            Arc::new(UInt32Vector::from_slice([10])) as _,
-        ];
-        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
-        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
-        assert_eq!(expect, result);
-
-        // invalid function args
-        let f = ReconcileDatabaseFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
-            Arc::new(UInt32Vector::from_slice([10])) as _,
-            Arc::new(StringVector::from(vec!["v1"])) as _,
-            Arc::new(StringVector::from(vec!["v2"])) as _,
-        ];
-        let err = f.eval(FunctionContext::mock(), &args).await.unwrap_err();
-        assert_matches!(err, Error::InvalidFuncArgs { .. });
-
-        // unsupported input data type
-        let f = ReconcileDatabaseFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
-            Arc::new(UInt32Vector::from_slice([10])) as _,
-            Arc::new(StringVector::from(vec!["v1"])) as _,
-        ];
-        let err = f.eval(FunctionContext::mock(), &args).await.unwrap_err();
-        assert_matches!(err, Error::UnsupportedInputDataType { .. });
-    }
-}
--- a/src/common/function/src/admin/reconcile_table.rs
+++ b/src/common/function/src/admin/reconcile_table.rs
@@ -1,149 +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 api::v1::meta::reconcile_request::Target;
-use api::v1::meta::{ReconcileRequest, ReconcileTable, ResolveStrategy};
-use common_catalog::format_full_table_name;
-use common_error::ext::BoxedError;
-use common_macro::admin_fn;
-use common_query::error::{
-    MissingProcedureServiceHandlerSnafu, Result, TableMutationSnafu, UnsupportedInputDataTypeSnafu,
-};
-use common_query::prelude::{Signature, TypeSignature, Volatility};
-use common_telemetry::info;
-use datatypes::prelude::*;
-use session::context::QueryContextRef;
-use session::table_name::table_name_to_full_name;
-use snafu::ResultExt;
-
-use crate::handlers::ProcedureServiceHandlerRef;
-use crate::helper::parse_resolve_strategy;
-
-const FN_NAME: &str = "reconcile_table";
-
-/// A function to reconcile a table.
-/// Returns the procedure id if success.
-///
-/// - `reconcile_table(table_name)`.
-/// - `reconcile_table(table_name, resolve_strategy)`.
-///
-/// The parameters:
-/// - `table_name`:  the table name
-#[admin_fn(
-    name = ReconcileTableFunction,
-    display_name = reconcile_table,
-    sig_fn = signature,
-    ret = string
-)]
-pub(crate) async fn reconcile_table(
-    procedure_service_handler: &ProcedureServiceHandlerRef,
-    query_ctx: &QueryContextRef,
-    params: &[ValueRef<'_>],
-) -> Result<Value> {
-    let (table_name, resolve_strategy) = match params {
-        [ValueRef::String(table_name)] => (table_name, ResolveStrategy::UseLatest),
-        [ValueRef::String(table_name), ValueRef::String(resolve_strategy)] => {
-            (table_name, parse_resolve_strategy(resolve_strategy)?)
-        }
-        _ => {
-            return UnsupportedInputDataTypeSnafu {
-                function: FN_NAME,
-                datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
-            }
-            .fail()
-        }
-    };
-    let (catalog_name, schema_name, table_name) = table_name_to_full_name(table_name, query_ctx)
-        .map_err(BoxedError::new)
-        .context(TableMutationSnafu)?;
-    info!(
-        "Reconciling table: {} with resolve_strategy: {:?}",
-        format_full_table_name(&catalog_name, &schema_name, &table_name),
-        resolve_strategy
-    );
-    let pid = procedure_service_handler
-        .reconcile(ReconcileRequest {
-            target: Some(Target::ReconcileTable(ReconcileTable {
-                catalog_name,
-                schema_name,
-                table_name,
-                resolve_strategy: resolve_strategy as i32,
-            })),
-            ..Default::default()
-        })
-        .await?;
-    match pid {
-        Some(pid) => Ok(Value::from(pid)),
-        None => Ok(Value::Null),
-    }
-}
-
-fn signature() -> Signature {
-    Signature::one_of(
-        vec![
-            // reconcile_table(table_name)
-            TypeSignature::Exact(vec![ConcreteDataType::string_datatype()]),
-            // reconcile_table(table_name, resolve_strategy)
-            TypeSignature::Exact(vec![
-                ConcreteDataType::string_datatype(),
-                ConcreteDataType::string_datatype(),
-            ]),
-        ],
-        Volatility::Immutable,
-    )
-}
-
-#[cfg(test)]
-mod tests {
-    use std::assert_matches::assert_matches;
-    use std::sync::Arc;
-
-    use common_query::error::Error;
-    use datatypes::vectors::{StringVector, VectorRef};
-
-    use crate::admin::reconcile_table::ReconcileTableFunction;
-    use crate::function::{AsyncFunction, FunctionContext};
-
-    #[tokio::test]
-    async fn test_reconcile_table() {
-        common_telemetry::init_default_ut_logging();
-
-        // reconcile_table(table_name)
-        let f = ReconcileTableFunction;
-        let args = vec![Arc::new(StringVector::from(vec!["test"])) as _];
-        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
-        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
-        assert_eq!(expect, result);
-
-        // reconcile_table(table_name, resolve_strategy)
-        let f = ReconcileTableFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["test"])) as _,
-            Arc::new(StringVector::from(vec!["UseMetasrv"])) as _,
-        ];
-        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
-        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
-        assert_eq!(expect, result);
-
-        // unsupported input data type
-        let f = ReconcileTableFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["test"])) as _,
-            Arc::new(StringVector::from(vec!["UseMetasrv"])) as _,
-            Arc::new(StringVector::from(vec!["10"])) as _,
-        ];
-        let err = f.eval(FunctionContext::mock(), &args).await.unwrap_err();
-        assert_matches!(err, Error::UnsupportedInputDataType { .. });
-    }
-}
--- a/src/common/function/src/aggrs.rs
+++ b/src/common/function/src/aggrs.rs
@@ -13,6 +13,7 @@
 // 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/count_hash.rs
+++ b/src/common/function/src/aggrs/count_hash.rs
@@ -0,0 +1,647 @@
+// 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,6 +21,7 @@ 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;
@@ -144,6 +145,9 @@ 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/handlers.rs
+++ b/src/common/function/src/handlers.rs
@@ -14,7 +14,6 @@

 use std::sync::Arc;

-use api::v1::meta::ReconcileRequest;
 use async_trait::async_trait;
 use catalog::CatalogManagerRef;
 use common_base::AffectedRows;
@@ -66,9 +65,6 @@ pub trait ProcedureServiceHandler: Send + Sync {
    /// Migrate a region from source peer to target peer, returns the procedure id if success.
    async fn migrate_region(&self, request: MigrateRegionRequest) -> Result<Option<String>>;

-    /// Reconcile a table, database or catalog, returns the procedure id if success.
-    async fn reconcile(&self, request: ReconcileRequest) -> Result<Option<String>>;
-
    /// Query the procedure' state by its id
    async fn query_procedure_state(&self, pid: &str) -> Result<ProcedureStateResponse>;

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

-use api::v1::meta::ResolveStrategy;
-use common_query::error::{
-    InvalidFuncArgsSnafu, InvalidInputTypeSnafu, Result, UnsupportedInputDataTypeSnafu,
-};
+use common_query::error::{InvalidInputTypeSnafu, Result};
 use common_query::prelude::{Signature, TypeSignature, Volatility};
 use datatypes::prelude::ConcreteDataType;
 use datatypes::types::cast::cast;
 use datatypes::value::ValueRef;
-use snafu::{OptionExt, ResultExt};
+use snafu::ResultExt;

 /// Create a function signature with oneof signatures of interleaving two arguments.
 pub fn one_of_sigs2(args1: Vec<ConcreteDataType>, args2: Vec<ConcreteDataType>) -> Signature {
@@ -46,64 +43,3 @@ pub fn cast_u64(value: &ValueRef) -> Result<Option<u64>> {
        })
        .map(|v| v.as_u64())
 }
-
-/// Cast a [`ValueRef`] to u32, returns `None` if fails
-pub fn cast_u32(value: &ValueRef) -> Result<Option<u32>> {
-    cast((*value).into(), &ConcreteDataType::uint32_datatype())
-        .context(InvalidInputTypeSnafu {
-            err_msg: format!(
-                "Failed to cast input into uint32, actual type: {:#?}",
-                value.data_type(),
-            ),
-        })
-        .map(|v| v.as_u64().map(|v| v as u32))
-}
-
-/// Parse a resolve strategy from a string.
-pub fn parse_resolve_strategy(strategy: &str) -> Result<ResolveStrategy> {
-    ResolveStrategy::from_str_name(strategy).context(InvalidFuncArgsSnafu {
-        err_msg: format!("Invalid resolve strategy: {}", strategy),
-    })
-}
-
-/// Default parallelism for reconcile operations.
-pub fn default_parallelism() -> u32 {
-    64
-}
-
-/// Default resolve strategy for reconcile operations.
-pub fn default_resolve_strategy() -> ResolveStrategy {
-    ResolveStrategy::UseLatest
-}
-
-/// Get the string value from the params.
-///
-/// # Errors
-/// Returns an error if the input type is not a string.
-pub fn get_string_from_params<'a>(
-    params: &'a [ValueRef<'a>],
-    index: usize,
-    fn_name: &'a str,
-) -> Result<&'a str> {
-    let ValueRef::String(s) = &params[index] else {
-        return UnsupportedInputDataTypeSnafu {
-            function: fn_name,
-            datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
-        }
-        .fail();
-    };
-    Ok(s)
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_parse_resolve_strategy() {
-        assert_eq!(
-            parse_resolve_strategy("UseLatest").unwrap(),
-            ResolveStrategy::UseLatest
-        );
-    }
-}
--- a/src/common/function/src/lib.rs
+++ b/src/common/function/src/lib.rs
@@ -14,7 +14,6 @@

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

 mod admin;
 mod flush_flow;
--- a/src/common/function/src/scalars/math.rs
+++ b/src/common/function/src/scalars/math.rs
@@ -14,7 +14,6 @@

 pub mod clamp;
 mod modulo;
-mod pow;
 mod rate;

 use std::fmt;
@@ -26,7 +25,6 @@ use datafusion::error::DataFusionError;
 use datafusion::logical_expr::Volatility;
 use datatypes::prelude::ConcreteDataType;
 use datatypes::vectors::VectorRef;
-pub use pow::PowFunction;
 pub use rate::RateFunction;
 use snafu::ResultExt;

@@ -39,7 +37,6 @@ pub(crate) struct MathFunction;
 impl MathFunction {
    pub fn register(registry: &FunctionRegistry) {
        registry.register_scalar(ModuloFunction);
-        registry.register_scalar(PowFunction);
        registry.register_scalar(RateFunction);
        registry.register_scalar(RangeFunction);
        registry.register_scalar(ClampFunction);
--- a/src/common/function/src/scalars/math/pow.rs
+++ b/src/common/function/src/scalars/math/pow.rs
@@ -1,120 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::fmt;
-use std::sync::Arc;
-
-use common_query::error::Result;
-use common_query::prelude::{Signature, Volatility};
-use datatypes::data_type::DataType;
-use datatypes::prelude::ConcreteDataType;
-use datatypes::types::LogicalPrimitiveType;
-use datatypes::vectors::VectorRef;
-use datatypes::with_match_primitive_type_id;
-use num::traits::Pow;
-use num_traits::AsPrimitive;
-
-use crate::function::{Function, FunctionContext};
-use crate::scalars::expression::{scalar_binary_op, EvalContext};
-
-#[derive(Clone, Debug, Default)]
-pub struct PowFunction;
-
-impl Function for PowFunction {
-    fn name(&self) -> &str {
-        "pow"
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::float64_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        Signature::uniform(2, ConcreteDataType::numerics(), Volatility::Immutable)
-    }
-
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        with_match_primitive_type_id!(columns[0].data_type().logical_type_id(), |$S| {
-            with_match_primitive_type_id!(columns[1].data_type().logical_type_id(), |$T| {
-                let col = scalar_binary_op::<<$S as LogicalPrimitiveType>::Native, <$T as LogicalPrimitiveType>::Native, f64, _>(&columns[0], &columns[1], scalar_pow, &mut EvalContext::default())?;
-                Ok(Arc::new(col))
-            },{
-                unreachable!()
-            })
-        },{
-            unreachable!()
-        })
-    }
-}
-
-#[inline]
-fn scalar_pow<S, T>(value: Option<S>, base: Option<T>, _ctx: &mut EvalContext) -> Option<f64>
-where
-    S: AsPrimitive<f64>,
-    T: AsPrimitive<f64>,
-{
-    match (value, base) {
-        (Some(value), Some(base)) => Some(value.as_().pow(base.as_())),
-        _ => None,
-    }
-}
-
-impl fmt::Display for PowFunction {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "POW")
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use common_query::prelude::TypeSignature;
-    use datatypes::value::Value;
-    use datatypes::vectors::{Float32Vector, Int8Vector};
-
-    use super::*;
-    use crate::function::FunctionContext;
-    #[test]
-    fn test_pow_function() {
-        let pow = PowFunction;
-
-        assert_eq!("pow", pow.name());
-        assert_eq!(
-            ConcreteDataType::float64_datatype(),
-            pow.return_type(&[]).unwrap()
-        );
-
-        assert!(matches!(pow.signature(),
-                         Signature {
-                             type_signature: TypeSignature::Uniform(2, valid_types),
-                             volatility: Volatility::Immutable
-                         } if  valid_types == ConcreteDataType::numerics()
-        ));
-
-        let values = vec![1.0, 2.0, 3.0];
-        let bases = vec![0i8, -1i8, 3i8];
-
-        let args: Vec<VectorRef> = vec![
-            Arc::new(Float32Vector::from_vec(values.clone())),
-            Arc::new(Int8Vector::from_vec(bases.clone())),
-        ];
-
-        let vector = pow.eval(&FunctionContext::default(), &args).unwrap();
-        assert_eq!(3, vector.len());
-
-        for i in 0..3 {
-            let p: f64 = (values[i] as f64).pow(bases[i] as f64);
-            assert!(matches!(vector.get(i), Value::Float64(v) if v == p));
-        }
-    }
-}
--- a/src/common/function/src/state.rs
+++ b/src/common/function/src/state.rs
@@ -32,7 +32,7 @@ impl FunctionState {
    pub fn mock() -> Self {
        use std::sync::Arc;

-        use api::v1::meta::{ProcedureStatus, ReconcileRequest};
+        use api::v1::meta::ProcedureStatus;
        use async_trait::async_trait;
        use catalog::CatalogManagerRef;
        use common_base::AffectedRows;
@@ -63,10 +63,6 @@ impl FunctionState {
                Ok(Some("test_pid".to_string()))
            }

-            async fn reconcile(&self, _request: ReconcileRequest) -> Result<Option<String>> {
-                Ok(Some("test_pid".to_string()))
-            }
-
            async fn query_procedure_state(&self, _pid: &str) -> Result<ProcedureStateResponse> {
                Ok(ProcedureStateResponse {
                    status: ProcedureStatus::Done.into(),
--- a/src/common/grpc-expr/src/alter.rs
+++ b/src/common/grpc-expr/src/alter.rs
@@ -29,8 +29,8 @@ use snafu::{ensure, OptionExt, ResultExt};
 use store_api::region_request::{SetRegionOption, UnsetRegionOption};
 use table::metadata::TableId;
 use table::requests::{
-    AddColumnRequest, AlterKind, AlterTableRequest, ModifyColumnTypeRequest, SetIndexOption,
-    UnsetIndexOption,
+    AddColumnRequest, AlterKind, AlterTableRequest, ModifyColumnTypeRequest, SetIndexOptions,
+    UnsetIndexOptions,
 };

 use crate::error::{
@@ -43,59 +43,6 @@ use crate::error::{
 const LOCATION_TYPE_FIRST: i32 = LocationType::First as i32;
 const LOCATION_TYPE_AFTER: i32 = LocationType::After as i32;

-fn set_index_option_from_proto(set_index: api::v1::SetIndex) -> Result<SetIndexOption> {
-    let options = set_index.options.context(MissingAlterIndexOptionSnafu)?;
-    Ok(match options {
-        api::v1::set_index::Options::Fulltext(f) => SetIndexOption::Fulltext {
-            column_name: f.column_name.clone(),
-            options: FulltextOptions::new(
-                f.enable,
-                as_fulltext_option_analyzer(
-                    Analyzer::try_from(f.analyzer).context(InvalidSetFulltextOptionRequestSnafu)?,
-                ),
-                f.case_sensitive,
-                as_fulltext_option_backend(
-                    PbFulltextBackend::try_from(f.backend)
-                        .context(InvalidSetFulltextOptionRequestSnafu)?,
-                ),
-                f.granularity as u32,
-                f.false_positive_rate,
-            )
-            .context(InvalidIndexOptionSnafu)?,
-        },
-        api::v1::set_index::Options::Inverted(i) => SetIndexOption::Inverted {
-            column_name: i.column_name,
-        },
-        api::v1::set_index::Options::Skipping(s) => SetIndexOption::Skipping {
-            column_name: s.column_name,
-            options: SkippingIndexOptions::new(
-                s.granularity as u32,
-                s.false_positive_rate,
-                as_skipping_index_type(
-                    PbSkippingIndexType::try_from(s.skipping_index_type)
-                        .context(InvalidSetSkippingIndexOptionRequestSnafu)?,
-                ),
-            )
-            .context(InvalidIndexOptionSnafu)?,
-        },
-    })
-}
-
-fn unset_index_option_from_proto(unset_index: api::v1::UnsetIndex) -> Result<UnsetIndexOption> {
-    let options = unset_index.options.context(MissingAlterIndexOptionSnafu)?;
-    Ok(match options {
-        api::v1::unset_index::Options::Fulltext(f) => UnsetIndexOption::Fulltext {
-            column_name: f.column_name,
-        },
-        api::v1::unset_index::Options::Inverted(i) => UnsetIndexOption::Inverted {
-            column_name: i.column_name,
-        },
-        api::v1::unset_index::Options::Skipping(s) => UnsetIndexOption::Skipping {
-            column_name: s.column_name,
-        },
-    })
-}
-
 /// Convert an [`AlterTableExpr`] to an [`AlterTableRequest`]
 pub fn alter_expr_to_request(table_id: TableId, expr: AlterTableExpr) -> Result<AlterTableRequest> {
    let catalog_name = expr.catalog_name;
@@ -174,34 +121,70 @@ pub fn alter_expr_to_request(table_id: TableId, expr: AlterTableExpr) -> Result<
                    .context(InvalidUnsetTableOptionRequestSnafu)?,
            }
        }
-        Kind::SetIndex(o) => {
-            let option = set_index_option_from_proto(o)?;
-            AlterKind::SetIndexes {
-                options: vec![option],
-            }
-        }
-        Kind::UnsetIndex(o) => {
-            let option = unset_index_option_from_proto(o)?;
-            AlterKind::UnsetIndexes {
-                options: vec![option],
-            }
-        }
-        Kind::SetIndexes(o) => {
-            let options = o
-                .set_indexes
-                .into_iter()
-                .map(set_index_option_from_proto)
-                .collect::<Result<Vec<_>>>()?;
-            AlterKind::SetIndexes { options }
-        }
-        Kind::UnsetIndexes(o) => {
-            let options = o
-                .unset_indexes
-                .into_iter()
-                .map(unset_index_option_from_proto)
-                .collect::<Result<Vec<_>>>()?;
-            AlterKind::UnsetIndexes { options }
-        }
+        Kind::SetIndex(o) => match o.options {
+            Some(opt) => match opt {
+                api::v1::set_index::Options::Fulltext(f) => AlterKind::SetIndex {
+                    options: SetIndexOptions::Fulltext {
+                        column_name: f.column_name.clone(),
+                        options: FulltextOptions::new(
+                            f.enable,
+                            as_fulltext_option_analyzer(
+                                Analyzer::try_from(f.analyzer)
+                                    .context(InvalidSetFulltextOptionRequestSnafu)?,
+                            ),
+                            f.case_sensitive,
+                            as_fulltext_option_backend(
+                                PbFulltextBackend::try_from(f.backend)
+                                    .context(InvalidSetFulltextOptionRequestSnafu)?,
+                            ),
+                            f.granularity as u32,
+                            f.false_positive_rate,
+                        )
+                        .context(InvalidIndexOptionSnafu)?,
+                    },
+                },
+                api::v1::set_index::Options::Inverted(i) => AlterKind::SetIndex {
+                    options: SetIndexOptions::Inverted {
+                        column_name: i.column_name,
+                    },
+                },
+                api::v1::set_index::Options::Skipping(s) => AlterKind::SetIndex {
+                    options: SetIndexOptions::Skipping {
+                        column_name: s.column_name,
+                        options: SkippingIndexOptions::new(
+                            s.granularity as u32,
+                            s.false_positive_rate,
+                            as_skipping_index_type(
+                                PbSkippingIndexType::try_from(s.skipping_index_type)
+                                    .context(InvalidSetSkippingIndexOptionRequestSnafu)?,
+                            ),
+                        )
+                        .context(InvalidIndexOptionSnafu)?,
+                    },
+                },
+            },
+            None => return MissingAlterIndexOptionSnafu.fail(),
+        },
+        Kind::UnsetIndex(o) => match o.options {
+            Some(opt) => match opt {
+                api::v1::unset_index::Options::Fulltext(f) => AlterKind::UnsetIndex {
+                    options: UnsetIndexOptions::Fulltext {
+                        column_name: f.column_name,
+                    },
+                },
+                api::v1::unset_index::Options::Inverted(i) => AlterKind::UnsetIndex {
+                    options: UnsetIndexOptions::Inverted {
+                        column_name: i.column_name,
+                    },
+                },
+                api::v1::unset_index::Options::Skipping(s) => AlterKind::UnsetIndex {
+                    options: UnsetIndexOptions::Skipping {
+                        column_name: s.column_name,
+                    },
+                },
+            },
+            None => return MissingAlterIndexOptionSnafu.fail(),
+        },
        Kind::DropDefaults(o) => {
            let names = o
                .drop_defaults
--- a/src/common/meta/Cargo.toml
+++ b/src/common/meta/Cargo.toml
@@ -32,7 +32,6 @@ common-procedure.workspace = true
 common-procedure-test.workspace = true
 common-query.workspace = true
 common-recordbatch.workspace = true
-common-runtime.workspace = true
 common-telemetry.workspace = true
 common-time.workspace = true
 common-wal.workspace = true
--- a/src/common/meta/src/cache/flow/table_flownode.rs
+++ b/src/common/meta/src/cache/flow/table_flownode.rs
@@ -15,7 +15,6 @@
 use std::collections::HashMap;
 use std::sync::Arc;

-use common_telemetry::info;
 use futures::future::BoxFuture;
 use moka::future::Cache;
 use moka::ops::compute::Op;
@@ -90,12 +89,6 @@ fn init_factory(table_flow_manager: TableFlowManagerRef) -> Initializer<TableId,
                // we have a corresponding cache invalidation mechanism to invalidate `(Key, EmptyHashSet)`.
                .map(Arc::new)
                .map(Some)
-                .inspect(|set| {
-                    info!(
-                        "Initialized table_flownode cache for table_id: {}, set: {:?}",
-                        table_id, set
-                    );
-                })
        })
    })
 }
@@ -174,13 +167,6 @@ fn invalidator<'a>(
        match ident {
            CacheIdent::CreateFlow(create_flow) => handle_create_flow(cache, create_flow).await,
            CacheIdent::DropFlow(drop_flow) => handle_drop_flow(cache, drop_flow).await,
-            CacheIdent::FlowNodeAddressChange(node_id) => {
-                info!(
-                    "Invalidate flow node cache for node_id in table_flownode: {}",
-                    node_id
-                );
-                cache.invalidate_all();
-            }
            _ => {}
        }
        Ok(())
@@ -188,10 +174,7 @@ fn invalidator<'a>(
 }

 fn filter(ident: &CacheIdent) -> bool {
-    matches!(
-        ident,
-        CacheIdent::CreateFlow(_) | CacheIdent::DropFlow(_) | CacheIdent::FlowNodeAddressChange(_)
-    )
+    matches!(ident, CacheIdent::CreateFlow(_) | CacheIdent::DropFlow(_))
 }

 #[cfg(test)]
--- a/src/common/meta/src/cache_invalidator.rs
+++ b/src/common/meta/src/cache_invalidator.rs
@@ -22,7 +22,6 @@ use crate::key::flow::flow_name::FlowNameKey;
 use crate::key::flow::flow_route::FlowRouteKey;
 use crate::key::flow::flownode_flow::FlownodeFlowKey;
 use crate::key::flow::table_flow::TableFlowKey;
-use crate::key::node_address::NodeAddressKey;
 use crate::key::schema_name::SchemaNameKey;
 use crate::key::table_info::TableInfoKey;
 use crate::key::table_name::TableNameKey;
@@ -54,10 +53,6 @@ pub struct Context {
 #[async_trait::async_trait]
 pub trait CacheInvalidator: Send + Sync {
    async fn invalidate(&self, ctx: &Context, caches: &[CacheIdent]) -> Result<()>;
-
-    fn name(&self) -> &'static str {
-        std::any::type_name::<Self>()
-    }
 }

 pub type CacheInvalidatorRef = Arc<dyn CacheInvalidator>;
@@ -142,13 +137,6 @@ where
                    let key = FlowInfoKey::new(*flow_id);
                    self.invalidate_key(&key.to_bytes()).await;
                }
-                CacheIdent::FlowNodeAddressChange(node_id) => {
-                    // other caches doesn't need to be invalidated
-                    // since this is only for flownode address change not id change
-                    common_telemetry::info!("Invalidate flow node cache for node_id: {}", node_id);
-                    let key = NodeAddressKey::with_flownode(*node_id);
-                    self.invalidate_key(&key.to_bytes()).await;
-                }
            }
        }
        Ok(())
--- a/src/common/meta/src/datanode.rs
+++ b/src/common/meta/src/datanode.rs
@@ -97,8 +97,6 @@ pub struct RegionStat {
    pub index_size: u64,
    /// The manifest infoof the region.
    pub region_manifest: RegionManifestInfo,
-    /// The total bytes written of the region since region opened.
-    pub written_bytes: u64,
    /// The latest entry id of topic used by data.
    /// **Only used by remote WAL prune.**
    pub data_topic_latest_entry_id: u64,
@@ -279,7 +277,6 @@ impl From<&api::v1::meta::RegionStat> for RegionStat {
            sst_size: region_stat.sst_size,
            index_size: region_stat.index_size,
            region_manifest: region_stat.manifest.into(),
-            written_bytes: region_stat.written_bytes,
            data_topic_latest_entry_id: region_stat.data_topic_latest_entry_id,
            metadata_topic_latest_entry_id: region_stat.metadata_topic_latest_entry_id,
        }
--- a/src/common/meta/src/ddl.rs
+++ b/src/common/meta/src/ddl.rs
@@ -15,17 +15,25 @@
 use std::collections::HashMap;
 use std::sync::Arc;

+use api::v1::meta::ProcedureDetailResponse;
+use common_telemetry::tracing_context::W3cTrace;
 use store_api::storage::{RegionId, RegionNumber, TableId};

 use crate::cache_invalidator::CacheInvalidatorRef;
 use crate::ddl::flow_meta::FlowMetadataAllocatorRef;
 use crate::ddl::table_meta::TableMetadataAllocatorRef;
+use crate::error::{Result, UnsupportedSnafu};
 use crate::key::flow::FlowMetadataManagerRef;
 use crate::key::table_route::PhysicalTableRouteValue;
 use crate::key::TableMetadataManagerRef;
 use crate::node_manager::NodeManagerRef;
 use crate::region_keeper::MemoryRegionKeeperRef;
 use crate::region_registry::LeaderRegionRegistryRef;
+use crate::rpc::ddl::{SubmitDdlTaskRequest, SubmitDdlTaskResponse};
+use crate::rpc::procedure::{
+    AddRegionFollowerRequest, MigrateRegionRequest, MigrateRegionResponse, ProcedureStateResponse,
+    RemoveRegionFollowerRequest,
+};
 use crate::DatanodeId;

 pub mod alter_database;
@@ -36,7 +44,6 @@ pub mod create_flow;
 pub mod create_logical_tables;
 pub mod create_table;
 mod create_table_template;
-pub(crate) use create_table_template::{build_template_from_raw_table_info, CreateRequestBuilder};
 pub mod create_view;
 pub mod drop_database;
 pub mod drop_flow;
@@ -51,6 +58,64 @@ pub(crate) mod tests;
 pub mod truncate_table;
 pub mod utils;

+#[derive(Debug, Default)]
+pub struct ExecutorContext {
+    pub tracing_context: Option<W3cTrace>,
+}
+
+/// The procedure executor that accepts ddl, region migration task etc.
+#[async_trait::async_trait]
+pub trait ProcedureExecutor: Send + Sync {
+    /// Submit a ddl task
+    async fn submit_ddl_task(
+        &self,
+        ctx: &ExecutorContext,
+        request: SubmitDdlTaskRequest,
+    ) -> Result<SubmitDdlTaskResponse>;
+
+    /// Add a region follower
+    async fn add_region_follower(
+        &self,
+        _ctx: &ExecutorContext,
+        _request: AddRegionFollowerRequest,
+    ) -> Result<()> {
+        UnsupportedSnafu {
+            operation: "add_region_follower",
+        }
+        .fail()
+    }
+
+    /// Remove a region follower
+    async fn remove_region_follower(
+        &self,
+        _ctx: &ExecutorContext,
+        _request: RemoveRegionFollowerRequest,
+    ) -> Result<()> {
+        UnsupportedSnafu {
+            operation: "remove_region_follower",
+        }
+        .fail()
+    }
+
+    /// Submit a region migration task
+    async fn migrate_region(
+        &self,
+        ctx: &ExecutorContext,
+        request: MigrateRegionRequest,
+    ) -> Result<MigrateRegionResponse>;
+
+    /// Query the procedure state by its id
+    async fn query_procedure_state(
+        &self,
+        ctx: &ExecutorContext,
+        pid: &str,
+    ) -> Result<ProcedureStateResponse>;
+
+    async fn list_procedures(&self, ctx: &ExecutorContext) -> Result<ProcedureDetailResponse>;
+}
+
+pub type ProcedureExecutorRef = Arc<dyn ProcedureExecutor>;
+
 /// Metadata allocated to a table.
 #[derive(Default)]
 pub struct TableMetadata {
--- a/src/common/meta/src/ddl/alter_logical_tables/update_metadata.rs
+++ b/src/common/meta/src/ddl/alter_logical_tables/update_metadata.rs
@@ -13,12 +13,12 @@
 // limitations under the License.

 use common_grpc_expr::alter_expr_to_request;
+use itertools::Itertools;
 use snafu::ResultExt;
 use table::metadata::{RawTableInfo, TableInfo};

 use crate::ddl::alter_logical_tables::executor::AlterLogicalTablesExecutor;
 use crate::ddl::alter_logical_tables::AlterLogicalTablesProcedure;
-use crate::ddl::utils::table_info::batch_update_table_info_values;
 use crate::error;
 use crate::error::{ConvertAlterTableRequestSnafu, Result};
 use crate::key::table_info::TableInfoValue;
@@ -48,8 +48,25 @@ impl AlterLogicalTablesProcedure {

    pub(crate) async fn update_logical_tables_metadata(&mut self) -> Result<()> {
        let table_info_values = self.build_update_metadata()?;
-        batch_update_table_info_values(&self.context.table_metadata_manager, table_info_values)
-            .await
+        let manager = &self.context.table_metadata_manager;
+        let chunk_size = manager.batch_update_table_info_value_chunk_size();
+        if table_info_values.len() > chunk_size {
+            let chunks = table_info_values
+                .into_iter()
+                .chunks(chunk_size)
+                .into_iter()
+                .map(|check| check.collect::<Vec<_>>())
+                .collect::<Vec<_>>();
+            for chunk in chunks {
+                manager.batch_update_table_info_values(chunk).await?;
+            }
+        } else {
+            manager
+                .batch_update_table_info_values(table_info_values)
+                .await?;
+        }
+
+        Ok(())
    }

    pub(crate) fn build_update_metadata(
--- a/src/common/meta/src/ddl/alter_logical_tables/validator.rs
+++ b/src/common/meta/src/ddl/alter_logical_tables/validator.rs
@@ -21,9 +21,7 @@ use store_api::storage::TableId;
 use table::table_reference::TableReference;

 use crate::ddl::utils::table_id::get_all_table_ids_by_names;
-use crate::ddl::utils::table_info::{
-    all_logical_table_routes_have_same_physical_id, get_all_table_info_values_by_table_ids,
-};
+use crate::ddl::utils::table_info::get_all_table_info_values_by_table_ids;
 use crate::error::{
    AlterLogicalTablesInvalidArgumentsSnafu, Result, TableInfoNotFoundSnafu,
    TableRouteNotFoundSnafu,
@@ -148,16 +146,23 @@ impl<'a> AlterLogicalTableValidator<'a> {
        table_route_manager: &TableRouteManager,
        table_ids: &[TableId],
    ) -> Result<()> {
-        let all_logical_table_routes_have_same_physical_id =
-            all_logical_table_routes_have_same_physical_id(
-                table_route_manager,
-                table_ids,
-                self.physical_table_id,
-            )
+        let table_routes = table_route_manager
+            .table_route_storage()
+            .batch_get(table_ids)
            .await?;

+        let physical_table_id = self.physical_table_id;
+
+        let is_same_physical_table = table_routes.iter().all(|r| {
+            if let Some(TableRouteValue::Logical(r)) = r {
+                r.physical_table_id() == physical_table_id
+            } else {
+                false
+            }
+        });
+
        ensure!(
-            all_logical_table_routes_have_same_physical_id,
+            is_same_physical_table,
            AlterLogicalTablesInvalidArgumentsSnafu {
                err_msg: "All the tasks should have the same physical table id"
            }
--- a/src/common/meta/src/ddl/alter_table.rs
+++ b/src/common/meta/src/ddl/alter_table.rs
@@ -227,6 +227,7 @@ impl AlterTableProcedure {
    }

    fn handle_alter_region_response(&mut self, mut results: Vec<RegionResponse>) -> Result<()> {
+        self.data.state = AlterTableState::UpdateMetadata;
        if let Some(column_metadatas) =
            extract_column_metadatas(&mut results, TABLE_COLUMN_METADATA_EXTENSION_KEY)?
        {
@@ -234,7 +235,7 @@ impl AlterTableProcedure {
        } else {
            warn!("altering table result doesn't contains extension key `{TABLE_COLUMN_METADATA_EXTENSION_KEY}`,leaving the table's column metadata unchanged");
        }
-        self.data.state = AlterTableState::UpdateMetadata;
+
        Ok(())
    }

--- a/src/common/meta/src/ddl/alter_table/executor.rs
+++ b/src/common/meta/src/ddl/alter_table/executor.rs
@@ -299,8 +299,8 @@ fn build_new_table_info(
        | AlterKind::ModifyColumnTypes { .. }
        | AlterKind::SetTableOptions { .. }
        | AlterKind::UnsetTableOptions { .. }
-        | AlterKind::SetIndexes { .. }
-        | AlterKind::UnsetIndexes { .. }
+        | AlterKind::SetIndex { .. }
+        | AlterKind::UnsetIndex { .. }
        | AlterKind::DropDefaults { .. } => {}
    }

--- a/src/common/meta/src/ddl/alter_table/region_request.rs
+++ b/src/common/meta/src/ddl/alter_table/region_request.rs
@@ -108,8 +108,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::SetIndexes(v) => Ok(Some(alter_request::Kind::SetIndexes(v.clone()))),
-        Kind::UnsetIndexes(v) => Ok(Some(alter_request::Kind::UnsetIndexes(v.clone()))),
        Kind::DropDefaults(v) => Ok(Some(alter_request::Kind::DropDefaults(v.clone()))),
    }
 }
--- a/src/common/meta/src/ddl/create_flow.rs
+++ b/src/common/meta/src/ddl/create_flow.rs
@@ -167,6 +167,25 @@ impl CreateFlowProcedure {
        }

        self.collect_source_tables().await?;
+
+        // Validate that source and sink tables are not the same
+        let sink_table_name = &self.data.task.sink_table_name;
+        if self
+            .data
+            .task
+            .source_table_names
+            .iter()
+            .any(|source| source == sink_table_name)
+        {
+            return error::UnsupportedSnafu {
+                operation: format!(
+                    "Creating flow with source and sink table being the same: {}",
+                    sink_table_name
+                ),
+            }
+            .fail();
+        }
+
        if self.data.flow_id.is_none() {
            self.allocate_flow_id().await?;
        }
--- a/src/common/meta/src/ddl/create_table.rs
+++ b/src/common/meta/src/ddl/create_table.rs
@@ -21,7 +21,7 @@ use common_error::ext::BoxedError;
 use common_procedure::error::{
    ExternalSnafu, FromJsonSnafu, Result as ProcedureResult, ToJsonSnafu,
 };
-use common_procedure::{Context as ProcedureContext, LockKey, Procedure, ProcedureId, Status};
+use common_procedure::{Context as ProcedureContext, LockKey, Procedure, Status};
 use common_telemetry::tracing_context::TracingContext;
 use common_telemetry::{info, warn};
 use futures::future::join_all;
@@ -246,6 +246,8 @@ impl CreateTableProcedure {
            }
        }

+        self.creator.data.state = CreateTableState::CreateMetadata;
+
        let mut results = join_all(create_region_tasks)
            .await
            .into_iter()
@@ -259,7 +261,6 @@ impl CreateTableProcedure {
            warn!("creating table result doesn't contains extension key `{TABLE_COLUMN_METADATA_EXTENSION_KEY}`,leaving the table's column metadata unchanged");
        }

-        self.creator.data.state = CreateTableState::CreateMetadata;
        Ok(Status::executing(true))
    }

@@ -267,9 +268,8 @@ impl CreateTableProcedure {
    ///
    /// Abort(not-retry):
    /// - Failed to create table metadata.
-    async fn on_create_metadata(&mut self, pid: ProcedureId) -> Result<Status> {
+    async fn on_create_metadata(&mut self) -> Result<Status> {
        let table_id = self.table_id();
-        let table_ref = self.creator.data.table_ref();
        let manager = &self.context.table_metadata_manager;

        let mut raw_table_info = self.table_info().clone();
@@ -289,10 +289,7 @@ impl CreateTableProcedure {
        self.context
            .register_failure_detectors(detecting_regions)
            .await;
-        info!(
-            "Successfully created table: {}, table_id: {}, procedure_id: {}",
-            table_ref, table_id, pid
-        );
+        info!("Created table metadata for table {table_id}");

        self.creator.opening_regions.clear();
        Ok(Status::done_with_output(table_id))
@@ -320,7 +317,7 @@ impl Procedure for CreateTableProcedure {
        Ok(())
    }

-    async fn execute(&mut self, ctx: &ProcedureContext) -> ProcedureResult<Status> {
+    async fn execute(&mut self, _ctx: &ProcedureContext) -> ProcedureResult<Status> {
        let state = &self.creator.data.state;

        let _timer = metrics::METRIC_META_PROCEDURE_CREATE_TABLE
@@ -330,7 +327,7 @@ impl Procedure for CreateTableProcedure {
        match state {
            CreateTableState::Prepare => self.on_prepare().await,
            CreateTableState::DatanodeCreateRegions => self.on_datanode_create_regions().await,
-            CreateTableState::CreateMetadata => self.on_create_metadata(ctx.procedure_id).await,
+            CreateTableState::CreateMetadata => self.on_create_metadata().await,
        }
        .map_err(map_to_procedure_error)
    }
--- a/src/common/meta/src/ddl/create_table_template.rs
+++ b/src/common/meta/src/ddl/create_table_template.rs
@@ -14,57 +14,17 @@

 use std::collections::HashMap;

-use api::v1::column_def::try_as_column_def;
 use api::v1::region::{CreateRequest, RegionColumnDef};
 use api::v1::{ColumnDef, CreateTableExpr, SemanticType};
-use snafu::{OptionExt, ResultExt};
-use store_api::metric_engine_consts::{LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME};
+use snafu::OptionExt;
+use store_api::metric_engine_consts::LOGICAL_TABLE_METADATA_KEY;
 use store_api::storage::{RegionId, RegionNumber};
-use table::metadata::{RawTableInfo, TableId};
+use table::metadata::TableId;

-use crate::error::{self, Result};
+use crate::error;
+use crate::error::Result;
 use crate::wal_options_allocator::prepare_wal_options;

-/// Builds a [CreateRequest] from a [RawTableInfo].
-///
-/// Note: **This method is only used for creating logical tables.**
-pub(crate) fn build_template_from_raw_table_info(
-    raw_table_info: &RawTableInfo,
-) -> Result<CreateRequest> {
-    let primary_key_indices = &raw_table_info.meta.primary_key_indices;
-    let column_defs = raw_table_info
-        .meta
-        .schema
-        .column_schemas
-        .iter()
-        .enumerate()
-        .map(|(i, c)| {
-            let is_primary_key = primary_key_indices.contains(&i);
-            let column_def = try_as_column_def(c, is_primary_key)
-                .context(error::ConvertColumnDefSnafu { column: &c.name })?;
-
-            Ok(RegionColumnDef {
-                column_def: Some(column_def),
-                // The column id will be overridden by the metric engine.
-                // So we just use the index as the column id.
-                column_id: i as u32,
-            })
-        })
-        .collect::<Result<Vec<_>>>()?;
-
-    let options = HashMap::from(&raw_table_info.meta.options);
-    let template = CreateRequest {
-        region_id: 0,
-        engine: METRIC_ENGINE_NAME.to_string(),
-        column_defs,
-        primary_key: primary_key_indices.iter().map(|i| *i as u32).collect(),
-        path: String::new(),
-        options,
-    };
-
-    Ok(template)
-}
-
 pub(crate) fn build_template(create_table_expr: &CreateTableExpr) -> Result<CreateRequest> {
    let column_defs = create_table_expr
        .column_defs
--- a/src/common/meta/src/ddl/table_meta.rs
+++ b/src/common/meta/src/ddl/table_meta.rs
@@ -175,10 +175,6 @@ impl TableMetadataAllocator {
            region_wal_options,
        })
    }
-
-    pub fn table_id_sequence(&self) -> SequenceRef {
-        self.table_id_sequence.clone()
-    }
 }

 pub type PeerAllocatorRef = Arc<dyn PeerAllocator>;
--- a/src/common/meta/src/ddl/test_util.rs
+++ b/src/common/meta/src/ddl/test_util.rs
@@ -17,7 +17,6 @@ pub mod columns;
 pub mod create_table;
 pub mod datanode_handler;
 pub mod flownode_handler;
-pub mod region_metadata;

 use std::assert_matches::assert_matches;
 use std::collections::HashMap;
--- a/src/common/meta/src/ddl/test_util/datanode_handler.rs
+++ b/src/common/meta/src/ddl/test_util/datanode_handler.rs
@@ -12,11 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use std::collections::HashMap;
 use std::sync::Arc;

 use api::region::RegionResponse;
-use api::v1::region::region_request::Body;
 use api::v1::region::RegionRequest;
 use common_error::ext::{BoxedError, ErrorExt, StackError};
 use common_error::status_code::StatusCode;
@@ -24,8 +22,6 @@ use common_query::request::QueryRequest;
 use common_recordbatch::SendableRecordBatchStream;
 use common_telemetry::debug;
 use snafu::{ResultExt, Snafu};
-use store_api::metadata::RegionMetadata;
-use store_api::storage::RegionId;
 use tokio::sync::mpsc;

 use crate::error::{self, Error, Result};
@@ -282,47 +278,3 @@ impl MockDatanodeHandler for AllFailureDatanodeHandler {
        unreachable!()
    }
 }
-
-#[derive(Clone)]
-pub struct ListMetadataDatanodeHandler {
-    pub region_metadatas: HashMap<RegionId, Option<RegionMetadata>>,
-}
-
-impl ListMetadataDatanodeHandler {
-    pub fn new(region_metadatas: HashMap<RegionId, Option<RegionMetadata>>) -> Self {
-        Self { region_metadatas }
-    }
-}
-
-#[async_trait::async_trait]
-impl MockDatanodeHandler for ListMetadataDatanodeHandler {
-    async fn handle(&self, _peer: &Peer, request: RegionRequest) -> Result<RegionResponse> {
-        let Some(Body::ListMetadata(req)) = request.body else {
-            unreachable!()
-        };
-        let mut response = RegionResponse::new(0);
-
-        let mut output = Vec::with_capacity(req.region_ids.len());
-        for region_id in req.region_ids {
-            match self.region_metadatas.get(&RegionId::from_u64(region_id)) {
-                Some(metadata) => {
-                    output.push(metadata.clone());
-                }
-                None => {
-                    output.push(None);
-                }
-            }
-        }
-
-        response.metadata = serde_json::to_vec(&output).unwrap();
-        Ok(response)
-    }
-
-    async fn handle_query(
-        &self,
-        _peer: &Peer,
-        _request: QueryRequest,
-    ) -> Result<SendableRecordBatchStream> {
-        unreachable!()
-    }
-}
--- a/src/common/meta/src/ddl/test_util/region_metadata.rs
+++ b/src/common/meta/src/ddl/test_util/region_metadata.rs
@@ -1,34 +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 api::v1::SemanticType;
-use store_api::metadata::{ColumnMetadata, RegionMetadata, RegionMetadataBuilder};
-use store_api::storage::RegionId;
-
-/// Builds a region metadata with the given column metadatas.
-pub fn build_region_metadata(
-    region_id: RegionId,
-    column_metadatas: &[ColumnMetadata],
-) -> RegionMetadata {
-    let mut builder = RegionMetadataBuilder::new(region_id);
-    let mut primary_key = vec![];
-    for column_metadata in column_metadatas {
-        builder.push_column_metadata(column_metadata.clone());
-        if column_metadata.semantic_type == SemanticType::Tag {
-            primary_key.push(column_metadata.column_id);
-        }
-    }
-    builder.primary_key(primary_key);
-    builder.build().unwrap()
-}
--- a/src/common/meta/src/ddl/tests/create_flow.rs
+++ b/src/common/meta/src/ddl/tests/create_flow.rs
@@ -141,3 +141,41 @@ async fn test_create_flow() {
    let err = procedure.on_prepare().await.unwrap_err();
    assert_matches!(err, error::Error::FlowAlreadyExists { .. });
 }
+
+#[tokio::test]
+async fn test_create_flow_same_source_and_sink_table() {
+    let table_id = 1024;
+    let table_name = TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "same_table");
+
+    // Use the same table for both source and sink
+    let source_table_names = vec![table_name.clone()];
+    let sink_table_name = table_name.clone();
+
+    let node_manager = Arc::new(MockFlownodeManager::new(NaiveFlownodeHandler));
+    let ddl_context = new_ddl_context(node_manager);
+
+    // Create the table first so it exists
+    let task = test_create_table_task("same_table", table_id);
+    ddl_context
+        .table_metadata_manager
+        .create_table_metadata(
+            task.table_info.clone(),
+            TableRouteValue::physical(vec![]),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+
+    // Try to create a flow with same source and sink table - should fail
+    let task = test_create_flow_task("my_flow", source_table_names, sink_table_name, false);
+    let query_ctx = QueryContext::arc().into();
+    let mut procedure = CreateFlowProcedure::new(task, query_ctx, ddl_context);
+    let err = procedure.on_prepare().await.unwrap_err();
+    assert_matches!(err, error::Error::Unsupported { .. });
+
+    // Verify the error message contains information about the same table
+    if let error::Error::Unsupported { operation, .. } = &err {
+        assert!(operation.contains("source and sink table being the same"));
+        assert!(operation.contains("same_table"));
+    }
+}
--- a/src/common/meta/src/ddl/utils.rs
+++ b/src/common/meta/src/ddl/utils.rs
@@ -13,8 +13,6 @@
 // limitations under the License.

 pub(crate) mod raw_table_info;
-#[allow(dead_code)]
-pub(crate) mod region_metadata_lister;
 pub(crate) mod table_id;
 pub(crate) mod table_info;

@@ -448,7 +446,6 @@ pub fn extract_column_metadatas(
        .collect::<Vec<_>>();

    if schemas.is_empty() {
-        warn!("extract_column_metadatas: no extension key `{key}` found in results");
        return Ok(None);
    }

--- a/src/common/meta/src/ddl/utils/raw_table_info.rs
+++ b/src/common/meta/src/ddl/utils/raw_table_info.rs
@@ -54,10 +54,7 @@ pub(crate) fn build_new_physical_table_info(
                }
            }
            SemanticType::Field => value_indices.push(idx),
-            SemanticType::Timestamp => {
-                value_indices.push(idx);
-                *time_index = Some(idx);
-            }
+            SemanticType::Timestamp => *time_index = Some(idx),
        }

        columns.push(col.column_schema.clone());
--- a/src/common/meta/src/ddl/utils/region_metadata_lister.rs
+++ b/src/common/meta/src/ddl/utils/region_metadata_lister.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.
-
-use std::collections::HashMap;
-
-use api::v1::region::region_request::Body as PbRegionRequest;
-use api::v1::region::{ListMetadataRequest, RegionRequest, RegionRequestHeader};
-use common_telemetry::tracing_context::TracingContext;
-use futures::future::join_all;
-use snafu::ResultExt;
-use store_api::metadata::RegionMetadata;
-use store_api::storage::{RegionId, TableId};
-
-use crate::ddl::utils::add_peer_context_if_needed;
-use crate::error::{DecodeJsonSnafu, Result};
-use crate::node_manager::NodeManagerRef;
-use crate::rpc::router::{find_leaders, region_distribution, RegionRoute};
-
-/// Collects the region metadata from the datanodes.
-pub struct RegionMetadataLister {
-    node_manager: NodeManagerRef,
-}
-
-impl RegionMetadataLister {
-    /// Creates a new [`RegionMetadataLister`] with the given [`NodeManagerRef`].
-    pub fn new(node_manager: NodeManagerRef) -> Self {
-        Self { node_manager }
-    }
-
-    /// Collects the region metadata from the datanodes.
-    pub async fn list(
-        &self,
-        table_id: TableId,
-        region_routes: &[RegionRoute],
-    ) -> Result<Vec<Option<RegionMetadata>>> {
-        let region_distribution = region_distribution(region_routes);
-        let leaders = find_leaders(region_routes)
-            .into_iter()
-            .map(|p| (p.id, p))
-            .collect::<HashMap<_, _>>();
-
-        let total_num_region = region_distribution
-            .values()
-            .map(|r| r.leader_regions.len())
-            .sum::<usize>();
-
-        let mut list_metadata_tasks = Vec::with_capacity(leaders.len());
-
-        // Build requests.
-        for (datanode_id, region_role_set) in region_distribution {
-            if region_role_set.leader_regions.is_empty() {
-                continue;
-            }
-            // Safety: must exists.
-            let peer = leaders.get(&datanode_id).unwrap();
-            let requester = self.node_manager.datanode(peer).await;
-            let region_ids = region_role_set
-                .leader_regions
-                .iter()
-                .map(|r| RegionId::new(table_id, *r).as_u64())
-                .collect();
-            let request = Self::build_list_metadata_request(region_ids);
-
-            let peer = peer.clone();
-            list_metadata_tasks.push(async move {
-                requester
-                    .handle(request)
-                    .await
-                    .map_err(add_peer_context_if_needed(peer))
-            });
-        }
-
-        let results = join_all(list_metadata_tasks)
-            .await
-            .into_iter()
-            .collect::<Result<Vec<_>>>()?
-            .into_iter()
-            .map(|r| r.metadata);
-
-        let mut output = Vec::with_capacity(total_num_region);
-        for result in results {
-            let region_metadatas: Vec<Option<RegionMetadata>> =
-                serde_json::from_slice(&result).context(DecodeJsonSnafu)?;
-            output.extend(region_metadatas);
-        }
-
-        Ok(output)
-    }
-
-    fn build_list_metadata_request(region_ids: Vec<u64>) -> RegionRequest {
-        RegionRequest {
-            header: Some(RegionRequestHeader {
-                tracing_context: TracingContext::from_current_span().to_w3c(),
-                ..Default::default()
-            }),
-            body: Some(PbRegionRequest::ListMetadata(ListMetadataRequest {
-                region_ids,
-            })),
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::collections::HashMap;
-    use std::sync::Arc;
-
-    use api::region::RegionResponse;
-    use api::v1::meta::Peer;
-    use api::v1::region::region_request::Body;
-    use api::v1::region::RegionRequest;
-    use store_api::metadata::RegionMetadata;
-    use store_api::storage::RegionId;
-    use tokio::sync::mpsc;
-
-    use crate::ddl::test_util::datanode_handler::{DatanodeWatcher, ListMetadataDatanodeHandler};
-    use crate::ddl::test_util::region_metadata::build_region_metadata;
-    use crate::ddl::test_util::test_column_metadatas;
-    use crate::ddl::utils::region_metadata_lister::RegionMetadataLister;
-    use crate::error::Result;
-    use crate::rpc::router::{Region, RegionRoute};
-    use crate::test_util::MockDatanodeManager;
-
-    fn assert_list_metadata_request(req: RegionRequest, expected_region_ids: &[RegionId]) {
-        let Some(Body::ListMetadata(req)) = req.body else {
-            unreachable!()
-        };
-
-        assert_eq!(req.region_ids.len(), expected_region_ids.len());
-        for region_id in expected_region_ids {
-            assert!(req.region_ids.contains(&region_id.as_u64()));
-        }
-    }
-
-    fn empty_list_metadata_handler(_peer: Peer, request: RegionRequest) -> Result<RegionResponse> {
-        let Some(Body::ListMetadata(req)) = request.body else {
-            unreachable!()
-        };
-
-        let mut output: Vec<Option<RegionMetadata>> = Vec::with_capacity(req.region_ids.len());
-        for _region_id in req.region_ids {
-            output.push(None);
-        }
-
-        Ok(RegionResponse::from_metadata(
-            serde_json::to_vec(&output).unwrap(),
-        ))
-    }
-
-    #[tokio::test]
-    async fn test_list_request() {
-        let (tx, mut rx) = mpsc::channel(8);
-        let handler = DatanodeWatcher::new(tx).with_handler(empty_list_metadata_handler);
-        let node_manager = Arc::new(MockDatanodeManager::new(handler));
-        let lister = RegionMetadataLister::new(node_manager);
-        let region_routes = vec![
-            RegionRoute {
-                region: Region::new_test(RegionId::new(1024, 1)),
-                leader_peer: Some(Peer::empty(1)),
-                follower_peers: vec![Peer::empty(5)],
-                leader_state: None,
-                leader_down_since: None,
-            },
-            RegionRoute {
-                region: Region::new_test(RegionId::new(1024, 2)),
-                leader_peer: Some(Peer::empty(3)),
-                follower_peers: vec![Peer::empty(4)],
-                leader_state: None,
-                leader_down_since: None,
-            },
-            RegionRoute {
-                region: Region::new_test(RegionId::new(1024, 3)),
-                leader_peer: Some(Peer::empty(3)),
-                follower_peers: vec![Peer::empty(4)],
-                leader_state: None,
-                leader_down_since: None,
-            },
-        ];
-        let region_metadatas = lister.list(1024, &region_routes).await.unwrap();
-        assert_eq!(region_metadatas.len(), 3);
-
-        let mut requests = vec![];
-        for _ in 0..2 {
-            let (peer, request) = rx.try_recv().unwrap();
-            requests.push((peer, request));
-        }
-        rx.try_recv().unwrap_err();
-
-        let (peer, request) = requests.remove(0);
-        assert_eq!(peer.id, 1);
-        assert_list_metadata_request(request, &[RegionId::new(1024, 1)]);
-        let (peer, request) = requests.remove(0);
-        assert_eq!(peer.id, 3);
-        assert_list_metadata_request(request, &[RegionId::new(1024, 2), RegionId::new(1024, 3)]);
-    }
-
-    #[tokio::test]
-    async fn test_list_region_metadata() {
-        let region_metadata =
-            build_region_metadata(RegionId::new(1024, 1), &test_column_metadatas(&["tag_0"]));
-        let region_metadatas = HashMap::from([
-            (RegionId::new(1024, 0), None),
-            (RegionId::new(1024, 1), Some(region_metadata.clone())),
-        ]);
-        let handler = ListMetadataDatanodeHandler::new(region_metadatas);
-        let node_manager = Arc::new(MockDatanodeManager::new(handler));
-        let lister = RegionMetadataLister::new(node_manager);
-        let region_routes = vec![
-            RegionRoute {
-                region: Region::new_test(RegionId::new(1024, 0)),
-                leader_peer: Some(Peer::empty(1)),
-                follower_peers: vec![],
-                leader_state: None,
-                leader_down_since: None,
-            },
-            RegionRoute {
-                region: Region::new_test(RegionId::new(1024, 1)),
-                leader_peer: Some(Peer::empty(3)),
-                follower_peers: vec![],
-                leader_state: None,
-                leader_down_since: None,
-            },
-        ];
-        let region_metadatas = lister.list(1024, &region_routes).await.unwrap();
-        assert_eq!(region_metadatas.len(), 2);
-        assert_eq!(region_metadatas[0], None);
-        assert_eq!(region_metadatas[1], Some(region_metadata));
-    }
-}
--- a/src/common/meta/src/ddl/utils/table_info.rs
+++ b/src/common/meta/src/ddl/utils/table_info.rs
@@ -12,16 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use itertools::Itertools;
 use snafu::OptionExt;
 use store_api::storage::TableId;
-use table::metadata::RawTableInfo;
 use table::table_reference::TableReference;

 use crate::error::{Result, TableInfoNotFoundSnafu};
 use crate::key::table_info::{TableInfoManager, TableInfoValue};
-use crate::key::table_route::{TableRouteManager, TableRouteValue};
-use crate::key::{DeserializedValueWithBytes, TableMetadataManager};
+use crate::key::DeserializedValueWithBytes;

 /// Get all table info values by table ids.
 ///
@@ -45,56 +42,3 @@ pub(crate) async fn get_all_table_info_values_by_table_ids<'a>(

    Ok(table_info_values)
 }
-
-/// Checks if all the logical table routes have the same physical table id.
-pub(crate) async fn all_logical_table_routes_have_same_physical_id(
-    table_route_manager: &TableRouteManager,
-    table_ids: &[TableId],
-    physical_table_id: TableId,
-) -> Result<bool> {
-    let table_routes = table_route_manager
-        .table_route_storage()
-        .batch_get(table_ids)
-        .await?;
-
-    let is_same_physical_table = table_routes.iter().all(|r| {
-        if let Some(TableRouteValue::Logical(r)) = r {
-            r.physical_table_id() == physical_table_id
-        } else {
-            false
-        }
-    });
-
-    Ok(is_same_physical_table)
-}
-
-/// Batch updates the table info values.
-///
-/// The table info values are grouped into chunks, and each chunk is updated in a single transaction.
-///
-/// Returns an error if any table info value fails to update.
-pub(crate) async fn batch_update_table_info_values(
-    table_metadata_manager: &TableMetadataManager,
-    table_info_values: Vec<(DeserializedValueWithBytes<TableInfoValue>, RawTableInfo)>,
-) -> Result<()> {
-    let chunk_size = table_metadata_manager.batch_update_table_info_value_chunk_size();
-    if table_info_values.len() > chunk_size {
-        let chunks = table_info_values
-            .into_iter()
-            .chunks(chunk_size)
-            .into_iter()
-            .map(|check| check.collect::<Vec<_>>())
-            .collect::<Vec<_>>();
-        for chunk in chunks {
-            table_metadata_manager
-                .batch_update_table_info_values(chunk)
-                .await?;
-        }
-    } else {
-        table_metadata_manager
-            .batch_update_table_info_values(table_info_values)
-            .await?;
-    }
-
-    Ok(())
-}
--- a/src/common/meta/src/ddl_manager.rs
+++ b/src/common/meta/src/ddl_manager.rs
@@ -14,6 +14,7 @@

 use std::sync::Arc;

+use api::v1::meta::ProcedureDetailResponse;
 use common_procedure::{
    watcher, BoxedProcedureLoader, Output, ProcedureId, ProcedureManagerRef, ProcedureWithId,
 };
@@ -36,20 +37,24 @@ use crate::ddl::drop_flow::DropFlowProcedure;
 use crate::ddl::drop_table::DropTableProcedure;
 use crate::ddl::drop_view::DropViewProcedure;
 use crate::ddl::truncate_table::TruncateTableProcedure;
-use crate::ddl::{utils, DdlContext};
+use crate::ddl::{utils, DdlContext, ExecutorContext, ProcedureExecutor};
 use crate::error::{
-    EmptyDdlTasksSnafu, ProcedureOutputSnafu, RegisterProcedureLoaderSnafu, Result,
-    SubmitProcedureSnafu, TableInfoNotFoundSnafu, TableNotFoundSnafu, TableRouteNotFoundSnafu,
-    UnexpectedLogicalRouteTableSnafu, WaitProcedureSnafu,
+    EmptyDdlTasksSnafu, ParseProcedureIdSnafu, ProcedureNotFoundSnafu, ProcedureOutputSnafu,
+    QueryProcedureSnafu, RegisterProcedureLoaderSnafu, Result, SubmitProcedureSnafu,
+    TableInfoNotFoundSnafu, TableNotFoundSnafu, TableRouteNotFoundSnafu,
+    UnexpectedLogicalRouteTableSnafu, UnsupportedSnafu, WaitProcedureSnafu,
 };
 use crate::key::table_info::TableInfoValue;
 use crate::key::table_name::TableNameKey;
 use crate::key::{DeserializedValueWithBytes, TableMetadataManagerRef};
-use crate::procedure_executor::ExecutorContext;
 #[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,
@@ -60,6 +65,8 @@ use crate::rpc::ddl::{
    CreateViewTask, DropDatabaseTask, DropFlowTask, DropTableTask, DropViewTask, QueryContext,
    SubmitDdlTaskRequest, SubmitDdlTaskResponse, TruncateTableTask,
 };
+use crate::rpc::procedure;
+use crate::rpc::procedure::{MigrateRegionRequest, MigrateRegionResponse, ProcedureStateResponse};
 use crate::rpc::router::RegionRoute;

 pub type DdlManagerRef = Arc<DdlManager>;
@@ -88,6 +95,14 @@ 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;
 }

@@ -403,70 +418,6 @@ impl DdlManager {

        Ok((procedure_id, output))
    }
-
-    pub async fn submit_ddl_task(
-        &self,
-        ctx: &ExecutorContext,
-        request: SubmitDdlTaskRequest,
-    ) -> Result<SubmitDdlTaskResponse> {
-        let span = ctx
-            .tracing_context
-            .as_ref()
-            .map(TracingContext::from_w3c)
-            .unwrap_or_else(TracingContext::from_current_span)
-            .attach(tracing::info_span!("DdlManager::submit_ddl_task"));
-        async move {
-            debug!("Submitting Ddl task: {:?}", request.task);
-            match request.task {
-                CreateTable(create_table_task) => {
-                    handle_create_table_task(self, create_table_task).await
-                }
-                DropTable(drop_table_task) => handle_drop_table_task(self, drop_table_task).await,
-                AlterTable(alter_table_task) => {
-                    handle_alter_table_task(self, alter_table_task).await
-                }
-                TruncateTable(truncate_table_task) => {
-                    handle_truncate_table_task(self, truncate_table_task).await
-                }
-                CreateLogicalTables(create_table_tasks) => {
-                    handle_create_logical_table_tasks(self, create_table_tasks).await
-                }
-                AlterLogicalTables(alter_table_tasks) => {
-                    handle_alter_logical_table_tasks(self, alter_table_tasks).await
-                }
-                DropLogicalTables(_) => todo!(),
-                CreateDatabase(create_database_task) => {
-                    handle_create_database_task(self, create_database_task).await
-                }
-                DropDatabase(drop_database_task) => {
-                    handle_drop_database_task(self, drop_database_task).await
-                }
-                AlterDatabase(alter_database_task) => {
-                    handle_alter_database_task(self, alter_database_task).await
-                }
-                CreateFlow(create_flow_task) => {
-                    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(
-                        self,
-                        create_trigger_task,
-                        request.query_context.into(),
-                    )
-                    .await
-                }
-            }
-        }
-        .trace(span)
-        .await
-    }
 }

 async fn handle_truncate_table_task(
@@ -709,6 +660,28 @@ 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,
@@ -773,8 +746,6 @@ async fn handle_create_trigger_task(
    query_context: QueryContext,
 ) -> Result<SubmitDdlTaskResponse> {
    let Some(m) = ddl_manager.trigger_ddl_manager.as_ref() else {
-        use crate::error::UnsupportedSnafu;
-
        return UnsupportedSnafu {
            operation: "create trigger",
        }
@@ -851,6 +822,119 @@ async fn handle_create_view_task(
    })
 }

+/// TODO(dennis): let [`DdlManager`] implement [`ProcedureExecutor`] looks weird, find some way to refactor it.
+#[async_trait::async_trait]
+impl ProcedureExecutor for DdlManager {
+    async fn submit_ddl_task(
+        &self,
+        ctx: &ExecutorContext,
+        request: SubmitDdlTaskRequest,
+    ) -> Result<SubmitDdlTaskResponse> {
+        let span = ctx
+            .tracing_context
+            .as_ref()
+            .map(TracingContext::from_w3c)
+            .unwrap_or(TracingContext::from_current_span())
+            .attach(tracing::info_span!("DdlManager::submit_ddl_task"));
+        async move {
+            debug!("Submitting Ddl task: {:?}", request.task);
+            match request.task {
+                CreateTable(create_table_task) => {
+                    handle_create_table_task(self, create_table_task).await
+                }
+                DropTable(drop_table_task) => handle_drop_table_task(self, drop_table_task).await,
+                AlterTable(alter_table_task) => {
+                    handle_alter_table_task(self, alter_table_task).await
+                }
+                TruncateTable(truncate_table_task) => {
+                    handle_truncate_table_task(self, truncate_table_task).await
+                }
+                CreateLogicalTables(create_table_tasks) => {
+                    handle_create_logical_table_tasks(self, create_table_tasks).await
+                }
+                AlterLogicalTables(alter_table_tasks) => {
+                    handle_alter_logical_table_tasks(self, alter_table_tasks).await
+                }
+                DropLogicalTables(_) => todo!(),
+                CreateDatabase(create_database_task) => {
+                    handle_create_database_task(self, create_database_task).await
+                }
+                DropDatabase(drop_database_task) => {
+                    handle_drop_database_task(self, drop_database_task).await
+                }
+                AlterDatabase(alter_database_task) => {
+                    handle_alter_database_task(self, alter_database_task).await
+                }
+                CreateFlow(create_flow_task) => {
+                    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(
+                        self,
+                        create_trigger_task,
+                        request.query_context.into(),
+                    )
+                    .await
+                }
+                #[cfg(feature = "enterprise")]
+                DropTrigger(drop_trigger_task) => {
+                    handle_drop_trigger_task(self, drop_trigger_task, request.query_context.into())
+                        .await
+                }
+            }
+        }
+        .trace(span)
+        .await
+    }
+
+    async fn migrate_region(
+        &self,
+        _ctx: &ExecutorContext,
+        _request: MigrateRegionRequest,
+    ) -> Result<MigrateRegionResponse> {
+        UnsupportedSnafu {
+            operation: "migrate_region",
+        }
+        .fail()
+    }
+
+    async fn query_procedure_state(
+        &self,
+        _ctx: &ExecutorContext,
+        pid: &str,
+    ) -> Result<ProcedureStateResponse> {
+        let pid =
+            ProcedureId::parse_str(pid).with_context(|_| ParseProcedureIdSnafu { key: pid })?;
+
+        let state = self
+            .procedure_manager
+            .procedure_state(pid)
+            .await
+            .context(QueryProcedureSnafu)?
+            .context(ProcedureNotFoundSnafu {
+                pid: pid.to_string(),
+            })?;
+
+        Ok(procedure::procedure_state_to_pb_response(&state))
+    }
+
+    async fn list_procedures(&self, _ctx: &ExecutorContext) -> Result<ProcedureDetailResponse> {
+        let metas = self
+            .procedure_manager
+            .list_procedures()
+            .await
+            .context(QueryProcedureSnafu)?;
+        Ok(procedure::procedure_details_to_pb_response(metas))
+    }
+}
+
 #[cfg(test)]
 mod tests {
    use std::sync::Arc;
--- a/src/common/meta/src/error.rs
+++ b/src/common/meta/src/error.rs
@@ -18,7 +18,6 @@ use std::sync::Arc;
 use common_error::ext::{BoxedError, ErrorExt};
 use common_error::status_code::StatusCode;
 use common_macro::stack_trace_debug;
-use common_procedure::ProcedureId;
 use common_wal::options::WalOptions;
 use serde_json::error::Error as JsonError;
 use snafu::{Location, Snafu};
@@ -141,21 +140,6 @@ pub enum Error {
        location: Location,
    },

-    #[snafu(display("Failed to get procedure state receiver, procedure id: {procedure_id}"))]
-    ProcedureStateReceiver {
-        procedure_id: ProcedureId,
-        #[snafu(implicit)]
-        location: Location,
-        source: common_procedure::Error,
-    },
-
-    #[snafu(display("Procedure state receiver not found: {procedure_id}"))]
-    ProcedureStateReceiverNotFound {
-        procedure_id: ProcedureId,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
    #[snafu(display("Failed to wait procedure done"))]
    WaitProcedure {
        #[snafu(implicit)]
@@ -403,13 +387,6 @@ pub enum Error {
        location: Location,
    },

-    #[snafu(display("Catalog not found, catalog: {}", catalog))]
-    CatalogNotFound {
-        catalog: String,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
    #[snafu(display("Invalid metadata, err: {}", err_msg))]
    InvalidMetadata {
        err_msg: String,
@@ -900,93 +877,6 @@ pub enum Error {
        #[snafu(source)]
        error: object_store::Error,
    },
-
-    #[snafu(display("Missing column ids"))]
-    MissingColumnIds {
-        #[snafu(implicit)]
-        location: Location,
-    },
-
-    #[snafu(display(
-        "Missing column in column metadata: {}, table: {}, table_id: {}",
-        column_name,
-        table_name,
-        table_id,
-    ))]
-    MissingColumnInColumnMetadata {
-        column_name: String,
-        #[snafu(implicit)]
-        location: Location,
-        table_name: String,
-        table_id: TableId,
-    },
-
-    #[snafu(display(
-        "Mismatch column id: column_name: {}, column_id: {}, table: {}, table_id: {}",
-        column_name,
-        column_id,
-        table_name,
-        table_id,
-    ))]
-    MismatchColumnId {
-        column_name: String,
-        column_id: u32,
-        #[snafu(implicit)]
-        location: Location,
-        table_name: String,
-        table_id: TableId,
-    },
-
-    #[snafu(display("Failed to convert column def, column: {}", column))]
-    ConvertColumnDef {
-        column: String,
-        #[snafu(implicit)]
-        location: Location,
-        source: api::error::Error,
-    },
-
-    #[snafu(display(
-        "Column metadata inconsistencies found in table: {}, table_id: {}",
-        table_name,
-        table_id
-    ))]
-    ColumnMetadataConflicts {
-        table_name: String,
-        table_id: TableId,
-    },
-
-    #[snafu(display(
-        "Column not found in column metadata, column_name: {}, column_id: {}",
-        column_name,
-        column_id
-    ))]
-    ColumnNotFound { column_name: String, column_id: u32 },
-
-    #[snafu(display(
-        "Column id mismatch, column_name: {}, expected column_id: {}, actual column_id: {}",
-        column_name,
-        expected_column_id,
-        actual_column_id
-    ))]
-    ColumnIdMismatch {
-        column_name: String,
-        expected_column_id: u32,
-        actual_column_id: u32,
-    },
-
-    #[snafu(display(
-        "Timestamp column mismatch, expected column_name: {}, expected column_id: {}, actual column_name: {}, actual column_id: {}",
-        expected_column_name,
-        expected_column_id,
-        actual_column_name,
-        actual_column_id,
-    ))]
-    TimestampMismatch {
-        expected_column_name: String,
-        expected_column_id: u32,
-        actual_column_name: String,
-        actual_column_id: u32,
-    },
 }

 pub type Result<T> = std::result::Result<T, Error>;
@@ -1006,16 +896,7 @@ impl ErrorExt for Error {
            | DeserializeFromJson { .. } => StatusCode::Internal,

            NoLeader { .. } => StatusCode::TableUnavailable,
-            ValueNotExist { .. }
-            | ProcedurePoisonConflict { .. }
-            | ProcedureStateReceiverNotFound { .. }
-            | MissingColumnIds { .. }
-            | MissingColumnInColumnMetadata { .. }
-            | MismatchColumnId { .. }
-            | ColumnMetadataConflicts { .. }
-            | ColumnNotFound { .. }
-            | ColumnIdMismatch { .. }
-            | TimestampMismatch { .. } => StatusCode::Unexpected,
+            ValueNotExist { .. } | ProcedurePoisonConflict { .. } => StatusCode::Unexpected,

            Unsupported { .. } => StatusCode::Unsupported,
            WriteObject { .. } | ReadObject { .. } => StatusCode::StorageUnavailable,
@@ -1099,13 +980,10 @@ impl ErrorExt for Error {
            AbortProcedure { source, .. } => source.status_code(),
            ConvertAlterTableRequest { source, .. } => source.status_code(),
            PutPoison { source, .. } => source.status_code(),
-            ConvertColumnDef { source, .. } => source.status_code(),
-            ProcedureStateReceiver { source, .. } => source.status_code(),

            ParseProcedureId { .. }
            | InvalidNumTopics { .. }
            | SchemaNotFound { .. }
-            | CatalogNotFound { .. }
            | InvalidNodeInfoKey { .. }
            | InvalidStatKey { .. }
            | ParseNum { .. }
--- a/src/common/meta/src/instruction.rs
+++ b/src/common/meta/src/instruction.rs
@@ -174,8 +174,6 @@ pub struct UpgradeRegion {
 /// The identifier of cache.
 pub enum CacheIdent {
    FlowId(FlowId),
-    /// Indicate change of address of flownode.
-    FlowNodeAddressChange(u64),
    FlowName(FlowName),
    TableId(TableId),
    TableName(TableName),
--- a/src/common/meta/src/key.rs
+++ b/src/common/meta/src/key.rs
@@ -167,7 +167,6 @@ pub const NAME_PATTERN: &str = r"[a-zA-Z_:-][a-zA-Z0-9_:\-\.@#]*";
 pub const LEGACY_MAINTENANCE_KEY: &str = "__maintenance";
 pub const MAINTENANCE_KEY: &str = "__switches/maintenance";
 pub const PAUSE_PROCEDURE_KEY: &str = "__switches/pause_procedure";
-pub const RECOVERY_MODE_KEY: &str = "__switches/recovery";

 pub const DATANODE_TABLE_KEY_PREFIX: &str = "__dn_table";
 pub const TABLE_INFO_KEY_PREFIX: &str = "__table_info";
@@ -182,11 +181,6 @@ pub const KAFKA_TOPIC_KEY_PREFIX: &str = "__topic_name/kafka";
 pub const LEGACY_TOPIC_KEY_PREFIX: &str = "__created_wal_topics/kafka";
 pub const TOPIC_REGION_PREFIX: &str = "__topic_region";

-/// The election key.
-pub const ELECTION_KEY: &str = "__metasrv_election";
-/// The root key of metasrv election candidates.
-pub const CANDIDATES_ROOT: &str = "__metasrv_election_candidates/";
-
 /// The keys with these prefixes will be loaded into the cache when the leader starts.
 pub const CACHE_KEY_PREFIXES: [&str; 5] = [
    TABLE_NAME_KEY_PREFIX,
--- a/src/common/meta/src/key/runtime_switch.rs
+++ b/src/common/meta/src/key/runtime_switch.rs
@@ -21,7 +21,7 @@ use moka::future::Cache;
 use snafu::ResultExt;

 use crate::error::{GetCacheSnafu, Result};
-use crate::key::{LEGACY_MAINTENANCE_KEY, MAINTENANCE_KEY, PAUSE_PROCEDURE_KEY, RECOVERY_MODE_KEY};
+use crate::key::{LEGACY_MAINTENANCE_KEY, MAINTENANCE_KEY, PAUSE_PROCEDURE_KEY};
 use crate::kv_backend::KvBackendRef;
 use crate::rpc::store::{BatchDeleteRequest, PutRequest};

@@ -131,21 +131,6 @@ impl RuntimeSwitchManager {
    pub async fn is_procedure_paused(&self) -> Result<bool> {
        self.exists(PAUSE_PROCEDURE_KEY).await
    }
-
-    /// Enables recovery mode.
-    pub async fn set_recovery_mode(&self) -> Result<()> {
-        self.put_key(RECOVERY_MODE_KEY).await
-    }
-
-    /// Unsets recovery mode.
-    pub async fn unset_recovery_mode(&self) -> Result<()> {
-        self.delete_keys(&[RECOVERY_MODE_KEY]).await
-    }
-
-    /// Returns true if the system is currently in recovery mode.
-    pub async fn recovery_mode(&self) -> Result<bool> {
-        self.exists(RECOVERY_MODE_KEY).await
-    }
 }

 #[cfg(test)]
@@ -236,15 +221,4 @@ mod tests {
        runtime_switch_manager.resume_procedure().await.unwrap();
        assert!(!runtime_switch_manager.is_procedure_paused().await.unwrap());
    }
-
-    #[tokio::test]
-    async fn test_recovery_mode() {
-        let runtime_switch_manager =
-            Arc::new(RuntimeSwitchManager::new(Arc::new(MemoryKvBackend::new())));
-        assert!(!runtime_switch_manager.recovery_mode().await.unwrap());
-        runtime_switch_manager.set_recovery_mode().await.unwrap();
-        assert!(runtime_switch_manager.recovery_mode().await.unwrap());
-        runtime_switch_manager.unset_recovery_mode().await.unwrap();
-        assert!(!runtime_switch_manager.recovery_mode().await.unwrap());
-    }
 }
--- a/src/common/meta/src/key/table_route.rs
+++ b/src/common/meta/src/key/table_route.rs
@@ -48,6 +48,11 @@ 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)]
@@ -184,17 +189,6 @@ impl TableRouteValue {
        }
    }

-    /// Converts to [`LogicalTableRouteValue`].
-    ///
-    /// # Panic
-    /// If it is not the [`LogicalTableRouteValue`].
-    pub fn into_logical_table_route(self) -> LogicalTableRouteValue {
-        match self {
-            TableRouteValue::Logical(x) => x,
-            _ => unreachable!("Mistakenly been treated as a Logical TableRoute: {self:?}"),
-        }
-    }
-
    pub fn region_numbers(&self) -> Vec<RegionNumber> {
        match self {
            TableRouteValue::Physical(x) => x
--- a/src/common/meta/src/lib.rs
+++ b/src/common/meta/src/lib.rs
@@ -37,9 +37,7 @@ pub mod node_expiry_listener;
 pub mod node_manager;
 pub mod peer;
 pub mod poison_key;
-pub mod procedure_executor;
 pub mod range_stream;
-pub mod reconciliation;
 pub mod region_keeper;
 pub mod region_registry;
 pub mod rpc;
--- a/Show More
+++ b/Show More